资讯动态

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

发布时间:2026/10/16 3:05:17 来源:尧图企业网站定制
个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangChain Agent 进阶错误处理、重试与人工介入一、先认清 Agent 的三种失败模式二、工具失效错误要转成模型读得懂的话2.1 统一包装写一个错误处理器三、推理跑偏给图加刹车3.1 硬刹车recursion_limit3.2 软刹车在 state 里数步数3.3 检测重复调用四、人工介入让危险操作停下来等批准4.1 用 interrupt 在图里暂停4.2 更简单的替代直接不给危险工具五、把三者组合起来六、可观测性出问题时能查七、小结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:rapi.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}。请检查参数后重试或告知用户转人工。returnwrappersafe_tooldefquery_inventory(sku:str)-str:查询库存。returnapi.inventory(sku)注意不要把完整堆栈喂给模型。堆栈里全是文件路径和内部实现既浪费 token 又可能泄露信息。只给异常类型和消息就够了。三、推理跑偏给图加刹车3.1 硬刹车recursion_limitfromlanggraph.errorsimportGraphRecursionErrortry:resultagent.invoke({messages:[(user,question)]},config{recursion_limit:12},)exceptGraphRecursionError:resultNonelogger.warning(Agent 步数超限已中断,extra{q:question})生产必设。这条是底线不管下面的软刹车做得多好都要留。3.2 软刹车在 state 里数步数recursion_limit是超限就崩体验不好。更优雅的是自己计数到阈值就主动结束并给出兜底回答fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messages MAX_STEPS5classState(TypedDict):messages:Annotated[list,add_messages]steps:intdefcall_model(state:State):respllm_with_tools.invoke(state[messages])return{messages:[resp],steps:state.get(steps,0)1}defshould_continue(state:State):ifstate.get(steps,0)MAX_STEPS:returngive_up# 主动兜底laststate[messages][-1]ifgetattr(last,tool_calls,None):returntoolsreturnenddefgive_up(state:State):return{messages:[(ai,抱歉这个问题我暂时没能查清楚已为你转接人工客服。)]}graphStateGraph(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:returngive_up# 连续 3 次一模一样 → 判定死循环returncontinue四、人工介入让危险操作停下来等批准这是生产 Agent 的必备能力不是可选项。4.1 用 interrupt 在图里暂停fromlanggraph.typesimportinterrupt,Command SENSITIVE{refund,send_email,cancel_order}defcall_model(state:State):respllm_with_tools.invoke(state[messages])return{messages:[resp]}defhuman_approval(state:State):laststate[messages][-1]tool_callsgetattr(last,tool_calls,[])or[]# 收集需要审批的调用pending[cforcintool_callsifc[name]inSENSITIVE]ifnotpending:return{}# 暂停图把待审批内容抛给外部decisioninterrupt({question:以下操作需要人工确认,calls:pending,})ifdecisionapprove:return{}# 放行继续走 tools# 拒绝把结果告诉模型让它另想办法return{messages:[ToolMessage(content用户拒绝了该操作请勿重试改为向用户说明情况。,tool_call_idpending[0][id],)]}调用方这样恢复执行resultagent.invoke(inputs,configconfig)# 图停在 interrupt 处检查 __interrupt__if__interrupt__inresult:payloadresult[interrupt][0].valueprint(待审批:,payload[calls])# 人工审核通过后用 Command 恢复resultagent.invoke(Command(resumeapprove),configconfig)核心概念就两个interrupt()暂停、Command(resume...)恢复。配合 checkpointer暂停状态可以落盘哪怕服务重启、隔了一天人工审核完还能接着跑。4.2 更简单的替代直接不给危险工具如果你的场景不需要那么灵活有个更省事的做法tooldefrequest_refund(order_id:str,amount:float)-str:【只读】生成退款申请单不会真正执行退款。 调用后会返回一个申请单号交由人工在后台审核后执行。ticket_idcreate_ticket(order_id,amount)returnf已生成退款申请单{ticket_id}将在 1 个工作日内由人工审核处理。把执行降级为申请Agent 的能力边界立刻清晰了也不需要复杂的 interrupt 机制。我的建议能用降级解决的就别上 interruptinterrupt 需要状态持久化、需要前端配合、需要处理超时复杂度不低。只有真正需要同步等批准的场景才用。五、把三者组合起来defshould_continue(state:State):# 1. 步数超限ifstate.get(steps,0)MAX_STEPS:returngive_up# 2. 死循环检测ifdetect_loop(state)give_up:returngive_uplaststate[messages][-1]tool_callsgetattr(last,tool_calls,[])or[]# 3. 敏感操作 → 人工审批ifany(c[name]inSENSITIVEforcintool_calls):returnhuman# 4. 普通工具 → 直接执行iftool_calls:returntoolsreturnend四路路由超限兜底 / 死循环兜底 / 人工审批 / 正常执行。这就是把不确定性显式建模成边的意思——每种异常情况在图上都是一条看得见的路径而不是散落在 prompt 里的祈祷。六、可观测性出问题时能查config{recursion_limit:12,configurable:{thread_id:user-123}}foreventinagent.stream(inputs,configconfig,stream_modevalues):lastevent[messages][-1]ifgetattr(last,tool_calls,None):forcinlast.tool_calls:logger.info(tool_call,extra{name:c[name],args:c[args]})eliflast.typetool:logger.info(tool_result,extra{preview:last.content[:200]})必须留日志。Agent 出错时你唯一能依靠的就是它到底调了什么、返回了什么。配合 LangSmith 做追踪会更省事但哪怕只是打日志也比没有强。七、小结三种失败模式工具失效、推理跑偏、越权操作都要在图结构层面解决工具错误转成有指导性的文本返回含下一步该怎么做模型会自己重试或换策略别把堆栈喂给模型recursion_limit是硬刹车底线生产必设软刹车自己计数 兜底回答体验更好检测连续 N 次相同调用能提前发现死循环人工介入两种方案interrupt Command(resume…) 同步审批或把危险工具降级为申请工具更简单优先选路由四路超限兜底 / 死循环 / 人工审批 / 正常执行一定要留日志出问题时这是唯一的线索。下一篇讲多工具协作让多个 Agent 分工干活。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑