LangChain Agent 进阶：错误处理、重试与人工介入

LangChain Agent 进阶：错误处理、重试与人工介入

上一篇的 ReAct Agent 跑通了 demo，但拿到真实业务里会立刻遇到三个问题：
工具崩了怎么办？模型陷入死循环怎么办？涉及退款、发邮件这类操作，
敢让它自己动手吗？

这篇解决这三件事。核心思路是：把不确定性显式建模成图的节点和边。

一、先认清 Agent 的三种失败模式

三种都要在图的结构层面解决，而不是在 prompt 里求模型"小心一点"。

二、工具失效：错误要转成"模型读得懂的话"

先看错误做法——让异常冒泡：

@tooldefrefund(order_id:str,amount:float)->str:"""发起退款。"""returnapi.refund(order_id,amount)# 抛异常 → 整个图崩溃

正确做法：把错误变成有指导性的返回文本。

@tooldefrefund(order_id:str,amount:float)->str:"""发起退款。order_id 为订单号，amount 为退款金额（元）。"""try:r=api.refund(order_id,amount)returnf"退款成功，流水号{r['tx_id']}"exceptOrderNotFound:returnf"订单{order_id}不存在，请核对订单号后重试。"exceptAmountExceededase:returnf"退款失败：金额{amount}超过订单实付{e.paid}。请重新确认金额。"exceptTimeoutError:return"退款接口超时，请稍后重试；若连续失败请告知用户转人工处理。"

关键在最后一句"请告知用户转人工处理"——
你不是简单地报错，而是在给模型下一条行动指令。
模型读到这段话，会自己决定是重试还是回复用户。

这比你在代码里写retry(3)灵活得多：模型能区分
"订单号错了（该改参数）“和"接口超时（该重试）”。

2.1 统一包装：写一个错误处理器

工具多了，逐个写 try 太啰嗦。可以抽个装饰器：

fromfunctoolsimportwrapsdefsafe_tool(func):@wraps(func)defwrapper(*args,**kwargs):try:returnfunc(*args,**kwargs)exceptExceptionase:# 只暴露异常类型和消息，不要把堆栈喂给模型returnf"[{func.__name__}执行失败]{type(e).__name__}:{e}。请检查参数后重试，或告知用户转人工。"returnwrapper@safe_tooldefquery_inventory(sku:str)->str:"""查询库存。"""returnapi.inventory(sku)

注意：不要把完整堆栈喂给模型。堆栈里全是文件路径和内部实现，
既浪费 token 又可能泄露信息。只给异常类型和消息就够了。

三、推理跑偏：给图加"刹车"

3.1 硬刹车：recursion_limit

fromlanggraph.errorsimportGraphRecursionErrortry:result=agent.invoke({"messages":[("user",question)]},config={"recursion_limit":12},)exceptGraphRecursionError:result=Nonelogger.warning("Agent 步数超限，已中断",extra={"q":question})

生产必设。这条是底线，不管下面的软刹车做得多好都要留。

3.2 软刹车：在 state 里数步数

recursion_limit是"超限就崩"，体验不好。更优雅的是自己计数，
到阈值就主动结束并给出兜底回答：

fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messages MAX_STEPS=5classState(TypedDict):messages:Annotated[list,add_messages]steps:intdefcall_model(state:State):resp=llm_with_tools.invoke(state["messages"])return{"messages":[resp],"steps":state.get("steps",0)+1}defshould_continue(state:State):ifstate.get("steps",0)>=MAX_STEPS:return"give_up"# 主动兜底last=state["messages"][-1]ifgetattr(last,"tool_calls",None):return"tools"return"end"defgive_up(state:State):return{"messages":[("ai","抱歉，这个问题我暂时没能查清楚，已为你转接人工客服。")]}graph=StateGraph(State)graph.add_node("agent",call_model)graph.add_node("tools",ToolNode(tools))graph.add_node("give_up",give_up)graph.add_edge(START,"agent")graph.add_conditional_edges("agent",should_continue,{"tools":"tools","give_up":"give_up","end":END,})graph.add_edge("tools","agent")graph.add_edge("give_up",END)

区别：硬刹车抛异常、用户看到报错；软刹车给一句得体的兜底话。
面向用户的产品一定要用后者。

3.3 检测"重复调用"

模型有时会拿着同样的参数反复调同一个工具。加个检测：

defdetect_loop(state:State):calls=[]forminstate["messages"]:forcingetattr(m,"tool_calls",[])or[]:calls.append((c["name"],str(sorted(c["args"].items()))))iflen(calls)>=3andlen(set(calls[-3:]))==1:return"give_up"# 连续 3 次一模一样 → 判定死循环return"continue"

四、人工介入：让危险操作"停下来等批准"

这是生产 Agent 的必备能力，不是可选项。

4.1 用 interrupt 在图里暂停

fromlanggraph.typesimportinterrupt,Command SENSITIVE={"refund","send_email","cancel_order"}defcall_model(state:State):resp=llm_with_tools.invoke(state["messages"])return{"messages":[resp]}defhuman_approval(state:State):last=state["messages"][-1]tool_calls=getattr(last,"tool_calls",[])or[]# 收集需要审批的调用pending=[cforcintool_callsifc["name"]inSENSITIVE]ifnotpending:return{}# 暂停图，把待审批内容抛给外部decision=interrupt({"question":"以下操作需要人工确认：","calls":pending,})ifdecision=="approve":return{}# 放行，继续走 tools# 拒绝：把结果告诉模型，让它另想办法return{"messages":[ToolMessage(content="用户拒绝了该操作，请勿重试，改为向用户说明情况。",tool_call_id=pending[0]["id"],)]}

调用方这样恢复执行：

result=agent.invoke(inputs,config=config)# 图停在 interrupt 处，检查 __interrupt__if"__interrupt__"inresult:payload=result["interrupt"][0].valueprint("待审批:",payload["calls"])# 人工审核通过后，用 Command 恢复result=agent.invoke(Command(resume="approve"),config=config)

核心概念就两个：interrupt()暂停、Command(resume=...)恢复。
配合 checkpointer，暂停状态可以落盘，
哪怕服务重启、隔了一天，人工审核完还能接着跑。

4.2 更简单的替代：直接不给危险工具

如果你的场景不需要那么灵活，有个更省事的做法：

@tooldefrequest_refund(order_id:str,amount:float)->str:"""【只读】生成退款申请单，不会真正执行退款。 调用后会返回一个申请单号，交由人工在后台审核后执行。"""ticket_id=create_ticket(order_id,amount)returnf"已生成退款申请单{ticket_id}，将在 1 个工作日内由人工审核处理。"

把"执行"降级为"申请"，Agent 的能力边界立刻清晰了，
也不需要复杂的 interrupt 机制。

我的建议：能用降级解决的就别上 interrupt，
interrupt 需要状态持久化、需要前端配合、需要处理超时，复杂度不低。
只有真正需要"同步等批准"的场景才用。

五、把三者组合起来

defshould_continue(state:State):# 1. 步数超限ifstate.get("steps",0)>=MAX_STEPS:return"give_up"# 2. 死循环检测ifdetect_loop(state)=="give_up":return"give_up"last=state["messages"][-1]tool_calls=getattr(last,"tool_calls",[])or[]# 3. 敏感操作 → 人工审批ifany(c["name"]inSENSITIVEforcintool_calls):return"human"# 4. 普通工具 → 直接执行iftool_calls:return"tools"return"end"

四路路由：超限兜底 / 死循环兜底 / 人工审批 / 正常执行。
这就是"把不确定性显式建模成边"的意思——
每种异常情况在图上都是一条看得见的路径，而不是散落在 prompt 里的祈祷。

六、可观测性：出问题时能查

config={"recursion_limit":12,"configurable":{"thread_id":"user-123"}}foreventinagent.stream(inputs,config=config,stream_mode="values"):last=event["messages"][-1]ifgetattr(last,"tool_calls",None):forcinlast.tool_calls:logger.info("tool_call",extra={"name":c["name"],"args":c["args"]})eliflast.type=="tool":logger.info("tool_result",extra={"preview":last.content[:200]})

必须留日志。Agent 出错时你唯一能依靠的就是"它到底调了什么、返回了什么"。
配合 LangSmith 做追踪会更省事，但哪怕只是打日志也比没有强。

七、小结

• 三种失败模式：工具失效、推理跑偏、越权操作，都要在图结构层面解决；
• 工具错误转成有指导性的文本返回（含"下一步该怎么做"），
  模型会自己重试或换策略；别把堆栈喂给模型；
• recursion_limit是硬刹车底线，生产必设；软刹车（自己计数 + 兜底回答）体验更好；
• 检测"连续 N 次相同调用"能提前发现死循环；
• 人工介入两种方案：interrupt + Command(resume=…) 同步审批，
  或把危险工具降级为"申请"工具（更简单，优先选）；
• 路由四路：超限兜底 / 死循环 / 人工审批 / 正常执行；
• 一定要留日志，出问题时这是唯一的线索。

下一篇讲多工具协作：让多个 Agent 分工干活。