模型按 JSON Schema "填表", 你的代码验表并执行 — 校验闸门是 agent 与玩具的分水岭
required 漏标 = 模型理直气壮漏传工具调用像点外卖: 模型只会照着菜单"念订单"(输出一段 JSON 意图), 而做菜(执行函数)、验单(参数校验)、送餐(结果回填)全是你后厨的事。JSON Schema 就是那份菜单格式约定——description 写得越清楚、enum 圈得越死, 模型下的"订单"就越少出错; 而校验闸门决定错误订单是被退回重下(回喂自愈), 还是直接把厨房炸了(未校验执行)。记住: 模型的参数是"猜"出来的, 你的校验才是"确定"的。
name / description / parameters。参数类型、枚举、范围全写进 Schema——Schema 是你与模型之间唯一的接口契约。 {"type": "object",
"properties": {"status": {"type": "string",
"enum": ["pending", "paid"]}},
"required": ["status"]} # 关键: enum 圈死取值, required 圈死必填"description": "按状态查询【当前用户】的订单。 不要用于查询其他用户; 结果最多 20 条/页" # 关键: 边界条件写进 description, 比事后校验更省一轮
arguments 是 JSON 字符串, 必须 json.loads; Anthropic 的 input 已经是 dict。两家的坑不一样。 args = json.loads(call.function.arguments) # OpenAI: 字符串 → dict # 直接 call.function.arguments["status"] # → TypeError: string indices must be integers
args = SearchOrders.model_validate_json(raw) # 校验 + 类型收口 # 模型给了 status="PAID" → # ValidationError: status: Input should be 'pending' or 'paid'
auto 自主决定、required 必须调一个、指定名字强制调某个。做"必须走流程"的场景用 required。 client.chat.completions.create(
..., tool_choice="required") # → 强制产出一次 tool_call
# 关键: 结构化抽取任务用 required, 防止模型自由发挥说废话tool_call_id。 results = await asyncio.gather(*[run(c) for c in calls]) for call, r in zip(calls, results): # gather 保序: 结果序=发起序 messages.append({"role": "tool", "tool_call_id": call.id, ...}) # 关键: 保序的是 gather; 换 as_completed 就得按 call.id 配对
strict: true(OpenAI)后, 模型在解码阶段被约束在 Schema 合法空间内, 生成的参数 100% 可解析、100% 含必填字段。 {"type": "function", "function": {..., "strict": true}}
# 关键: strict 下 additionalProperties 必须 false, 全字段必填
# 可选语义用 "type": ["string", "null"] 表达except ValidationError as e: return json.dumps({"error": "invalid_arguments", "detail": str(e)[:300]}) # → 模型读到错在哪一个字段, 下一轮自动改对
next_cursor 分页。 return {"rows": [slim(r) for r in rows[:20]], "next_cursor": cur, "total_estimate": n} # 关键: 2MB 全量结果会让之后每一轮都为它付 token 钱
get_order_with_logistics(id) 一个顶俩, 模型少走两轮循环, 你少防一半的中间态。 # 细碎: 模型要连调 3 次才拼出答案, 每次都是出错机会 get_order(id) + get_logistics(id) + get_payment(id) # 粗粒: get_order_detail(id) 一次带全 → 少 2 轮循环
REGISTRY = {"search_orders": (fn, "read"),
"issue_refund": (fn, "critical")}
# 关键: critical 工具先挂起等审批, 绝不静默放行key = f"refund:{order_id}:{amount}" # 由业务参数派生 INSERT INTO idem_keys(key) ON CONFLICT DO NOTHING # rowcount=0 → 重复请求, 返回上次结果, 不再执行
github__create_issue, 冲突在注册时检测。 name = f"{server}__{tool}" # github__create_issue # 裸用 create_issue: # → 两个 server 都有, 模型调用不知道打到哪一个
手写 JSON Schema 又长又易错。用 pydantic 模型当"单一事实源": 校验、文档、Schema 三者永远同步。
from pydantic import BaseModel, Field from enum import Enum class OrderStatus(str, Enum): pending = "pending"; paid = "paid"; shipped = "shipped" class SearchOrders(BaseModel): """按状态查询当前用户的订单, 默认返回最新的 20 条""" # docstring → 工具描述 status: OrderStatus = Field(..., description="订单状态") limit: int = Field(20, ge=1, le=100, description="每页条数") schema = SearchOrders.model_json_schema() # pydantic v2 直接产出合法 JSON Schema tools = [{"type": "function", "function": { "name": "search_orders", "description": SearchOrders.__doc__, "parameters": schema}}]
chat.completions 上的标准流程, 记住三步: 注册 tools → 判 tool_calls → parse arguments。
resp = client.chat.completions.create(
model="gpt-4.1", messages=messages, tools=TOOLS)
msg = resp.choices[0].message
messages.append(msg) # assistant 消息先入列
if msg.tool_calls:
for call in msg.tool_calls:
args = json.loads(call.function.arguments) # 关键: 字符串 → dict
result = dispatch(call.function.name, args)
messages.append({"role": "tool", "tool_call_id": call.id,
"content": json.dumps(result, ensure_ascii=False)})
Anthropic 的工具调用藏在 content blocks 里, input 已经是 dict——但 stop_reason 判断不能省。
resp = client.messages.create(
model="claude-sonnet-4-5", max_tokens=1024,
tools=[{"name": "search_orders",
"description": "按状态查询当前用户的订单",
"input_schema": {"type": "object",
"properties": {"status": {"type": "string",
"enum": ["pending", "paid"]}},
"required": ["status"]}}],
messages=messages)
if resp.stop_reason == "tool_use":
for block in resp.content:
if block.type == "tool_use":
result = dispatch(block.name, block.input) # input 已是 dict
查天气+查日历+查机票互不依赖, 串行 3 次往返太慢。并行跑, 回填时按 id 对号入座。
calls = resp.message.tool_calls # → [call_a, call_b, call_c] results = await asyncio.gather( *[run_tool(c) for c in calls], return_exceptions=True) for call, r in zip(calls, results): # gather 保序: 结果序=发起序 body = ("工具内部错误: " + str(r)[:150]) if isinstance(r, Exception) \ else json.dumps(r, ensure_ascii=False) messages.append({"role": "tool", "tool_call_id": call.id, "content": body}) # → 3 次串行 2.4s → 并行 0.9s; 异常也不崩, 变成错误观察
模型把 "PAID" 传给了小写枚举。别报 500, 把 ValidationError 原文喂回去——它一看就懂。
from pydantic import ValidationError def guarded_tool(model_cls, fn, raw_args: str) -> str: try: args = model_cls.model_validate_json(raw_args) # 校验 + 收口 return json.dumps(fn(args), ensure_ascii=False) except ValidationError as e: return json.dumps({"error": "invalid_arguments", "detail": str(e)[:300]}) # → 模型读到 "status: Input should be 'pending' or 'paid'" # 下一轮自动改传 "paid", 无需人工介入
订单表百万行, 模型一句"查所有待支付订单"就能拉爆上下文。工具返回值要按"模型视角"瘦身。
def search_orders(status: str, cursor: str | None = None, limit: int = 20): rows, next_cursor = db.query(status, cursor=cursor, limit=limit) return { "total_estimate": count(status), # 给模型全局感, 防止无脑翻页 "rows": [slim(r) for r in rows], # 只留 id/amount/status, 丢大字段 "next_cursor": next_cursor, # None 表示没有下一页 } # → 每页 3KB 而不是 300KB; 模型按 cursor 自己决定翻不翻页
网络抖动后模型重发同一个退款调用。幂等键 + 唯一约束, 让"重试"与"重复执行"分家。
def create_refund(order_id: str, amount_cents: int): key = f"refund:{order_id}:{amount_cents}" # 幂等键由业务参数派生 with db.transaction(): inserted = db.execute( "INSERT INTO idem_keys(key) VALUES (%s) ON CONFLICT DO NOTHING", (key,)) if inserted.rowcount == 0: return {"status": "duplicate", # → 重放被识别 "refund_id": lookup(key)} # 返回首次结果 return refund_now(order_id, amount_cents)
所有工具收进一张注册表, 危险等级显式声明, 分发只走白名单——权限不再散落在 if/else 里。
REGISTRY = {
"search_orders": (search_orders, "read"), # 只读: 默认放行
"create_ticket": (create_ticket, "write"), # 写入: 记审计日志
"issue_refund": (issue_refund, "critical"), # 高危: 人工确认
}
def dispatch(name: str, args: dict, user):
fn, tier = REGISTRY[name] # 关键: 白名单, 不用 getattr
if tier == "critical" and not user.approved(name, args):
return {"status": "pending_approval"} # 挂起等审批, 不静默放行
audit.log(user.id, name, tier, args)
return fn(**args)
业务长到 30+ 工具, 选择错误率肉眼可见地涨。拆两级: 便宜模型选域, 主模型只在域内 2~3 个工具里挑。
DOMAINS = {"订单": ["search_orders", "get_logistics"],
"售后": ["create_refund", "list_returns"],
"物流": ["track_shipment"]}
# 第一级: 小模型做分类, 成本≈0
domain = llm_small("用户问题属于哪个域? " + str(list(DOMAINS)))
tools = pick(DOMAINS[domain])
resp = llm.create(messages=messages, tools=tools) # 候选 30 → 2
# → 选择准确率回升, 单次任务 token 反而降(少塞 27 份 schema)
下游慢是常态。超时后不能让任务吊死, 也不能静默吞掉——给模型一个带建议的观察, 让它决策。
async def resilient_tool(coro, timeout=10): try: return await asyncio.wait_for(coro, timeout) except asyncio.TimeoutError: # 关键: 超时也是观察, 让模型决定重试/换路/收尾 return json.dumps({"error": "timeout", "seconds": timeout, "suggestion": "可稍后重试, 或改用缓存数据源"}) # → 再没有"整个任务卡死在一个 30s 的下游"的连环雪崩
# 错: "查询订单信息" # → 和 get_order 语义撞车 # 对: "按状态+时间范围批量查询【当前用户】订单, 单工具不带详情"
date 这种名字看不出是下单日还是支付日. 正解: 语义化命名 + description 注明格式。 # 错: {"date": "2026-01-01"} # → 模型自作主张填当天 # 对: {"paid_after": "ISO8601, 如 2026-01-01T00:00:00+08:00"}
required, 模型认为可省. 正解: 必填字段全列入 required。 # 错: properties 里有 status 但 required: [] # → 半数调用缺 status # 对: "required": ["status"]
enum 列全合法值。 # 错: {"status": {"type": "string"}} # → "payed"(拼错) 照样生成 # 对: {"status": {"type": "string", "enum": ["pending", "paid"]}}
arguments 是 JSON 字符串. 正解: json.loads 后再用。 # 错: call.function.arguments["status"] # → TypeError: string indices must be integers # 对: args = json.loads(call.function.arguments)
# 错: db.execute(f"SELECT * FROM t WHERE c = '{args['c']}'") # 对: db.execute("SELECT ... WHERE c = %s", (validated.c,))
# 错: tools=ALL_40_TOOLS # → schema 就吃掉几千 token # 对: tools = pick(DOMAINS[route_first(messages)])
# 错: return json.dumps(db.fetch_all(sql)) # → 2MB 观察入列 # 对: return {"rows": rows[:20], "next_cursor": cur}
# 错: def refund(o, amt): refund_now(o, amt) # → 重放=双倍退款 # 对: key=f"refund:{o}:{amt}"; ON CONFLICT DO NOTHING
as_completed 按完成顺序返回, 却按位置 zip. 正解: 用 gather 保序或按 call.id 配对。 # 错: for c, fut in zip(calls, as_completed(tasks)): ... # → 错配 # 对: results = await asyncio.gather(*tasks) # 保序, 或按 call.id 查表
# 错: result = requests.get(url, timeout=10) # → ConnectionError 冒泡 # 对: except Exception as e: return {"error": str(e)[:150]}
# 错: "date": "2026-01-01" # → 被当 UTC, 北京时间差 8h # 对: "date": "2026-01-01T00:00:00+08:00" # description 写明必须带时区
next_cursor, 并在 description 说明。 # 错: return rows[:20] # → 模型不知道还有下一页 # 对: return {"rows": rows, "next_cursor": cur} # None=到底了
# 错: filter: {range: {time: {gte: ..., lte: ...}}} # → 3 层嵌套乱填 # 对: time_from / time_to 两个顶层 string 字段
# 错: oneOf: [按 id 查, 按名查, 按邮箱查] # → 3 选 1 还经常混填 # 对: get_user_by_id / get_user_by_email 两个工具
get_user. 正解: 注册时加来源前缀并检测冲突。 # 错: 两个 get_user 直接注册 # → 后者被静默覆盖 # 对: "github__get_user" / "crm__get_user" + 冲突即报错
# 错: log.info("tool=%s args=%s", name, args) # → PII 全落盘 # 对: log.info("tool=%s args=%s", name, redact(args))
# 错: 手写 schema 写 amount, 函数已改名 amount_cents # → 运行时才炸 # 对: schema = ToolModel.model_json_schema() # 单一事实源
"20" + 1 之类类型错乱, 排序按字典序. 原因: 模型把 number 生成成 string 且闸门没拦. 正解: Schema 标 type:number, pydantic 强转。 # 错: {"limit": "20"} # → 字符串混进 SQL LIMIT 语法错 # 对: limit: int = Field(20) # pydantic 校验层强转+拦截
# 错: run_tool(call); # 忘了 messages.append(...) → 模型只能编 # 对: messages.append({"role": "tool", "tool_call_id": call.id, "content": body})