写到第七篇,是时候回头盘点一下了。你翻翻前几篇的代码,会发现一个挺有意思的现象: 几乎每一篇,我们都在往代码里加 try/except 。
第二篇里,我们担心模型输出的 JSON 参数不合法,加了一层 try/except json.JSONDecodeError;第三篇里,我们担心模型陷入无休止的推理循环,加了个 max_steps 上限;第五篇的规划复盘、第六篇的审核逻辑,只要涉及解析模型输出,全都得加一层“万一解析失败怎么办”的兜底。
这些补丁东一块西一块地打,说明一件事:「智能体在真实世界里跑起来,出错是家常便饭,而不是意外情况」
工具会超时、API会限流、模型会抽风输出格式不对的东西、逻辑会绕进死胡同,这些不是“小概率的边角情况”,这是 智能体系统的日常 。
这一篇,我们不再零敲碎打地贴创可贴,而是把“错误处理”当成一个正经的系统模块,从头到尾捋一遍:故障怎么分类、什么情况该重试、什么情况该认怂、怎么防止死循环、以及怎么让智能体在交卷之前,自己先“回头看看”有没有说瞎话。
本文看点
-
- 01、故障分类:瞬时 vs 永久
- 02、重试与防死循环
- 03、交卷前的反思工序
01、故障也分三六九等:别一棍子打死
先问一个问题:如果你打电话给朋友,没人接,你会怎么办?大概率是过一会儿再打一次,电话没接通,可能只是他当时正忙, 这是个“瞬时性”的问题,稍后重试大概率就通了 。
但如果你拨的这个号码压根就是空号,你会一直重拨吗?当然不会, 这是个“永久性”的问题,重试100次结果都一样 。智能体调用工具时遇到的错误,本质上也分这两类:
瞬时错误
网络超时、API 限流、服务器临时抖动。特点是同样的请求换个时间再发一次,很可能就成功了,这样值得重试。
永久错误
参数本身就错、权限不够、资源不存在。不管重试多少次结果都一样,这样不该重试,而应把失败事实告诉模型让它换思路。
很多人写智能体系统的第一版代码,都会犯一个共同的错:不分青红皂白,捕获到异常就无脑重试,或者干脆捕获到异常就直接放弃。前者会让“参数错误”这种重试也没用的情况白白拖慢速度、浪费额度;后者会让“网络抖动一下”这种重试就好的情况直接判了死刑,任务失败得莫名其妙。
正确的做法,是先把错误分门别类,再对症下药
02、动手写一套靠谱的错误处理框架
我们用 deepseek-v4-flash 来实现。先自定义两种异常类型,用来区分“值得重试”和“不值得重试”:
import time
import json
import random
from openai import OpenAI
client = OpenAI(
api_key="sk-xxxx",
base_url="https://api.deepseek.com",
)
class TransientError(Exception):
"""瞬时错误:值得重试,比如网络超时、接口限流"""
pass
class PermanentError(Exception):
"""永久错误:重试没有意义,比如参数无效、资源不存在"""
pass
接下来写一个模拟的“不太稳定”的搜索工具,故意设计成偶尔超时、偶尔遇到无效参数,方便我们演示:
def flaky_search(query: str) -> str:
"""模拟一个不太稳定的搜索工具"""
if "不存在的城市" in query:
# 这种参数本身就有问题,重试没有意义
raise PermanentError(f"查询目标不存在:{query}")
if random.random() < 0.4:
# 40%概率模拟网络超时,这是值得重试的瞬时问题
raise TransientError("请求超时,服务暂时不可用")
return f"关于「{query}」的搜索结果:近期相关信息一切正常。"
然后是核心:一个带 指数退避(exponential backoff) 的重试包装器。所谓指数退避,就是第一次失败等1秒重试,第二次等2秒,第三次等4秒……每次失败后等待时间翻倍,既给了系统“喘口气”恢复的时间,又不会因为固定间隔死磕同一个繁忙的服务而雪上加霜:
def call_tool_with_retry(func, args: dict, max_retries: int = 3):
"""带指数退避重试的工具调用包装器"""
for attempt in range(max_retries):
try:
return func(**args)
except TransientError as e:
wait_time = (2 ** attempt) + random.random()
print(f"瞬时错误(第{attempt+1}次),{wait_time:.1f}秒后重试:{e}")
time.sleep(wait_time)
except PermanentError as e:
# 永久错误直接认怂,不浪费时间重试
print(f"永久错误,放弃重试:{e}")
return f"[工具执行失败-不可重试] {e}"
except Exception as e:
# 未知异常,保守起见也不无限重试,直接上报
print(f"未预期的异常:{e}")
return f"[工具执行失败-未知错误] {e}"
return f"[工具执行失败] 重试{max_retries}次后依然失败,已放弃"
注意这里的设计原则:不管最终是重试成功、永久失败、还是重试耗尽,这个函数永远返回一个字符串结果,而不是让异常继续往外抛。因为往外抛异常会直接让整个智能体循环崩溃,而我们更希望 把“这个工具失败了”这个事实,当成一条正常的观察结果,喂回给模型 ,让模型自己决定接下来怎么办(换个工具?换种问法?还是老实告诉用户查不到?),而不是让整个程序直接挂掉。这跟我们第二篇里讲过的道理是一脉相承的。
03、防止死循环:给智能体设一个“闹钟”
第三篇我们提到过 max_steps 这个安全阀,防止模型陷入逻辑死循环、无休止地“想了又想”。但光限制步数还不够,真实场景里还有一种情况: 每一步单独看都不算死循环,但每一步都很慢,几步加起来,整体耗时已经长到不合理了 。
所以我们还得加一层 墙钟超时(wall-clock timeout) :不管走了几步,只要总耗时超过某个上限,就直接掐断,不再继续瞎折腾:
def run_robust_agent(user_input: str, max_steps: int = 6, timeout_seconds: int = 30):
start_time = time.time()
messages = [
{"role": "system", "content": "你是一个可靠的助手,请通过搜索工具帮用户查找信息。"},
{"role": "user", "content": user_input},
]
tools_schema = [{
"type": "function",
"function": {
"name": "flaky_search",
"description": "搜索相关信息",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
},
}]
draft = None
for step in range(max_steps):
# 双重保险:既限制步数,也限制总耗时
if time.time() - start_time > timeout_seconds:
print("已超过总耗时上限,终止执行")
draft = "抱歉,处理超时,未能获得完整结果。"
break
response = client.chat.completions.create(
model="deepseek-v4-flash", messages=messages, tools=tools_schema
)
reply = response.choices[0].message
if reply.content:
print(f"[Thought {step+1}] {reply.content}")
if not reply.tool_calls:
draft = reply.content
break
messages.append(reply)
for call in reply.tool_calls:
try:
args = json.loads(call.function.arguments)
except json.JSONDecodeError:
# 模型自己生成的参数格式都不合法,这种情况也别崩溃
result = "[参数解析失败] 请重新生成合法的JSON格式参数"
else:
result = call_tool_with_retry(flaky_search, args)
print(f"🔧 [Action] flaky_search({args if 'args' in dir() else ''}) → {result}")
messages.append({
"role": "tool", "tool_call_id": call.id, "content": str(result)
})
else:
draft = "已达到最大步数上限,未能得出结论。"
return draft
这样一来,flaky_search 不管是超时重试成功、还是彻底失败,整个 run_robust_agent 函数都能扛住,不会中途崩溃;就算模型不小心陷入了“一直查一直查”的循环,也会被 max_steps 和 timeout_seconds 双重兜底掐断,而不是让你的服务器一直空转到天荒地老。
04、临交卷前,让它自己先“回头看一眼”
前面几节解决的都是“过程中出错怎么办”,但还有一种更隐蔽的问题: 过程中一切顺利,工具都执行成功了,但模型最后组织出来的答案,本身就编了一些没被验证过的内容 ,也就是我们常说的“幻觉”。这种问题不会报错、不会崩溃,看起来一切正常,却悄悄把错误信息交到了用户手上。
第六篇我们讲过用一个独立的“审核者”角色来抓这种问题,那是多智能体架构下的解法。但哪怕是单智能体场景,我们也可以给它加一道简化版的自我审查工序,业界管这个叫 反思(Reflection) :在给出最终答案之前,让模型专门检查一遍:“这个答案是不是完整回应了问题?有没有编造未经工具验证的内容?”
def reflect_on_answer(user_input: str, draft: str) -> dict:
"""反思机制:交卷前自我检查一遍,看有没有编造未经验证的信息"""
prompt = f"""请检查下面这个回答,是否完整地回应了用户的问题,
是否存在任何看起来像是凭空编造、并未经过实际工具验证的具体数据或结论。
用户问题:{user_input}
草稿回答:{draft}
请以JSON格式输出:{{"pass": true或false, "feedback": "具体说明,如果通过请说明理由"}}"""
response = client.chat.completions.create(
model="deepseek-v4-flash", messages=[{"role": "user", "content": prompt}]
)
try:
return json.loads(response.choices[0].message.content)
except json.JSONDecodeError:
return {"pass": True, "feedback": ""}
把它接到主流程末尾,交卷之前多绕这一道弯:
def run_robust_agent_with_reflection(user_input: str):
draft = run_robust_agent(user_input)
reflection = reflect_on_answer(user_input, draft)
if not reflection.get("pass"):
print(f"反思发现问题:{reflection.get('feedback')}")
# 把反思意见交回去,让模型修正后再输出
fix_prompt = f"""你刚才的回答存在以下问题:{reflection.get('feedback')}
请基于这个反馈,修正你的回答。原回答:{draft}"""
fixed = client.chat.completions.create(
model="deepseek-v4-flash",
messages=[{"role": "user", "content": fix_prompt}],
)
return fixed.choices[0].message.content
return draft
这道“回头看一眼”的工序不能完全杜绝幻觉(毕竟负责反思的还是同一类模型,跟第六篇讲过的“当局者迷”问题一样,独立性有限),但它确实能像多一双眼睛一样, 拦住一部分“顺嘴编造”的明显错误内容 ,是性价比很高的一道保险丝。
05、这一篇,我们把“补丁”变成了“体系”
回顾一下这一篇解决的问题:
1、故障要分类:瞬时错误值得重试,永久错误直接认怂;
2、重试要讲策略:指数退避,而不是傻乎乎地立刻重试;
3、双重兜底防死循环:步数上限 + 总耗时上限;
4、交卷前留一道反思工序:检查有没有信口开河。
这几件事单独拎出来看,好像都不复杂,但组合在一起,才是一个智能体系统从“能跑起来的 demo”迈向“能扛得住真实世界折腾的可靠系统”最关键的一步。
前六篇教它“变聪明”,这一篇教它“变皮实”——遇到糟心事不崩溃、不瞎编、不空转
下一步:这套东西怎么真正搬上线
到目前为止,我们写的所有代码,都还停留在“能在你自己电脑上跑通”的阶段。真要把它变成一个 7x24 小时在线、有真实用户在用的系统,还差着一大截,比如:怎么知道它每一步到底在干嘛( 可观测性 )、Token 花费怎么监控别一不小心把预算烧穿、怎么给智能体的工具调用加上权限边界防止它“越权”操作、以及怎么科学地评估“我这个智能体到底是变强了还是变弱了”。
下一篇,我们正式聊聊「从 Demo 到生产」
智能体系统真正上线前,那些容易被忽视、但绝对不能省的工程化功课。下一篇见。
END
我是写代码的中年人,长期专注于 AI 算法、智能体、以及 AI 工程化落地。
190