Function Calling 是大语言模型（LLM）与外部系统交互的核心机制，允许模型通过结构化请求调用预定义功能。其核心价值在于：
能力扩展：突破模型固有知识限制（如实时数据获取）
精准控制：约束输出格式（JSON/XML）确保下游系统兼容
动态路由：根据上下文选择最佳工具链
典型应用场景：
实时天气查询：模型生成 {"location":"北京"}，触发天气API调用
数据库操作：将自然语言转换为SQL查询
多模型协作：GPT-4生成代码 → 文心ERNIE执行行业知识校验
import openai
import requests
# 1. 定义功能描述
functions = [
{
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "城市名称"}
},
"required": ["location"]
}
}
]
# 2. 模型解析请求
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": "北京今天气温如何？"}],
functions=functions
)
args = json.loads(response.choices[0].message.function_call.arguments)
# 3. 调用外部功能
def get_weather(location):
api_url = f"https://api.weather.com/v3/?location={location}"
return requests.get(api_url).json()
weather_data = get_weather(args["location"])
# 4. 结果整合输出
final_response = openai.ChatCompletion.create(
model="gpt-4",
messages=[
{"role": "user", "content": "北京今天气温如何？"},
{"role": "function", "name": "get_weather", "content": str(weather_data)}
]
)
print(final_response.choices[0].message.content)
代码示例：FastAPI实现REST服务
from fastapi import FastAPI
app = FastAPI()
@app.post("/weather")
def weather_api(location: str):
# 模拟天气数据
return {"temp": 25, "humidity": 60}
# 启动服务
# uvicorn main:app --port 8000代码示例：gRPC服务定义
syntax = "proto3";
service WeatherService {
rpc GetWeather (Location) returns (WeatherData) {}
}
message Location {
string name = 1;
}
message WeatherData {
int32 temp = 1;
int32 humidity = 2;
}异步IO：并行处理多个调用请求
import asyncio
async def async_get_weather(location):
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.weather.com/{location}") as resp:
return await resp.json()
# 批量处理
locations = ["北京", "上海", "广州"]
tasks = [async_get_weather(loc) for loc in locations]
results = await asyncio.gather(*tasks)缓存机制：减少重复调用
from functools import lru_cache @lru_cache(maxsize=100) def get_weather_cached(location: str): return get_weather(location)
重试策略：指数退避重试
import tenacity @tenacity.retry( stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(multiplier=1, max=10) ) def call_external_api(url): response = requests.get(url) response.raise_for_status() return response.json()
熔断机制：防止级联故障
from circuitbreaker import circuit @circuit(failure_threshold=5, recovery_timeout=60) def critical_api_call(): # 关键业务调用
附：工具与资源推荐
注：本文代码需安装以下依赖：
pip install openai fastapi grpcio tenacity circuitbreaker aiohttp
更多AI大模型应用开发学习内容，尽在聚客AI学院。
