☰
LangGraph分支执行指南:条件边与并行分支让Agent流程可控
2026/10/6 20:22:40 网站建设 项目流程

写Agent应用最让人上头的不是Prompt,而是流程控制。LLM返回的结果你没法提前完全预知,所以代码里免不了出现“如果它想查库存就走A,想查订单就走B,如果两个都想就得并行”这类逻辑。我最早是写if/else硬串的,串到第三层嵌套、工具一多的时候,已经分不清到底是Agent在跑还是我在跑迷宫。后来切到LangGraph,才把这团乱麻理成一张图。分支执行逻辑是LangGraph里最值得吃透的部分,它决定了你的Agent能不能从“玩具demo”变成真正“下地干活”的样子。这篇文章会从执行模型讲起,再到条件边的写法、并行分支的合并、工具调用场景的落地,最后给出调试分支时能用上的手段,适合刚接触LangGraph被文档绕晕的读者,也适合那些已经跑通demo但流程一复杂就不知道怎么画图的人。

1. 别再手撕if/else了:分支执行解决的正是Agent流程失控

1.1 顺序执行很好写,真实Agent却到处是岔路口

先说我自己的经历。最早我用纯LangChain写Agent,结构大概是这样的:接住用户问题,调LLM,LLM说要调用工具我就调,工具返回结果再喂给LLM,直到它说“我回答你”。这个循环用while True也能写,问题出在你想加“如果用户只是想闲聊就直接回答”、加“意图置信度低要反问确认”、加“工具出错要换一条重试路径”这些分支时,循环体开始膨胀,每个判断都跟全局状态纠缠在一起。

比如你加了一个“紧急订单优先处理”的判断,这个判断可能影响后续三个工具的选择。在命令式代码里,要嘛把这个判断结果作为参数传来传去,要嘛搞个全局变量被各处读取。改一处,后面全得跟着改。这不是代码风格问题,是控制流和状态没有分离开。LangGraph对这种问题的回答是:把分支逻辑从业务代码里拆出去,变成图上的一条“路”。

1.2 分支不是if/else的封装,而是一张可以运行的结构图

给还没接触过的朋友一句话介绍:LangGraph是一个基于LangChain的状态图框架,你用节点(Node)表示“要做的一件事”,用边(Edge)表示“做完这件事下一步去哪”。一个节点就是一个普通Python函数,接收当前状态(State),返回状态的部分更新。边带有执行方向,其中条件边(Conditional Edge)会在运行时读取状态,由路由函数决定下一步流向哪个节点。

这条路不是写死在代码顺序里的,而是跟图本身一起被框架管理和调度。你可以在编译之后调get_graph(),直接把整个路的形状打印出来看。这是手写if/else永远给不了的东西——因为你写的判断逻辑散落在各个函数里,没有一个人能在不读完所有代码的情况下回答“这个Agent到底有几条路可以走”。

1.3 分支的两种形态:静态扇出与动态路由

我当时读文档容易在这两个词里打转,现在一句话就能分清楚。

  • 静态扇出(fan-out):从一个节点出发,不加任何判断,直接连到多个节点,LangGraph会并行跑它们。适合“需要同时做几件独立事情”的场景,比如同时查库存和查物流。
  • 动态路由:从节点出发时,执行路由函数,看状态的取值,决定去哪个分支。适合意图判断、工具调用循环这类“结果取决于运行时状态”的场景。

这两者经常叠用:先动态路由决定进哪个流程,流程里再静态扇出做并行任务。理解这两层的区别,后面看复杂图就不会晕。

2. 状态怎么流过分支:节点返回的是更新,不是赋值

2.1 节点签名与“update-only”约定

先看一个最小结构。定义状态用一个TypedDict:

from typing import TypedDict class AgentState(TypedDict): user_input: str route: str messages: list

节点就是一个函数:

def decide_route(state: AgentState) -> dict: if "下单" in state["user_input"]: return {"route": "checkout"} if "查询" in state["user_input"]: return {"route": "search"} return {"route": "chat"}

注意这里的返回值只是一个dict,它表示“我想把route改成什么”,而不是“我把整个state改成一个新对象”。节点永远不要直接改动传入的state对象再原样返回,因为LangGraph对状态的管理是基于合并(merge)的,你返回什么,框架就把它合并进状态。这样做的直接好处是:每个节点只声明自己关心的字段,其他字段天然共享。

这个约定对分支特别重要。多个分支并行执行时,每个分支的节点都拿到的是同一份“当前状态快照”,它们各自返回各自的更新。最后这些更新如何合并,取决于字段上有没有定义reducer——这一点放到第4节展开,先记住“节点是更新器”这个心法。

2.2 边的两种定义:无条件连线和“看到状态再连线”

在StateGraph里加普通边很简单:

from langgraph.graph import StateGraph, START, END builder = StateGraph(AgentState) builder.add_node("decide_route", decide_route) builder.add_edge(START, "decide_route") builder.add_edge("decide_route", END)

这表示从START进入decide_route,做完直接到END,没有分支。条件边稍微多两个参数:

builder.add_conditional_edges( "decide_route", # 从哪个节点出发 route_function, # 路由函数:读状态,返回“去向” path_map # 可选:路由函数返回值 → 目标节点名 )

这里的route_function是LangGraph里分支决策的灵魂。它接收当前state,返回一个字符串(节点名或path_map的key);如果你返回一个列表,那么这个节点出发就会动态地产生多个并行分支。这正是LangGraph图执行的精髓:流程长得像图,但走到哪一步是运行时由状态决定的。

2.3 路由函数写在哪里、什么时候执行

先说执行时机:路由函数不是在编译时跑,而是在每次执行到该节点结束后的一个超步(super-step)里跑。这句话的意思可以简单理解为:decide_route节点执行完,框架紧接着调用route_function,拿到返回值后确定下一步该调度哪些节点。如果返回列表,框架会把列表里的每个目标节点作为一个分支,在同一个超步里并行调度。

一个容易忽略的约束:路由函数应当是“所见即所得”的,不能有副作用,也不要在里面调LLM。技术上你可以这么做,但会把一次图执行的时间拉长,而且让调试变得很玄学。意图判断本身应该放在节点里完成,把结果写进state,路由函数只负责读state字段。这个分层习惯会极大降低你排查问题时的脑力负担。

3. 条件边三种写法:字符串、映射表、返回列表就是三条路

3.1 最简单:路由函数直接返回节点名

先写一个能直接跑的最小示例。假设有个decide节点先判断用户意图,路由函数再根据意图分流:

from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class State(TypedDict): user_input: str route: str def decide(state: State) -> dict: if "订单" in state["user_input"]: return {"route": "order"} return {"route": "chat"} def handle_order(state: State) -> dict: print("处理订单查询") return {} def handle_chat(state: State) -> dict: print("进入闲聊模式") return {} def router(state: State) -> str: return state["route"] builder = StateGraph(State) builder.add_node("decide", decide) builder.add_node("order", handle_order) builder.add_node("chat", handle_chat) builder.add_edge(START, "decide") builder.add_conditional_edges("decide", router) builder.add_edge("order", END) builder.add_edge("chat", END) graph = builder.compile() result = graph.invoke({"user_input": "我要查订单"})

这种写法要求router的返回值必须恰好是图里已注册的节点名。decide返回的route字段是"order",所以router返回"order"时,框架去找order这个节点。稍有不慎返回一个不存在的名字,就会抛InvalidConditionalEdgeError。新手最容易踩的就是路由值“逻辑上对了,但名字对不上”。

我的建议:节点名和路由值用一套命名规范。比如节点叫order,路由字段的值也叫order,谁也不用翻译谁。如果你倾向于给节点加_node后缀,那就统一在映射表里处理,别夹在两套命名之间反复横跳。

3.2 带path_map:路由key和节点名解耦

如果不想让路由返回值跟节点名强绑定,或者同一个路由值想去到不同节点,就加映射表:

builder.add_node("order_node", handle_order) builder.add_node("chat_node", handle_chat) builder.add_conditional_edges( "decide", router, { "order": "order_node", "chat": "chat_node", # 后面扩展路由时,在这里加一行就够了 # "refund": "refund_node", } )

这时候router只需要返回语义化的key(order、chat),具体去哪由表里的映射决定。好处是路由逻辑和图的拓扑结构解耦:你想把order改路由到新写的order_v2_node,不用动router函数,只改path_map。Agent的意图识别模型频繁迭代时,这种解耦非常值钱。

3.3 返回列表:一个节点扇出N个并行分支

第三种写法可能是你以后最常用的。路由函数返回一个字符串列表时,LangGraph会把列表里的每个字符串当作一个目标分支去并行执行。

举一个多任务并行的场景:一段文本要同时做摘要、翻译、情感分析,三个任务互不依赖。路由函数可以这样写:

def parallel_router(state: State) -> list[str]: return ["summarize_node", "translate_node", "sentiment_node"]

这样一次执行就会同时启动三个节点。顺序执行需要三倍时间,分支并行之后总耗时大致等于最慢的那一路。注意:并行分支更新state时,如果都去写同一个key且没有reducer,LangGraph在较新版本会直接报错,而不是“最后写入者胜”。这样做是为了避免静默丢状态。所以我给并行分支的每个节点安排的字段都是独立的,汇总交给下游的一个合并节点去读。

4. 并行分支的收口:状态合并与动态分发

4.1 分支终点:一块儿扎进END,还是汇到公共下游

分支跑完后,最直觉的写法是把每条路都连到END。但如果分支结果需要一个汇总节点(比如把摘要、翻译、情感结果拼成一份报告再输出),就应该让所有分支都连到同一个下游节点:

builder.add_edge("summarize_node", "aggregate_node") builder.add_edge("translate_node", "aggregate_node") builder.add_edge("sentiment_node", "aggregate_node") builder.add_edge("aggregate_node", END)

LangGraph的执行调度会保证:aggregate_node等到三个上游分支都完成之后才执行。这是图框架内置的“汇合点”语义,是手写并发逻辑时最容易漏掉的部分,在这里是免费送的。理解这一点,设计长流程时心里就有底:多个分支可以分头跑,但想汇合时,只要把边都指向同一个下游节点就行。

4.2 状态冲突:Annotated + reducer才是正解

如果你有并行分支要往同一个字段里追加内容,必须显式声明合并策略。用Annotated[list, operator.add]告诉框架:这个字段要做列表拼接,而不是覆盖。

import operator from typing import Annotated, TypedDict class ParallelState(TypedDict): subtask_results: Annotated[list, operator.add] def summarize_node(state: ParallelState) -> dict: return {"subtask_results": ["摘要完成"]} def translate_node(state: ParallelState) -> dict: return {"subtask_results": ["翻译完成"]}

没有这个reducer时,两个并行节点同时返回subtask_results,会触发InvalidUpdateError。这曾经是LangGraph社区最常见的报错之一。我的经验:凡是涉及并行分支的字段,从定义State那一刻就要想清楚“这个字段的并集语义是什么”——追加、取最大还是保留最新?在模板里就定好,不要等跑挂了再补。

4.3 动态创建分支:Send的map-reduce玩法

add_conditional_edges返回列表已经能解决“数量在运行时才能确定”的分支问题,但如果你想让分支的数量和参数都完全动态,需要Send。官方map-reduce模式大概长这样:

from langgraph.graph import Send def continue_to_review(state: ParallelState): langs = state["langs"] # 运行时拿到的语言列表 return [ Send("code_review_node", {"language": lang, "code": state["code"]}) for lang in langs ] builder.add_conditional_edges("fan_out_node", continue_to_review)

上面这个return会让框架为langs里的每一项都创建一次对code_review_node的执行,每次执行拿到独立的子状态。这种模式在批量数据处理、批量工具调用里非常实用。但注意Send不是万金油,别把简单流程强行拆成动态图,图的调试成本是真实存在的。判断标准很简单:分支数量在写代码时能不能确定?能确定就用普通条件边/列表路由,不能确定再上Send。

5. 实战:工具调用型Agent里那根“带球回传”的条件边

5.1 一条边撑起整个ReAct循环

聊完框架再回到大家最常见的场景:基于LangChain的Agent。ReAct循环其实就是一个在Agent和Tools之间来回跳的循环,而循环的开关就是条件边。标准写法的简化版本大致是这样:

import operator from typing import Annotated, TypedDict, Literal from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langgraph.graph import StateGraph, START, END # 假设 tools 已定义好,且模型绑定了这些工具 # llm_with_tools = llm.bind_tools(tools) # tool_map = {t.name: t for t in tools} class AgentState(TypedDict): messages: Annotated[list, operator.add] def agent_node(state: AgentState) -> dict: response = llm_with_tools.invoke(state["messages"]) return {"messages": [response]} def tools_node(state: AgentState) -> dict: last_message = state["messages"][-1] tool_messages = [] for tool_call in last_message.tool_calls: tool = tool_map[tool_call["name"]] tool_result = tool.invoke(tool_call["args"]) tool_messages.append( ToolMessage(content=str(tool_result), tool_call_id=tool_call["id"]) ) return {"messages": tool_messages} def should_continue(state: AgentState) -> Literal["tools_node", "end"]: last_message = state["messages"][-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: return "tools_node" return "end" builder = StateGraph(AgentState) builder.add_node("agent_node", agent_node) builder.add_node("tools_node", tools_node) builder.add_edge(START, "agent_node") builder.add_conditional_edges( "agent_node", should_continue, {"tools_node": "tools_node", "end": END}, ) builder.add_edge("tools_node", "agent_node") graph = builder.compile() result = graph.invoke( {"messages": [HumanMessage(content="帮我查一下北京的天气")]}, config={"recursion_limit": 15}, )

这里最值得品的是should_continue:它判断上一步agent_node产出的消息里,LLM是不是想要调用工具。如果AIMessage.tool_calls非空,就走tools_node执行工具;否则说明LLM已经准备生成最终回答了,走end结束。工具执行完了再回到agent_node,让LLM看到工具结果后决定下一步。一条条件边,撑起了整个“让AI真的下地干活”的循环。

我第一次看这个图时有困惑:为什么工具执行完要回到agent_node而不是直接结束?因为工具结果是给LLM看的中间产物,最终答案必须由LLM基于这些结果生成。所以tools_node -> agent_node这条回头边必不可少,它跟条件边配套,构成了循环。

5.2 一次多个tool_calls,分支在哪一层处理

LLM经常一次返回好几个工具调用。这里有两个方案。

  1. 在tools_node内部循环:一个节点把多个工具调用都执行完,返回一组ToolMessage。简单、顺序执行,适合工具内部没有相互依赖、时效性不敏感的场景。
  2. 把每个工具调用拆成并行分支:利用前面说的列表路由或Send,让多个工具并行执行。快,但要注意状态合并和工具间的数据依赖。

我实际更常先用方案1跑通,性能遇到瓶颈再改并行。原因是一个节点内循环,出问题时排查路径最短;拆成并行分支后,如果工具A的结果要被工具B使用,这种依赖图的维护成本会高不少。

5.3 死循环防护

条件边写循环很容易,忘了兜底就是死循环。LangGraph默认的recursion_limit是25,超过会抛GraphRecursionError。实际项目中我遇到过模型陷入“反复调用同一个搜索工具、拿到同样结果、再调用一次”的循环,光靠默认limit兜底不行,用户体验很差。我自己的做法有三层:

  • 设置合理的recursion_limit,比如通过config={"recursion_limit": 15}传入,别让Agent无限跑下去;
  • 在状态里加轮次计数器,节点内判断达到上限就强制生成最终答复,而不是继续循环;
  • 结合Checkpointer做人工中断:循环出现异常模式时,暂停执行,引入人工判断后再恢复。

6. 调试分支逻辑:画图、stream、以及我踩过的三个坑

6.1 先画图再写代码

LangGraph官方有LangGraph Studio这个可视化工具,确实好用,但依赖配置偶尔会折腾一下。命令行里最快的校验是draw_ascii:

print(graph.get_graph().draw_ascii())

它会直接把图的ASCII结构打印在控制台。我在写复杂分支前,习惯先把预期的图用文字画一遍,再对照代码结构。这个习惯帮我少改了很多轮代码,也让我在评审别人代码时能快速建立全局视角。

6.2 stream看每一步状态变化

想观察分支到底走了哪条路,最直观的是stream:

for chunk in graph.stream( {"user_input": "我要查订单"}, stream_mode="updates", ): for node_name, update in chunk.items(): print(f"{node_name}: {update}")

stream_mode="updates"会告诉你每个超步哪个节点跑了、返回了什么更新。配合stream_mode="values"看完整状态,基本能定位绝大多数路由问题。我调试路由函数时,甚至会在路由函数里临时打印入参state,确认走到这一步时状态字段到底是什么值——很多时候不是条件边写错,而是上游节点根本没把字段更新到预期值。

6.3 三个常见报错

报错出现原因排查方向
InvalidConditionalEdgeError路由函数返回值/列表里出现未注册的节点名对照builder.add_node注册的名字逐个检查
InvalidUpdateError并行分支同时写一个没有reducer的字段看该字段是否用Annotated[..., reducer]声明了合并策略
GraphRecursionError条件边循环没有出口检查路由函数是否永远返回同一个工具分支,增加计数器或人工中断

这三个坑我全踩过。印象最深的是第一个:当时把路由key写成了"order_node",注册节点名写的"order",名字没对上。报错信息因为在超步里包裹了一层,不仔细看根本找不到原因。后来规范成“路由键统一用状态字段的语义值、节点名统一加_node后缀、映射表兜底”之后,这类问题基本绝迹了。

6.4 状态路径越短越好,但分支是必要的复杂度

最后说一个方法论层面的体会。分支执行逻辑之所以值得认真学,是因为它给了流程设计一张可讨论的蓝图。但别忘了,每多一个分支,就是多一组状态组合要测试。我现在的设计原则是:能用顺序流就用顺序流,分支只给真正“看状态下菜”的地方用;复杂的并行依赖一定用公共汇合节点收口,不给它散养。

最后分享一个小技巧:条件边路由函数里,我习惯返回语义值并留一个默认兜底分支。意思就是,哪怕识别不出任何意图,也有一个fallback节点接住对话,不要把分支设计成“必须猜中”才走得了的路。图一旦变得复杂,先保住能走通的路径,再谈优化。这是我踩过多次空路由报错后形成的习惯,希望对你有用。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询