AI Agent 工具调用生产模式：6 个月全天候运行总结的 8 条经验

Others 2026-07-14 17:38:13 7

让一个大语言模型（LLM，即大型语言模型）调用一次工具很容易。但要让它每天可靠地调用 500 次，连续跑 6 个月——这是完全不同的挑战。

2026 年了，OpenAI、Claude、Gemini 这些主流大语言模型都原生支持函数调用（Function Calling）。用 LangChain、CrewAI 这类框架，10 行代码就能把工具接上去。

但生产环境的可靠性是另一回事。我的自动化流水线每天执行 400-600 次工具调用——网页搜索、数据库查询、内容生成、发布、API 集成。第一个月遇到的问题简直让人头皮发麻：

参数幻觉：LLM 发明了工具 schema 里根本不存在的参数名
调用不一致：同一个工具 3 次调用传了 3 套不同的参数
超时没控制：一个慢接口把整个流水线卡了 60 多秒
危险操作没确认：一条 DELETE 直接执行了，没有任何确认机制

这不是框架的问题，是架构的问题。下面这 8 个模式是我一点点踩坑踩出来的解决方案。

1. 参数校验墙

LLM 对工具 schema 的理解是模糊的。它会传字符串 "42" 当作整数，会发明不存在的枚举值，或者传根本不在 schema 里的参数。

解决方案：在工具边界做校验。不要信任 LLM。

VALID_DOMAINS = {"example.com", "myservice.io"} def send_email(to: str, subject: str, body: str):
if "@" not in to or to.split("@")[1] not in VALID_DOMAINS:
return {"error": f"Domain not in whitelist: {to}"}
if len(subject) > 200:
return {"error": "Subject too long"}
return smtp.send(to, subject, body)

结果：在我的流水线上，18% 的工具调用被这个校验墙拦下了——不是因为 LLM "不行"，而是因为工具边界对语言模型来说本身就是模糊的。

2. 幂等性 Key

最隐蔽的 bug：工具调用超时了，框架重试，两次调用都成功了——但只收到一个响应。结果：重复执行。

import hashlib, json CACHE = {} def execute_with_idempotency(tool_name: str, args: dict, ttl=3600):
key = hashlib.sha256(
f"{tool_name}:{json.dumps(args, sort_keys=True)}".encode()
).hexdigest()
if key in CACHE:
return CACHE[key]
result = actual_execute(tool_name, args)
CACHE[key] = result
return result

这一招直接消灭了 100% 的重复发布事故。连续 6 个月，零发生。

3. 超时 + 优雅降级

一次慢工具调用卡住 60 秒，整个流水线就全停了。解决方案简单粗暴：

import asyncio async def call_with_fallback(tool_fn, *args, timeout=15):
try:
return await asyncio.wait_for(tool_fn(*args), timeout=timeout)
except asyncio.TimeoutError:
return {"error": f"timeout after {timeout}s", "fallback": True}

关键思路：返回降级结果而不是抛异常。Agent 读到 fallback: True，自己决定是重试、跳过还是换个工具。流水线不会崩。

4. 工具注册中心（别搞装饰器满天飞）

当工具超过 20 个，@tool 装饰器散落在 15 个文件里，调试起来简直是噩梦。搞个集中式注册中心，审计变成查表。

TOOL_REGISTRY = {} def register(name: str, schema: dict):
def decorator(func):
func._meta = {"name": name, "schema": schema}
TOOL_REGISTRY[name] = func
return func
return decorator def audit_tools():
for name, fn in sorted(TOOL_REGISTRY.items()):
schema_type = fn._meta["schema"].get("type", "?")
print(f" {name} [{schema_type}]: {fn.__doc__ or 'no doc'}") def call(name: str, args: dict):
if name not in TOOL_REGISTRY:
return {"error": f"Unknown tool: {name}"}
return TOOL_REGISTRY[name](**args)

5. 危险操作的双重确认门

有些操作根本不应该在单次 LLM 推理中直接执行。删除、覆盖、发布、执行 SQL——这些都需要二次确认。

我的方案：敏感工具要求 confirmed: bool = False 参数。LLM 必须显式确认才能真正执行。

SENSITIVE = {"delete", "publish", "execute_sql", "overwrite"} def needs_confirmation(tool_name: str, confirmed: bool = False):
if tool_name in SENSITIVE and not confirmed:
return {"status": "needs_confirmation", "message": f"About to execute {tool_name}, confirm?"}
return None

确认行为本身会被记录。每一个"差点删库"的事件都变成了审计数据，而不是灾难。

6. 调用回放日志

出问题的时候，你必须知道 Agent 到底干了什么。搞个事务日志，记录每次工具调用——输入、输出、耗时：

import sqlite3, json, time class ToolLogger:
def __init__(self, db="tool_calls.db"):
self.conn = sqlite3.connect(db)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS calls (
id INTEGER PRIMARY KEY,
session TEXT, tool TEXT,
args TEXT, result TEXT,
ms INTEGER, ts TEXT
)""") def log(self, session, tool, args, result, ms):
self.conn.execute(
"INSERT INTO calls VALUES (NULL,?,?,?,?,?,datetime('now'))",
(session, tool, json.dumps(args), json.dumps(result), ms))
self.conn.commit() def replay(self, session):
return self.conn.execute(
"SELECT * FROM calls WHERE session=? ORDER BY id", (session,)).fetchall()

之前：调试全靠猜。之后：直接 SQL 查询。平均调试时间从 45 分钟降到 3 分钟。

7. 工具调用的熔断器

当工具连续失败时，Agent 应该停下来升级问题，而不是继续在注定失败的调用上烧 token。

breaker = {"fails": 0, "last": 0} def is_open(max_fails=3, cooldown=60):
if breaker["fails"] >= max_fails:
if time.time() - breaker["last"] < cooldown:
return False # circuit open

breaker["fails"] = 0 # cooldown reset

return True

简单。有效。连续失败 3 次就直接熔断，防止死亡螺旋。

8. 输入标准化 & 类型强制转换

LLM 对类型的处理很不一致。字符串 vs 整数、枚举 vs 同义词、蛇形命名 vs 驼峰命名。在工具边界加一层标准化来处理这些：

def normalize(schema: dict, args: dict) -> dict:
result = {}
for key, spec in schema.items():
if key not in args:
if spec.get("required"):
raise ValueError(f"Missing: {key}")
continue
v = args[key]
t = spec.get("type")
if t == "int": result[key] = int(v)
elif t == "float": result[key] = float(v)
elif t == "bool": result[key] = str(v).lower() in ("true","1","yes")
elif "enum"