Function Calling 的工程化实践:函数定义设计、幻觉防护与失败重试的 12 个坑

一、Function Calling 是 Agent 的基石,也是翻车重灾区

如果说单轮对话只是"调用一次 LLM",那 Function Calling 就是让 LLM **从"会说"变成"能做"**的那一步:模型不再只吐文本,而是输出结构化调用意图,由你的代码去执行真实操作——查订单、发消息、写数据库。

但所有做过生产的工程师都有同感:Function Calling 在 Demo 里百发百中,上线后天天翻车。翻车点集中在三个层面:函数定义写得不够好,模型就选错、猜错;模型幻觉产生,应用层毫无防护;工具执行失败,重试策略反而放大事故。这篇文章把 12 个最常见的坑按这四个层面拆开讲,每个坑都给出"现象 → 根因 → 解法"。

二、函数定义设计:模型一半的"错"都是你造成的

坑 1:description 写得像废话,模型只能靠函数名瞎猜

模型做工具选择的唯一依据,就是 name 和 description。很多人写:

{"name": "get_order", "description": "查询订单"}

"查询订单"四个字,模型既不知道参数格式,也不知道返回什么、什么时候该用。正确的描述要回答三个问题:这个函数做什么、参数是什么格式、什么场景下应该调用它:

{
"name": "query_order",
"description": "根据订单号查询订单当前状态与物流轨迹。当用户询问'我的订单到哪了''退款到哪了'时使用。订单号是 13 位数字。",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"pattern": "^\\\\d{13}$",
"description": "13 位数字订单号,例如 2024010112345"
}
},
"required": ["order_id"],
"additionalProperties": False
}
}

规则:description 里写"何时用 + 参数格式 + 一句话示例",比堆砌功能描述有效得多。

坑 2:参数约束能省则省,幻觉参数就是这么来的

参数 schema 是 JSON Schema,它不只是文档,更是运行时校验的合同。不写 pattern、enum、minimum,模型就会编出 order_id="第3个订单" 这种值。上面例子里的 pattern 和 enum 就是最便宜的幻觉拦截器:

"status": {"type": "string", "enum": ["pending", "shipped", "done"]},
"amount": {"type": "number", "minimum": 0.01}

坑 3:一次塞 30 个工具,选择准确率断崖下跌

工具越多,模型"选错"的概率越高,这是有实测结论的。生产上永远不要把所有工具一股脑暴露给模型,而是按意图预路由,只给模型当前场景需要的 3~5 个:

def pick_tools(user_query: str) –> list[dict]:
"""按意图筛选工具,控制单次暴露数量"""
if "退款" in user_query or "退货" in user_query:
return [TOOLS["query_order"], TOOLS["refund_order"]]
if "物流" in user_query:
return [TOOLS["query_order"], TOOLS["query_logistics"]]
return [TOOLS["query_order"]]

坑 4:参数名用缩写,模型理解成本翻倍

sid、uid、amt 这种缩写,对模型来说等于没写。参数名要"见名知义":order_id、user_id、refund_amount,配合 description 里给真实示例值,模型生成参数的准确率会明显提升。参数名的成本是零,收益是实打实的。

三、幻觉防护:别把安全寄托在模型的自律上

坑 5:模型可能调用一个不存在的函数

模型是概率生成,它完全可能输出一个你根本没注册过的函数名(尤其在模型版本更新、system prompt 很长之后)。应用层必须做白名单校验,未知函数一律拒绝并回传给模型:

WHITELIST = {t["function"]["name"] for t in TOOLS}

def is_allowed(name: str) –> bool:
return name in WHITELIST # 不在白名单 = 幻觉,直接拦截

坑 6:模型会编造参数,json.loads 之后必须 schema 校验

tool_calls 里的 arguments 是字符串,解析成 JSON 后必须用定义的 schema 再校验一次。模型输出的参数可能缺字段、类型错误、值非法——这一层校验没有,后面工具函数就要自己兜底,而工具函数往往兜不住:

from jsonschema import validate, ValidationError

try:
args = json.loads(tc.function.arguments)
validate(instance=args, schema=SCHEMAS[name]) # 严格按定义校验
except (json.JSONDecodeError, ValidationError) as e:
# 把错误信息作为 tool 结果回传,模型会自我修正参数
results.append({"tool_call_id": tc.id, "content": f"参数非法:{e}"})

注意一个关键设计:校验失败不要直接终止,而是把错误信息作为 tool 结果回传给模型,让模型自己修正参数重来一轮——这是"模型自愈"的标准做法。

坑 7:敏感操作没有人工闸门,幻觉直接变成事故

即使参数合法,模型也可能在错误的上下文里发起退款、转账。权限类、资金类、删除类操作,必须在应用层加人工确认(interrupt 或工单审批流),LLM 永远不应该直接拥有执行敏感操作的权限。这一条是合规红线,不是工程偏好。

四、失败重试:策略错了,重试就是二次事故

坑 8:对副作用操作盲目重试,等于重复扣款

工具执行失败,很多人无脑 for i in range(3): try: …。但退款、发消息这类非幂等操作,重试一次就是重复执行一次。正确姿势:要么拒绝重试非幂等操作,要么给每次调用生成幂等键,让下游网关去重:

async def refund_order(order_id: str, amount: float, idempotency_key: str = None):
headers = {}
if idempotency_key:
headers["Idempotency-Key"] = idempotency_key # 同一 key 下游只生效一次
# 重试时复用同一个 key,而不是生成新的

坑 9:没有超时控制,一个卡死的工具拖垮整个 Agent 循环

外部 API 可能永远不返回。工具执行必须包超时,超时按"瞬时失败"处理并返回给模型:

async def _run_with_timeout(self, fn, args, timeout: float = 10.0):
return await asyncio.wait_for(fn(**args), timeout=timeout)

坑 10:重试一刀切,瞬时错误和永久错误混为一谈

网络抖动可以重试,参数非法重试一百次也没用。重试前先分类:TimeoutError/ConnectionError 走指数退避重试,ValidationError/PermissionError 直接失败回传:

async def _run_with_retry(self, name, args):
for attempt in range(self.max_retries):
try:
return await self._run_with_timeout(self.registry[name], args)
except (asyncio.TimeoutError, ConnectionError) as e:
if attempt == self.max_retries – 1:
raise
await asyncio.sleep(2 ** attempt + 0.1) # 指数退避 + 抖动

五、工程细节:两个容易被忽视的坑

坑 11:工具结果原样塞回上下文,把窗口撑爆

一个查询返回 50KB 日志,全塞进 messages,token 消耗暴涨、模型注意力被稀释。工具结果必须截断或摘要再回传,通常保留 1~2KB 就够模型做下一步判断:

def truncate(content: str, max_chars: int = 2000) –> str:
return content if len(content) <= max_chars else content[:max_chars] + " …[已截断]"

坑 12:只处理第一个 tool_call,忽略并行调用

新版模型一次可能返回多个 tool_calls(并行调用多个工具)。只取第一个,等于让模型白算,还容易丢结果。必须循环处理,能并行的用 asyncio.gather:

async def execute_all(self, tool_calls):
return await asyncio.gather(*(self.execute_one(tc) for tc in tool_calls))

六、完整参考实现:一个 100 行的 ToolExecutor

把上面所有防护收拢成一个可复用的执行器,再配上多轮 Agent 循环:

import json, asyncio
from openai import AsyncOpenAI
from jsonschema import validate, ValidationError

client = AsyncOpenAI(api_key="sk-…")

class ToolExecutor:
def __init__(self, tools, registry, timeout=10.0, max_retries=3):
self.tools = tools
self.registry = registry # {name: async_callable}
self.whitelist = {t["function"]["name"] for t in tools} # 坑5
self.schemas = {t["function"]["name"]: t["function"]["parameters"] for t in tools}
self.timeout, self.max_retries = timeout, max_retries

async def execute_one(self, tc):
name = tc.function.name
if name not in self.whitelist: # 坑5:白名单
return {"tool_call_id": tc.id, "content": "error: 未知工具"}
try:
args = json.loads(tc.function.arguments)
validate(instance=args, schema=self.schemas[name]) # 坑6:schema 校验
except (json.JSONDecodeError, ValidationError) as e:
return {"tool_call_id": tc.id, "content": f"参数非法:{e}"} # 回传让模型自愈
for attempt in range(self.max_retries): # 坑10:分类重试
try:
out = await asyncio.wait_for(self.registry[name](**args), timeout=self.timeout) # 坑9
return {"tool_call_id": tc.id, "content": truncate(out, 2000)} # 坑11
except (asyncio.TimeoutError, ConnectionError):
if attempt == self.max_retries – 1:
return {"tool_call_id": tc.id, "content": "error: 工具执行超时"}
await asyncio.sleep(2 ** attempt + 0.1)
except Exception as e:
return {"tool_call_id": tc.id, "content": f"error: {e}"}
raise RuntimeError("unreachable")

async def execute_all(self, tool_calls): # 坑12:并行执行
return await asyncio.gather(*(self.execute_one(tc) for tc in tool_calls))

async def agent_loop(user_query: str, max_rounds: int = 5):
messages = [{"role": "user", "content": user_query}]
executor = ToolExecutor(TOOLS, REGISTRY)
for _ in range(max_rounds):
resp = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=executor.tools,
tool_choice="auto",
)
msg = resp.choices[0].message
if not msg.tool_calls: # 不再要工具 → 输出最终答案
return msg.content
messages.append(msg) # 保留模型的工具调用请求
for r in await executor.execute_all(msg.tool_calls):
messages.append({"role": "tool", "tool_call_id": r["tool_call_id"], "content": r["content"]})
return "已达最大轮数,请稍后重试或转人工"

这个执行器的每一行注释,对应上面一个坑。面试时能把它讲清楚,比背十个框架 API 有用得多。

七、12 个坑速查清单

核心心法就一句话:把 Function Calling 当成"不可信输入",函数定义是合同,应用层是最后的校验闸门,失败重试要按副作用分类。 模型会犯错,但工程化可以把错误控制在可观测、可恢复、不产生事故的范围内。