从零写一个 AI Agent (3):多工具协作与错误处理

系列：「从零写一个 AI Agent」 这篇讲：当 Agent 同时调多个工具时，怎么保证不翻车。以及翻车了怎么自己爬起来。

从零写一个 AI Agent (1)：50 行写一个 Function Calling 循环

从零写一个 AI Agent (2)：给 Agent 装上记忆和工具

前两篇写完，有读者说"跑起来了，但太脆弱了"——工具传错参数就崩、网络超时就挂、Agent 说胡话也没人兜底。

说得对。玩具和产品的区别，很大一部分在于错误处理。

并行调用：模型比你想象的能干

第一篇 demo 里有个细节你可能没注意——用户问"杭州天气 + 35*12"，模型一次返回了两个 tool_calls，而不是先查天气再算数。

这就是 LLM 的并行能力。DeepSeek、GPT、Claude 都支持一次响应里返回多个工具调用。我们只要在循环里批量执行就行：

# 并行执行所有工具调用results = []for tc in msg["tool_calls"]:    name = tc["function"]["name"]    args = json.loads(tc["function"]["arguments"])    results.append((tc["id"], name, args, call_tool(name, args)))# 批量写回for tc_id, name, args, result in results:    messages.append({"role": "tool","tool_call_id": tc_id,"content": result,    })

这带来一个有意思的现象：模型会在一次推理中规划好任务依赖图。独立的操作（查天气、算数）并行；有依赖的操作（先搜索再总结）串行。你不需要写 DAG，模型自己会判断。

当然，强依赖场景（"先搜 X 的结果，然后分析"）它会分两轮调用——第一轮搜、看到结果、第二轮分析。这就是 ReAct 模式天然的优势。

问题 1：工具超时了

如果你的工具要去调外部 API（查天气、搜网页），网络是不靠谱的。万一 API 挂了怎么办？

import concurrent.futuresdefcall_tool_with_timeout(name, args, timeout=10):"""带超时的工具调用"""with concurrent.futures.ThreadPoolExecutor() as pool:        future = pool.submit(call_tool, name, args)try:return future.result(timeout=timeout)except concurrent.futures.TimeoutError:returnf"[错误] 工具 {name} 超时（{timeout}秒）"except Exception as e:returnf"[错误] 工具 {name} 执行失败：{str(e)}"

把错误信息作为工具返回内容喂给 LLM——LLM 会自动理解"这个工具挂了"，然后决定怎么办。大多数情况下它会说"天气查询暂时不可用，其他信息如下"或者重试一次。

这里有个重要原则：永远不要把异常抛给循环，要把异常转化为 LLM 能理解的自然语言错误消息。

# ❌ 不要这样try:    result = call_tool(name, args)except Exception:raise# 循环直接崩了# ✅ 要这样try:    result = call_tool(name, args)except Exception as e:    result = f"[工具错误] {name} 执行失败：{e}。请决定是否重试或跳过。"

LLM 很擅长处理这种"软错误"。它知道你给了它一个糟糕的结果，会做出合理判断。

问题 2：工具返回数据太大了

如果你让 Agent 读了一个 10 万行的日志文件，或者搜了一个返回大量内容的网页——结果可能撑爆 context window。

方案：摘要 + 截断。

MAX_TOOL_RETURN = 4000# 工具返回最多 4000 字符deftruncate_result(text):if len(text) <= MAX_TOOL_RETURN:return textreturn (text[:MAX_TOOL_RETURN] +f"\n\n...（结果过长，已截断，共 {len(text)} 字符）")

然后把截断后的结果给 LLM。它知道数据被截断了，会合理地选择"只需要部分数据"或者"换个更精确的查询再试"。

问题 3：模型陷入死循环

Agent 可能会不断调工具、从不直接回答。原因很多：工具设计不当、指令模糊、模型 bug。

解决方案之一：最大轮次限制。

MAX_TURNS = 15for turn in range(MAX_TURNS):    msg = llm(messages)ifnot msg.get("tool_calls"):return msg["content"]# ... 执行工具 ...# 如果超过最大轮次还没回答return"[已达最大推理轮次，请简化问题或重新提问]"

方案之二：工具调用计数器。如果同一个工具被连续调用超过 3 次且没进展，强制终止。

tool_call_count = {}for tc in msg["tool_calls"]:    name = tc["function"]["name"]    tool_call_count[name] = tool_call_count.get(name, 0) + 1if tool_call_count[name] > 3:        result = "[警告：此工具已被连续调用多次，请尝试其他方法或直接回答]"

问题 4：工具参数格式错了

LLM 生成 JSON 不是 100% 可靠的。有时多了一个逗号，有时字符串没用双引号。好在 Python 的 json.loads 不够宽容，但我们可以加一层防御：

defsafe_json_parse(raw):"""更宽容的 JSON 解析"""import retry:return json.loads(raw)except json.JSONDecodeError:# 尝试修复常见问题：单引号、末尾逗号        cleaned = raw.replace("'", '"')        cleaned = re.sub(r",\s*}", "}", cleaned)try:return json.loads(cleaned)except:return {"_parse_error": f"无法解析参数: {raw[:200]}"}

把 _parse_error 作为工具返回，LLM 会意识到自己输出格式错了，下次改正。这就是自我修正闭环。

把这些组合起来

下面是一个更健壮的循环结构，集成了上面所有的错误处理：

defagent_loop(user_input):    messages.append({"role": "user", "content": user_input})for turn in range(MAX_TURNS):        msg = llm(messages)ifnot msg.get("tool_calls"):            messages.append(msg)return msg["content"]        messages.append(msg)for tc in msg["tool_calls"]:            name = tc["function"]["name"]            args = safe_json_parse(tc["function"]["arguments"])if"_parse_error"in args:                result = args["_parse_error"]else:                result = call_tool_with_timeout(name, args)            result = truncate_result(result)            messages.append({"role": "tool","tool_call_id": tc["id"],"content": result,            })return"问题太复杂了，我可能需要分成多步来回答。请简化问题。"

问题 5：把流式输出加上

如果你用到现在，可能会发现一个问题：Agent 调用工具时，屏幕上没有任何反馈，然后突然整段回答一起冒出来。这在终端里还能忍，但如果要做网页应用或者 API 服务，用户会以为程序卡死了。

解决方案：流式输出（Streaming）。在工具调用阶段维持原状，只在 LLM 最终回答时逐 token 输出。

import jsondefstream_llm(messages):"""流式调用 LLM，逐步产出最终回答的 token"""    body = json.dumps({"model": "deepseek-chat","messages": messages,"tools": TOOLS,"stream": True,              # ← 开启流式    }).encode()    req = Request(f"{BASE}/chat/completions", data=body,                  headers={"Authorization": f"Bearer {API_KEY}","Content-Type": "application/json"})    resp = urlopen(req)    collected = ""for raw_line in resp:        line = raw_line.decode().strip()ifnot line.startswith("data: ") or line == "data: [DONE]":continue        chunk = json.loads(line[6:])        delta = chunk["choices"][0]["delta"]if delta.get("content"):            collected += delta["content"]            print(delta["content"], end="", flush=True)  # 逐字输出    print()return collected

在 ReAct 循环中，当模型直接回答时（没有 tool_calls）改为调用 stream_llm() 而不是 llm()，用户就能看到打字机效果了。

ifnot msg.get("tool_calls"):    print("Agent: ", end="")    answer = stream_llm(messages)    messages.append({"role": "assistant", "content": answer})return answer

💡 避坑贴士：这里展示的“双重请求流式方案”仅用来以最少代码向大家展示打字机原理。在实际开发中，先通过非流式 API 确认没有工具调用、再请求流式 API 重写一次，会导致响应延迟和 Token 费用直接翻倍，且前后生成的回答可能不一致。 生产环境的正确做法是：直接发起单次流式请求，并在单次 Chunks 接收中同时解析流式文本和工具调用（根据 index 拼接参数）。由于高阶拼接代码逻辑稍显繁复，为了文章阅读体验不在此贴出，我们已在 GitHub 配套源码 robust_agent.py 中实现了这一生产级单请求流式解析器，非常建议大家运行对比。

核心原则：工具调用阶段不流式（因为需要等结果），最终回答阶段流式（改善体验）。这叫「先苦后甜」策略。

这个版本加上流式输出，作为个人助理的基本盘已经够用了。

一条经验法则

在 Agent 的错误处理上，我有一条经验：

工具返回的每个错误，都是 LLM 的一次学习机会。

你给它的错误信息质量，决定了它下次犯错的可能性。不要说"Error: 500"，要说"天气 API 暂时不可用（HTTP 500），可能是服务端临时故障，建议重试或换个城市名试试"。

LLM 读到这种信息，不仅能处理当前错误，还会在后续推理中更谨慎地选择调用时机。

你自己试试

1. 故意让一个工具超时——把 call_tool_with_timeout 的 timeout=10 改成 timeout=0.1，然后问一个有网络延迟的问题
2. 给 Agent 发一个很大的 JSON，触发 truncate_result，看看 LLM 如何应对"数据被截断"
3. 把 safe_json_parse 的容错逻辑去掉（直接 json.loads），然后发一个容易让 LLM 输出错误 JSON 的问题，观察 Agent 怎么崩

下一篇：「让 Agent 自己规划任务——从被动执行到主动拆解」