🧠 图解记忆： 先按依赖决定并行，再用校验、超时、幂等重试和部分失败隔离保证可靠。
💡 答案要点
Function Calling = LLM通过结构化JSON调用外部函数，是Agent工具使用的核心机制
基础Function Calling
展开 Python 代码示例（92 行）
python
from openai import OpenAI
import json
client = OpenAI()
# 1. 定义工具
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'"
},
"date": {
"type": "string",
"description": "日期，格式YYYY-MM-DD，默认今天"
}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_news",
"description": "搜索最新新闻",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"limit": {"type": "integer", "description": "返回条数，默认5"}
},
"required": ["query"]
}
}
}
]
# 2. 实际工具函数
def get_weather(city: str, date: str = None) -> dict:
# 调用天气API
return {"city": city, "temp": "22°C", "weather": "晴天"}
def search_news(query: str, limit: int = 5) -> list:
# 调用新闻API
return [{"title": f"关于{query}的新闻{i}", "url": f"..."} for i in range(limit)]
TOOL_MAP = {"get_weather": get_weather, "search_news": search_news}
# 3. 完整对话循环
def run_with_function_calling(user_message: str):
messages = [{"role": "user", "content": user_message}]
while True:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = response.choices[0].message
messages.append(msg)
# 没有工具调用 → 最终回答
if not msg.tool_calls:
return msg.content
# 执行所有工具调用
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# 调用实际函数
result = TOOL_MAP[func_name](**func_args)
# 把结果加入消息
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
result = run_with_function_calling("北京明天天气怎样？同时帮我搜一下最新AI新闻")
print(result)并行工具调用（Parallel Tool Calls）
展开 Python 代码示例（36 行）
python
import concurrent.futures
import time
def execute_tools_parallel(tool_calls: list) -> list:
"""并行执行多个工具调用，大幅缩短响应时间"""
results = [None] * len(tool_calls)
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
futures = {}
for i, tool_call in enumerate(tool_calls):
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
future = executor.submit(TOOL_MAP[func_name], **func_args)
futures[future] = (i, tool_call.id)
for future in concurrent.futures.as_completed(futures):
idx, call_id = futures[future]
try:
results[idx] = {
"tool_call_id": call_id,
"result": future.result(timeout=10)
}
except Exception as e:
results[idx] = {
"tool_call_id": call_id,
"result": {"error": str(e)}
}
return results
# 性能对比：
# 串行：天气(1s) + 新闻(1s) + 股价(1s) = 3s
# 并行：max(天气1s, 新闻1s, 股价1s) = 1s ← 快3倍错误重试机制
展开 Python 代码示例（64 行）
python
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
class RobustToolExecutor:
"""带重试、熔断、超时的工具执行器"""
def __init__(self):
self.failure_counts = {} # 记录失败次数
self.circuit_open = {} # 熔断状态
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type((TimeoutError, ConnectionError))
)
def execute_with_retry(self, func_name: str, func_args: dict):
"""带指数退避重试"""
return TOOL_MAP[func_name](**func_args)
def execute_safe(self, func_name: str, func_args: dict, timeout=5):
"""带熔断器的执行"""
# 检查熔断器
if self.circuit_open.get(func_name, False):
return {"error": f"工具{func_name}当前不可用（熔断中）"}
try:
# 带超时执行
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(self.execute_with_retry, func_name, func_args)
result = future.result(timeout=timeout)
# 成功，重置失败计数
self.failure_counts[func_name] = 0
return result
except concurrent.futures.TimeoutError:
self._record_failure(func_name)
return {"error": f"工具{func_name}超时（>{timeout}s）"}
except Exception as e:
self._record_failure(func_name)
return {"error": str(e)}
def _record_failure(self, func_name: str):
"""记录失败，超阈值触发熔断"""
self.failure_counts[func_name] = self.failure_counts.get(func_name, 0) + 1
if self.failure_counts[func_name] >= 3:
self.circuit_open[func_name] = True
print(f"🔴 熔断触发：{func_name} 已失败3次，30秒内不再调用")
# 30秒后自动恢复
import threading
def reset():
time.sleep(30)
self.circuit_open[func_name] = False
self.failure_counts[func_name] = 0
print(f"🟢 熔断恢复：{func_name}")
threading.Thread(target=reset, daemon=True).start()
# 使用
executor = RobustToolExecutor()
result = executor.execute_safe("get_weather", {"city": "北京"}, timeout=5)面试话术：
"Function Calling是Agent工具使用的核心。基础实现是对话循环：LLM输出tool_calls → 执行函数 → 结果加入消息 → 继续对话。两个关键优化：1）并行执行：多个工具用ThreadPoolExecutor并发执行，从串行3s降到1s；2）三层容错：retry指数退避重试、timeout超时保护、circuit breaker熔断防止雪崩。生产上工具失败率从8%降到0.5%。"