条件分支与循环让Agent学会思考和重试上一篇讲了状态管理。状态搞定了节点之间能共享信息了。但光有状态还不够。Agent执行任务的时候很少一条路走到底。有时候要根据中间结果决定走哪条路有时候做完了检查一下发现不满意得重来。这就需要条件分支和循环。这一篇讲条件边的进阶用法、多路分支、循环重试机制以及怎么防止Agent陷入死循环。条件边回顾上一篇简单提过条件边这里展开讲。条件边的核心是一个路由函数。这个函数接收当前状态返回一个字符串告诉LangGraph下一步走哪个节点。fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDictclassWorkState(TypedDict):question:strstatus:strdefroute_by_status(state:WorkState)-str:ifstate[status]need_search:returnsearchifstate[status]ready:returnanswerreturnEND workflowStateGraph(WorkState)workflow.add_conditional_edges(decide,route_by_status)这是最基本的用法。实际项目里条件边还有几个进阶技巧。多路分支条件边不止能二选一。一个路由函数可以返回多个不同的目标节点实现多路分支。举个实际场景。用户提问进来先做意图识别。简单问题直接回答需要查资料的去搜索需要写代码的去代码生成节点闲聊的去对话节点。fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassAgentState(TypedDict):query:strintent:stranswer:strdefclassify_node(state:AgentState)-dict:querystate[query]if代码inqueryor函数inquery:intentcodingelif最新inqueryor搜索inquery:intentsearchelif你好inquery:intentchatelse:intentdirectreturn{intent:intent}defroute_by_intent(state:AgentState)-str:returnstate[intent]workflowStateGraph(AgentState)workflow.add_node(classify,classify_node)workflow.add_node(coding,coding_node)workflow.add_node(search,search_node)workflow.add_node(chat,chat_node)workflow.add_node(direct,answer_node)workflow.add_edge(START,classify)# path_map让分支关系一目了然workflow.add_conditional_edges(classify,route_by_intent,{coding:coding,search:search,chat:chat,direct:direct},)这里用到了path_map参数。它是一个字典把路由函数返回的字符串映射到具体的节点名。加上path_map以后分支结构更清晰别人读你代码的时候一眼就能看出有几条路。我之前踩过一个坑。路由函数返回了一个节点名但那个节点忘记加到图里了。编译时不报错运行时直接抛异常报错信息还不太直观。排查了半天才发现是节点没注册。后来我养成了习惯写路由函数之前先把所有节点加好再画分支省得来回查。循环重试机制循环是LangGraph区别于链式调用的关键能力。条件边返回一个已经执行过的节点就形成了循环。最典型的场景是重试生成一个结果检查发现问题回去重新生成。LangGraph新版本提供了Command对象可以在节点内部直接控制跳转方向不用单独写路由函数。fromlanggraph.typesimportCommandfromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDictclassCodeState(TypedDict):task:strcode:strtest_result:strerror_msg:strretry_count:intmax_retries:intdefcode_gen_node(state:CodeState)-dict:taskstate[task]prev_errorstate.get(error_msg,)promptf任务:{task}ifprev_error:promptf\n上次报错:{prev_error}\n请修复# 调用模型生成代码省略调用细节codefdef solve():\n #{task}的实现return{code:code}deftest_node(state:CodeState)-Command:# 执行测试省略执行细节passedFalseerror测试未通过结果不符合预期retrystate.get(retry_count,0)1ifpassed:returnCommand(gotoEND,update{test_result:passed})ifretrystate.get(max_retries,3):returnCommand(gotoEND,update{test_result:failed})returnCommand(gotocode_gen,update{retry_count:retry,test_result:failed,error_msg:error},)workflowStateGraph(CodeState)workflow.add_node(code_gen,code_gen_node)workflow.add_node(test,test_node)workflow.add_edge(START,code_gen)workflow.add_edge(code_gen,test)appworkflow.compile()test_node用Command同时做了两件事。更新状态把重试次数和错误信息写回去。决定下一步去哪测试通过就结束没通过就回code_gen重新生成。这个写法比单独写路由函数更紧凑逻辑全在一个函数里不用跳来跳去看。如果你的路由逻辑比较复杂或者多个节点共用同一条路由规则单独写路由函数还是更合适的。两种方式可以混着用看场景选。防止死循环循环好用但有个风险。如果条件一直不满足Agent会一直转圈永远停不下来。我有一次跑一个代码修复的图忘了加重试上限。模型生成的代码一直通不过测试Agent就无限循环地改代码、测试、改代码、测试。等我反应过来已经跑了半小时token烧了不少。从那以后我写循环图的第一件事就是加重试限制。防死循环有几个办法。加重试计数器。上面代码里的retry_count和max_retries就是干这个的。超过次数强制结束哪怕结果还不完美。宁可返回一个半成品也别让Agent在那空转烧钱。设置recursion_limit。编译图的时候可以限制图的最大执行步数。resultapp.invoke({task:写一个快速排序,max_retries:3},config{recursion_limit:20},)recursion_limit是LangGraph的兜底机制。就算你的重试逻辑写错了到步数上限也会停下来。建议每次跑有循环的图都显式设一下别依赖默认值。默认25步对简单任务够用复杂任务可能不够。加超时控制。除了步数限制还可以在外层包一层超时。比如用asyncio.wait_for包裹异步调用超过指定时间就中断。这对调用外部API的节点特别有用网络卡住的时候不会一直傻等。实战代码生成到测试到修复的完整循环把前面的东西串起来做一个完整的代码自修复Agent。用户描述一个编程任务Agent生成代码自动测试测试不过就拿错误信息回去修修完再测直到通过或者达到重试上限。fromlanggraph.typesimportCommandfromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDictclassCodeFixState(TypedDict):task:strcode:strtest_output:strerror_msg:strretry_count:intmax_retries:intdefgenerate_code(state:CodeFixState)-dict:taskstate[task]errorstate.get(error_msg,)iferror:promptf之前的代码有问题\n{error}\n请重新实现{task}else:promptf实现{task}# 省略模型调用假设生成了以下代码codedef quicksort(arr):\n if len(arr) 1:\n return arr\n ...return{code:code}defrun_test(state:CodeFixState)-Command:# 省略测试执行passedFalseerrorIndexError: list index out of rangeretrystate.get(retry_count,0)1ifpassed:returnCommand(gotoEND,update{test_output:passed})ifretrystate.get(max_retries,3):returnCommand(gotoEND,update{test_output:failed,error_msg:error})returnCommand(gotogenerate_code,update{retry_count:retry,error_msg:error},)graphStateGraph(CodeFixState)graph.add_node(generate_code,generate_code)graph.add_node(run_test,run_test)graph.add_edge(START,generate_code)graph.add_edge(generate_code,run_test)appgraph.compile()resultapp.invoke({task:实现快速排序,max_retries:3,retry_count:0},config{recursion_limit:20},)print(f最终结果:{result.get(test_output)})print(f重试次数:{result.get(retry_count,0)})这个流程的关键在于每次测试失败以后错误信息会写回状态。下一轮生成代码的时候模型能看到上次哪里出了问题有针对性地修改。这就是让Agent从失败中学习的基本思路每次重试都带着上一次的经验往前走。实际效果取决于模型能力和测试质量。测试写得越具体错误信息越明确模型修好的概率越高。模糊的报错比如测试失败这种模型看了也白看。所以写好测试用例本身跟写好Agent逻辑一样重要。下一篇讲多Agent协作架构。单个Agent能力有限复杂任务需要多个Agent分工配合下一篇聊怎么把多个Agent组织起来协作完成任务。