Function Calling 与工具系统设计：让 Agent 一次调用多个工具，并学会自我纠错

Function Calling 与工具系统设计让 Agent 一次调用多个工具并学会自我纠错系列第 2 篇。第 1 篇我们用正则从文本里抠Action:能跑但很脆。这一篇换成结构化工具调用Function Calling / Tool Calling解决三个真实痛点一次要调多个工具、参数校验要提前、错误要能自愈。 全文代码可直接运行末尾附完整测试用例。一、正则解析的三个死穴第 1 篇的OutputParser用正则从模型输出里提取工具调用。它在 Demo 里表现不错但上真机会遇到三件麻烦事死穴 1一次只能调一个工具。用户问「北京和上海今天气温差多少」。理想执行是并行调两次天气接口再算差值。ReAct 文本协议只能一轮一个动作于是变成查北京 → 等 → 查上海 → 等 → 算差值。三次串行 LLM 往返延迟三倍token 也多烧一轮。死穴 2格式漂移无法根治。你用stop[Observation:]和精心设计的 prompt把格式错误率压到 1%。但 1% 在每天十万次调用里就是一千次失败。而且这个错误率会随模型版本更新、上下文变长、任务变复杂而上升你完全不可控。死穴 3参数错误只能在运行时发现。模型传了{city: 北京市, date: 昨天}你的函数签名是date: strISO 格式。正则解析阶段看不出任何问题只有执行到业务代码才炸而这时你已经付了一轮 token。Function Calling 把这三件事一起解决了结构化输出模型返回的不是文本而是tool_calls数组字段名和类型由 JSON Schema 约束格式错误率趋近于 0。原生并行一次响应可以包含多个tool_calls天然支持并行。Schema 即校验参数结构在请求时就声明清楚非法参数在解析阶段就能拦住。二、JSON Schema给工具写一份“说明书”Function Calling 的核心是向模型声明工具。声明写得越精确模型调得越准——这不是玄学Schema 里的description会被拼进 prompt直接影响模型的参数选择。先看一个反面例子{ name: query, description: 查询数据, parameters: {type: object, properties: {a: {type: string}}} }模型看到这个只能靠猜。它不知道a是什么不知道该传什么格式于是各种幻觉参数就来了。再看正面例子{ name: query_orders, description: 按条件查询订单列表。时间范围必填最多返回 100 条。仅查询不做任何修改。, parameters: { type: object, properties: { start_date: { type: string, format: date, description: 开始日期ISO 格式 YYYY-MM-DD例如 2026-09-01 }, end_date: { type: string, format: date, description: 结束日期ISO 格式 YYYY-MM-DD含当天 }, status: { type: string, enum: [paid, unpaid, refunded], description: 订单状态不传表示全部 }, limit: { type: integer, minimum: 1, maximum: 100, default: 20, description: 返回条数上限 } }, required: [start_date, end_date], additionalProperties: false } }差别在哪description 写清楚语义和边界「仅查询不做任何修改」是在防止模型拿它当写接口用。给示例值例如 2026-09-01。模型对格式示例极其敏感一个例子胜过十句描述。用enum收敛取值把开放问题变成选择题。用required明确必填用additionalProperties: false禁止编造参数。给默认值和上下界减少模型的自由发挥空间。一个经验工具数量控制在 15 个以内。超过之后模型的选型准确率会明显下降业内常引用的数据是 20 个工具时准确率掉到 60% 左右。工具多了就分层先给一组「路由工具」由它决定加载哪个工具子集。三、代码 1用 Pydantic 自动生成 Schema手写 JSON Schema 既累又容易出错。正确做法是用 Python 类型注解当唯一事实源Schema 自动生成。# tools/schema.py from __future__ import annotations import inspect from typing import Any, Callable, get_type_hints from pydantic import BaseModel, Field, create_model def func_to_schema( name: str, description: str, func: Callable[..., Any], arg_model: type[BaseModel] | None None, ) - dict: 把 Python 函数转成 OpenAI tools 的 JSON Schema。 参数元数据来自两处优先级从高到低 1. 显式传入的 pydantic model 2. 函数签名里的 Annotated[str, Field(description...)] if arg_model is not None: schema arg_model.model_json_schema() else: hints get_type_hints(func, include_extrasTrue) sig inspect.signature(func) fields: dict[str, Any] {} for pname, p in sig.parameters.items(): if pname in (self, cls): continue ann hints.get(pname, str) default ... if p.default is inspect.Parameter.empty else p.default fields[pname] (ann, default) model create_model(f{name}_args, **fields) schema model.model_json_schema() # OpenAI 要求 parameters 是 object且不能有 $defs 之外的顶层 title schema.pop(title, None) return { type: function, function: { name: name, description: description, parameters: { type: object, properties: schema.get(properties, {}), required: schema.get(required, []), additionalProperties: False, }, }, }用法注意AnnotatedField的写法这是把说明写进类型的标准姿势# tools/biz_tools.py from __future__ import annotations from typing import Annotated from pydantic import BaseModel, Field class WeatherArgs(BaseModel): city: Annotated[str, Field(description城市中文名例如 北京)] date: Annotated[str, Field(description日期ISO 格式 YYYY-MM-DDtoday 表示今天)] def get_weather(city: str, date: str) - dict: 查询指定城市指定日期的气温返回摄氏度。 return {city: city, date: date, temp_c: 26.5, cond: 晴} class SearchArgs(BaseModel): query: Annotated[str, Field(description搜索关键词3-50 字)] top_k: Annotated[int, Field(4, ge1, le10, description返回条数)] 4 def web_search(query: str, top_k: int 4) - list[dict]: 联网搜索返回标题、链接和摘要列表。 return [{title: f{query} 结果{i}, url: fhttps://x/{i}, snippet: ...} for i in range(top_k)] class CalcArgs(BaseModel): expression: Annotated[str, Field(description数学表达式如 (26.5-18.2)*2)] def calc(expression: str) - float: 计算数学表达式返回数值结果。 return eval(expression, {__builtins__: {}}, {}) # 生产环境请换成第 1 篇的 AST 版本四、代码 2支持并行调用的 ToolRuntime现在写运行时。与第 1 篇的ToolRegistry相比这里多了三样东西并行执行、超时控制、结构化错误返回。# tools/runtime.py from __future__ import annotations import asyncio import concurrent.futures import json import time from dataclasses import dataclass, field from typing import Any, Callable from pydantic import BaseModel, ValidationError from .schema import func_to_schema class ToolResult: 统一工具结果。永远不抛异常永远可序列化。 __slots__ (ok, data, error, elapsed_ms) def __init__(self, ok: bool, data: Any None, error: str | None None, elapsed_ms: float 0.0): self.ok ok self.data data self.error error self.elapsed_ms elapsed_ms def to_message(self, tool_call_id: str) - dict: 转成 OpenAI 要求的 tool message。 if self.ok: content self.data if isinstance(self.data, str) else json.dumps( self.data, ensure_asciiFalse, defaultstr) else: # 错误也走 tool 消息但带 [ERROR] 前缀模型能看到并自愈 content f[ERROR] {self.error} return {role: tool, tool_call_id: tool_call_id, content: content[:8000]} def __repr__(self) - str: return fToolResult ok{self.ok} data{str(self.data)[:60]} err{self.error} dataclass class RegisteredTool: name: str description: str func: Callable[..., Any] arg_model: type[BaseModel] | None None timeout: float 10.0 # 是否可以与其他工具并行写操作通常设 False parallel_safe: bool True def schema(self) - dict: return func_to_schema(self.name, self.description, self.func, self.arg_model) class ToolRuntime: def __init__(self) - None: self._tools: dict[str, RegisteredTool] {} self._executor concurrent.futures.ThreadPoolExecutor(max_workers16) def register(self, tool: RegisteredTool) - None: if tool.name in self._tools: raise ValueError(f工具重名: {tool.name}) self._tools[tool.name] tool def tool(self, name: str, description: str, *, timeout: float 10.0, parallel_safe: bool True, arg_model: type[BaseModel] | None None): def deco(fn): self.register(RegisteredTool(name, description, fn, arg_model, timeout, parallel_safe)) return fn return deco def schemas(self) - list[dict]: return [t.schema() for t in self._tools.values()] def has(self, name: str) - bool: return name in self._tools def _invoke_one(self, name: str, raw_args: dict | str) - ToolResult: 单个工具调用校验 - 执行 - 统一返回。 t0 time.perf_counter() tool self._tools.get(name) if tool is None: return ToolResult(False, errorf未知工具 {name}。可用工具: {sorted(self._tools)}) # 1) 参数解析模型有时会把 JSON 字符串塞进来而不是对象 if isinstance(raw_args, str): try: raw_args json.loads(raw_args) except json.JSONDecodeError as e: return ToolResult(False, errorf参数不是合法 JSON: {e}) if raw_args is None: raw_args {} # 2) 参数校验提前拦截不浪费执行 if tool.arg_model is not None: try: args_obj tool.arg_model(**raw_args) call_args: dict dict(args_obj.model_dump()) except ValidationError as e: # 把 pydantic 错误翻译成人话模型看得懂才能改 details ; .join( f{..join(str(x) for x in err[loc])}: {err[msg]} for err in e.errors()[:5] ) return ToolResult(False, errorf参数校验失败: {details}) else: call_args raw_args # 3) 执行 超时 try: fut self._executor.submit(tool.func, **call_args) data fut.result(timeouttool.timeout) return ToolResult(True, datadata, elapsed_ms(time.perf_counter() - t0) * 1000) except concurrent.futures.TimeoutError: return ToolResult(False, errorf工具执行超时{tool.timeout}s请简化参数或稍后重试) except TypeError as e: return ToolResult(False, errorf参数不匹配: {e}) except Exception as e: # noqa: BLE001 return ToolResult(False, errorf{type(e).__name__}: {e}) def invoke_many(self, calls: list[dict]) - list[tuple[str, ToolResult]]: 并行执行一批 tool_calls保持输入顺序。 calls: [{id: ..., name: ..., arguments: {...}}] results: list[tuple[str, ToolResult]] [None] * len(calls) # type: ignore parallel_idx, serial_idx [], [] for i, c in enumerate(calls): t self._tools.get(c[name]) (parallel_idx if (t is None or t.parallel_safe) else serial_idx).append(i) with concurrent.futures.ThreadPoolExecutor(max_workers8) as ex: futs { ex.submit(self._invoke_one, calls[i][name], calls[i].get(arguments) or {}): i for i in parallel_idx } for fut in concurrent.futures.as_completed(futs): results[futs[fut]] (calls[futs[fut]][id], fut.result()) # 非并行安全写操作串行执行避免竞态 for i in serial_idx: results[i] (calls[i][id], self._invoke_one(calls[i][name], calls[i].get(arguments) or {})) return results几个关键决策parallel_safe标志读操作并行写操作串行。两个「写入文件」并行会互相覆盖这类标志必须在工具声明时就想清楚。超时放在线程池上concurrent.futures的timeout只能取消未开始的任务跑起来的任务杀不掉。真要硬杀得上子进程第 8 篇讲沙箱时展开。pydantic 错误翻译成人话ValidationError默认是 dump 一堆 JSON模型看了只会更糊涂。抽成字段: 原因的一行文字模型基本都能一轮改正。五、代码 3支持并行调用的 Agent 循环# agent/fc_agent.py from __future__ import annotations import json import uuid from dataclasses import dataclass, field from tools.runtime import ToolResult, ToolRuntime dataclass class Turn: 一轮 一次 LLM 调用 零到多个工具执行。 index: int assistant_msg: dict tool_results: list[tuple[str, ToolResult]] field(default_factorylist) prompt_tokens: int 0 completion_tokens: int 0 dataclass class RunResult: answer: str turns: list[Turn] total_tokens: int stop_reason: str # done | max_turns | forced tool_calls_count: int class ToolCallingAgent: def __init__(self, llm, tools: ToolRuntime, max_turns: int 10, max_parallel: int 8, verbose: bool True): self.llm llm self.tools tools self.max_turns max_turns self.max_parallel max_parallel self.verbose verbose def run(self, user_input: str, system: str 你是一个有用的助手必要时调用工具完成任务。) - RunResult: messages: list[dict] [ {role: system, content: system}, {role: user, content: user_input}, ] turns: list[Turn] [] total_tokens 0 n_calls 0 stop_reason max_turns for i in range(1, self.max_turns 1): resp self.llm.chat(messages, toolsself.tools.schemas()) total_tokens resp.prompt_tokens resp.completion_tokens assistant_msg resp.assistant_message # 含 content 和 tool_calls messages.append(assistant_msg) turn Turn(i, assistant_msg, prompt_tokensresp.prompt_tokens, completion_tokensresp.completion_tokens) calls assistant_msg.get(tool_calls) or [] if not calls: stop_reason done turns.append(turn) return RunResult(assistant_msg.get(content) or , turns, total_tokens, stop_reason, n_calls) # 防御模型一口气调 50 个工具多半是失控了 if len(calls) self.max_parallel: calls calls[: self.max_parallel] messages.append({ role: tool, tool_call_id: calls[-1][id], content: f[ERROR] 单次最多调用 {self.max_parallel} 个工具已截断。, }) n_calls len(calls) results self.tools.invoke_many(calls) turn.tool_results results for call_id, res in results: messages.append(res.to_message(call_id)) if self.verbose: mark OK if res.ok else ERR print(f [{mark}] {call_id[:8]} {res.elapsed_ms:.0f}ms - f{str(res.data)[:80] if res.ok else res.error}) turns.append(turn) # 达到上限强制收口让模型基于已有信息给结论 messages.append({role: user, content: 已达轮次上限请基于已有信息直接给出结论 并说明哪些信息未能确认。}) resp self.llm.chat(messages) total_tokens resp.prompt_tokens resp.completion_tokens return RunResult(resp.assistant_message.get(content) or , turns, total_tokens, max_turns, n_calls)六、代码 4LLM 客户端tool_calls 版本# llm/client.py from __future__ import annotations import json import os import time from dataclasses import dataclass import requests dataclass class ChatResponse: assistant_message: dict # 直接可 append 进 messages prompt_tokens: int completion_tokens: int finish_reason: str class ChatClient: def __init__(self, model: str gpt-4o-mini, base_url: str | None None, api_key: str | None None, timeout: int 60, max_retries: int 3): self.model model self.base_url (base_url or os.getenv(LLM_BASE_URL, https://api.openai.com/v1)).rstrip(/) self.api_key api_key or os.environ.get(LLM_API_KEY, ) self.timeout timeout self.max_retries max_retries self.s requests.Session() def chat(self, messages: list[dict], tools: list[dict] | None None, tool_choice: str | dict auto, temperature: float 0.0) - ChatResponse: payload { model: self.model, messages: messages, temperature: temperature, } if tools: payload[tools] tools payload[tool_choice] tool_choice last None for attempt in range(self.max_retries): try: r self.s.post(f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, jsonpayload, timeoutself.timeout) if r.status_code 429 or r.status_code 500: raise RuntimeError(f{r.status_code}: {r.text[:200]}) if r.status_code 400: # 400 常见于 messages 里 tool_call_id 对不上直接抛重试无用 raise ValueError(fHTTP {r.status_code}: {r.text[:800]}) data r.json() break except (requests.RequestException, RuntimeError) as e: last e time.sleep(2 ** attempt) else: raise RuntimeError(fLLM 请求失败: {last}) msg data[choices][0][message] usage data.get(usage, {}) or {} return ChatResponse( assistant_messagemsg, prompt_tokensusage.get(prompt_tokens, 0), completion_tokensusage.get(completion_tokens, 0), finish_reasondata[choices][0].get(finish_reason, ), )踩坑提示messages里如果有一条roleassistant带tool_calls后面必须紧跟对应数量的roletool消息且tool_call_id一一对应否则 API 直接 400。所以永远不要手动删改 messages 中间的某一条要做上下文裁剪必须从系统提示之后、用户首轮之前开始切且保证 assistanttool 成对。七、实战一次调用两个工具# demo_parallel.py from llm.client import ChatClient from tools.runtime import RegisteredTool, ToolRuntime from tools.biz_tools import CalcArgs, WeatherArgs, calc, get_weather from agent.fc_agent import ToolCallingAgent rt ToolRuntime() rt.register(RegisteredTool(get_weather, 查询城市某天的气温摄氏度, get_weather, arg_modelWeatherArgs, timeout8)) rt.register(RegisteredTool(calc, 计算数学表达式, calc, arg_modelCalcArgs, timeout3)) agent ToolCallingAgent(ChatClient(), rt, max_turns6) res agent.run(北京和上海今天的气温差多少度) print(res.answer, | 轮次:, len(res.turns), | 工具调用:, res.tool_calls_count)预期轨迹turn 1: tool_calls [get_weather(北京, today), get_weather(上海, today)] # 并行 - OK 26.5 / OK 31.2 turn 2: tool_calls [calc(31.2-26.5)] - OK 4.7 turn 3: 北京 26.5°C上海 31.2°C相差 4.7°C。3 轮完成两个天气查询是并发的。换成第 1 篇的 ReAct 文本协议需要 5 步、5 次串行 LLM 调用。这就是结构化调用在延迟上的直接收益。再看并行是否真的生效加个计时import time from tools.runtime import RegisteredTool, ToolRuntime rt ToolRuntime() rt.tool(slow_api, 模拟一个耗时 2 秒的查询, timeout10) def slow_api(x: str) - str: time.sleep(2) return fresult for {x} t0 time.perf_counter() calls [{id: fc{i}, name: slow_api, arguments: {x: str(i)}} for i in range(5)] rt.invoke_many(calls) print(f5 个 2 秒任务耗时 {time.perf_counter() - t0:.2f}s) # 约 2.0s而非 10s八、让 Agent 学会自我纠错结构化调用解决的是「格式错」但参数错、工具选错、结果不可用依然会发生。我们的策略是分层处理错误层检测点处理JSON 格式客户端由 API 保证几乎不发生参数类型/枚举ToolRuntime._invoke_onepydantic 校验返回字段级错误未知工具名ToolRuntime._invoke_one返回可用工具列表业务失败查无数据工具内部返回{ok: false, reason: ...}不要抛异常结果不满足需求模型判断换工具 / 换参数重试关键是工具函数永远不要raise要返回结构化的失败。看个对比# 错误做法 def get_user(uid: str) - dict: row db.query(uid) if not row: raise ValueError(not found) # 模型只看到 ValueError: not found return row # 正确做法 def get_user(uid: str) - dict: row db.query(uid) if not row: return {ok: False, reason: f用户 {uid} 不存在, suggestion: 可先用 list_users 查正确的 uid或确认 uid 是否为 8 位数字} return {ok: True, user: row}后者给了模型「下一步该干什么」一轮就能自愈。前者只会让模型盲目重试同一个 uid。再补一个自动重试上限同一个工具同样参数失败两次就把该工具从本轮可用列表里摘掉逼模型换路子。# agent/retry_guard.py from collections import defaultdict class RetryGuard: 记录 (工具名, 参数指纹) 的失败次数超限则临时封禁该工具。 def __init__(self, max_same_fail: int 2): self.max_same_fail max_same_fail self._fails: dict[tuple[str, str], int] defaultdict(int) self._banned: set[str] set() staticmethod def _fp(args: dict | str) - str: import json return json.dumps(args, sort_keysTrue, ensure_asciiFalse) if isinstance(args, dict) else str(args) def record(self, name: str, args: dict | str, ok: bool) - None: key (name, self._fp(args)) if ok: self._fails.pop(key, None) else: self._fails[key] 1 if self._fails[key] self.max_same_fail: self._banned.add(name) def banned(self) - set[str]: return self._banned def filter_tools(self, schemas: list[dict]) - list[dict]: return [s for s in schemas if s[function][name] not in self._banned]在循环里接上guard RetryGuard(max_same_fail2) # 每轮请求前 schemas guard.filter_tools(self.tools.schemas()) resp self.llm.chat(messages, toolsschemas) # 每轮执行后 for cid, r in results: guard.record(name_by_id[cid], args_by_id[cid], r.ok)九、测试不花钱验证工具系统# tests/test_fc.py import json import pytest from tools.runtime import RegisteredTool, ToolRuntime, ToolResult from tools.schema import func_to_schema from pydantic import BaseModel, Field from typing import Annotated class DivArgs(BaseModel): a: Annotated[float, Field(description被除数)] b: Annotated[float, Field(description除数)] def divide(a: float, b: float) - float: 两数相除 return a / b pytest.fixture def rt(): r ToolRuntime() r.register(RegisteredTool(divide, 两数相除, divide, arg_modelDivArgs, timeout1)) return r def test_schema_has_description_and_required(rt): s rt.schemas()[0] assert s[type] function assert s[function][name] divide assert s[function][parameters][required] [a, b] assert 被除数 in json.dumps(s, ensure_asciiFalse) assert s[function][parameters][additionalProperties] is False def test_unknown_tool_returns_error(rt): ok, res rt.invoke_many([{id: 1, name: nope, arguments: {}}])[0] assert not res.ok assert 可用工具 in res.error def test_validation_error_is_readable(rt): _, res rt.invoke_many([{id: 1, name: divide, arguments: {a: abc, b: 2}}])[0] assert not res.ok assert 参数校验失败 in res.error def test_tool_exception_is_captured(rt): _, res rt.invoke_many([{id: 1, name: divide, arguments: {a: 1, b: 0}}])[0] assert not res.ok assert ZeroDivisionError in res.error def test_parallel_order_preserved(rt): calls [{id: fi{i}, name: divide, arguments: {a: i, b: 2}} for i in range(1, 6)] out rt.invoke_many(calls) assert [i for i, _ in out] [i1, i2, i3, i4, i5] assert [r.data for _, r in out] [0.5, 1.0, 1.5, 2.0, 2.5] def test_message_shape(rt): res ToolResult(True, data{x: 1}) m res.to_message(call_abc) assert m {role: tool, tool_call_id: call_abc, content: {x: 1}} err ToolResult(False, error炸了).to_message(call_abc) assert err[content].startswith([ERROR]) class SlowArgs(BaseModel): n: int def test_timeout(rt): import time def slow(n: int) - str: time.sleep(5) return done rt.register(RegisteredTool(slow, 慢工具, slow, arg_modelSlowArgs, timeout0.5)) _, res rt.invoke_many([{id: s, name: slow, arguments: {n: 1}}])[0] assert not res.ok and 超时 in res.errorpytest tests/test_fc.py -q # 8 passed十、Function Calling vs 文本协议怎么选维度文本协议ReActFunction Calling格式稳定性依赖 prompt约 1%-5% 漂移Schema 约束0.5%并行调用不支持原生支持参数校验运行时才发现解析阶段拦截模型兼容性任何模型需模型支持 tool calling或用 Grammer/JSON mode 兜底可解释性Thought 文本天然可见需在 prompt 里额外要求 reasoning复杂控制流灵活需配合状态机建议主力走 Function Calling这是生产环境的默认选择。如果模型不支持比如某些小模型或本地模型用JSON Mode strict schema 重试兜底要求模型输出{thought, calls: [...]}用 pydantic 校验失败就把错误回灌重试最多 2 次。这个兜底路径的代码结构和 Function Calling 几乎一致只是多了个解析层。兜底示例from pydantic import BaseModel, ValidationError from typing import Annotated, Literal class Call(BaseModel): tool: str arguments: dict class PlanOut(BaseModel): thought: Annotated[str, Field(description本轮推理)] calls: list[Call] Field(default_factorylist) final_answer: str | None None def parse_with_retry(llm, messages, tools_desc: str, max_retry: int 2) - PlanOut: schema_hint PlanOut.model_json_schema() for attempt in range(max_retry 1): msgs messages [{role: user, content: f可用工具: {tools_desc}\n请严格输出如下 JSON Schema 的 JSON不要 markdown 代码块\n f{json.dumps(schema_hint, ensure_asciiFalse)}}] text llm.chat(msgs, temperature0).assistant_message[content] text re.sub(r^[a-z]*\n|\n$, , text.strip(), flagsre.M) try: return PlanOut.model_validate_json(text) except ValidationError as e: if attempt max_retry: raise messages messages [{role: assistant, content: text}, {role: user, content: fJSON 不合法: {e}。请只输出纯 JSON。}]十一、小结Schema 是新的 prompt。description里写清语义、格式示例、边界条件比在 system prompt 里喊十句「请仔细检查参数」有用得多。工具结果永远结构化、永不抛异常。带上失败原因和「下一步建议」模型才能自愈。并行要显式声明安全性。读并行、写串行parallel_safe这个标志值得每个工具都填一次。工具数量别超过 15 个超了就做分层路由。给失败计数加封禁同一工具同参数失败两次就摘掉它避免模型原地打转。下一篇《Agent 的记忆系统短期、长期与工作记忆》我们会给这个 Agent 加上跨会话记忆如何压缩历史、如何做向量检索、如何避免「记忆污染」。依赖pip install pydantic requests pytest