夜雨聆风系列：「从零写一个 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 读到这种信息，不仅能处理当前错误，还会在后续推理中更谨慎地选择调用时机。
你自己试试
故意让一个工具超时——把 call_tool_with_timeout的timeout=10改成timeout=0.1，然后问一个有网络延迟的问题给 Agent 发一个很大的 JSON，触发 truncate_result，看看 LLM 如何应对"数据被截断"把 safe_json_parse的容错逻辑去掉（直接json.loads），然后发一个容易让 LLM 输出错误 JSON 的问题，观察 Agent 怎么崩
https://github.com/DiBuilder/ai-agent-series-demo/blob/main/03-multi-tools/robust_agent.py下一篇：「让 Agent 自己规划任务——从被动执行到主动拆解」
