接触LangGraph一段时间后我发现大多数人包括我自己一开始最容易绕晕的不是节点怎么写、状态怎么传而是图结构里的分支到底怎么控制。尤其当Agent同时面对“调用工具”“直接回答”“找人工介入”“换个模型再试”这几种选择时整个图的执行路径就像城市晚高峰的路口——看起来每条路都通但如果没有一套顺手的调度规则流程跑着跑着就不知道拐到哪个死胡同里去了。LangGraph的分支执行逻辑核心说穿了就两件事条件边conditional edge决定“下一步去哪”状态state决定“凭什么这么选”。而这两件事的组合方式几乎覆盖了Agent开发里所有动态决策场景从工具调用的循环、多角色协作到批量并行处理都是这套机制在不同纬度的体现。这篇内容我会从最简单的基础路由讲起一直拆到工具调用循环和真正并行的Send机制配合能直接跑的示例代码和踩坑经验。适合正在用LangChain/LangGraph做Agent开发、被复杂流程搞得一头雾水的人也适合还没开始用LangGraph、但是已经受不了纯链式调用硬编码流程的人。1. 为什么AI应用需要分支执行逻辑1.1 固定流程解决不了决策问题只用Chain串联Prompt所有路径在写代码那一刻就全部钉死了。比如“接收问题 - 调用搜索 - 拼装回答”这确实能跑但下一步工序永远取决于上一步的结果没有任何弹性。而真实的Agent应用核心特征是模型的输出内容会反过来决定执行路径。用户问天气你可能去调天气API用户问Excel里某列的总和你可能去写代码执行用户只是想闲聊你最好别调任何工具直接回复就行了。这些决策在运行之前是不可预知的。如果不用分支逻辑你就只能把所有可能性都写在同一条执行链里然后靠模型输出伪装成“万能节点”来硬扛。这种设计在Demo里没问题一旦工具变多、任务变复杂就会出现两种让人头疼的情况要么模型在多个候选操作之间犹豫不决输出格式一乱整个流程崩掉要么为了覆盖所有情况单个节点内部塞满了十几个if-else逻辑变得极其难维护。所以LangGraph把决策这件事提到了图的层面某个节点执行完之后要走哪条边由当前状态实时决定。图结构本身就是逻辑的一部分谁在什么条件下干什么一目了然。1.2 LangGraph把“岔路口”变成显式的图结构LangGraph里的几个核心概念很容易理解直接跟地图做类比就行。State是当前所有的上下文信息相当于车辆的实时位置和路况Node是具体干活的函数相当于某个目的地Edge是节点之间的连线相当于道路而Conditional Edge就是那个带红绿灯的岔路口车辆走到这里系统看一眼状态决定放行到哪条路上。用StateGraph构建应用本质上是在画一张“流程地图”。你最终编译出来的是一个可执行的图对象调用invoke()时它会从START进入沿着边一路执行遇到条件边就动态选择方向直到抵达END。执行逻辑的核心就是把“模型在某个步骤做什么”的决策从写死的业务代码里分离出来放到图的路由规则中显式表达。这种设计的好处是你可以清晰地回答“系统在什么状态下会调用工具”“什么状态下会结束”“什么状态下会走人工审核”这些问题。而不是对着一大段Prompt和一堆工具函数的调用逻辑猜来猜去。2. 分支执行的基本实现条件边与路由函数2.1 条件边的工作机制条件边是LangGraph里最常用、也最值得先搞懂的分支机制。它的写法很固定先有一个普通节点作为“决策的起点”然后在这个节点上挂add_conditional_edges传入一个路由函数函数读取当前状态返回一个字符串类型的路径名系统根据这个字符串找到对应的目标节点。注意这里有个关键点路由函数返回的是字符串路径名不是布尔值更不是节点对象本身。我一开始理解偏了想着能不能直接return一个node名结果发现返回值必须映射到一个由代码显式声明的路径集合里。这个设计乍看多此一举实际很实用——它相当于给每条可选路线起了一个稳定的别名代码可读性大幅提升。路由函数的第二个要点是它必须是纯函数只依赖传入的State返回路径名不要在函数里做耗时的网络请求或复杂的业务操作。因为LangGraph在生成图、调试、并行执行时都可能调用条件函数你在里面埋了个HTTP请求系统执行图的时候会莫名其妙多出一堆延迟。2.2 最简分支示例数字大小路由先看一个能直接跑通的最小示例我用的LangGraph是0.2.x版本。from typing import TypedDict, Literal from langgraph.graph import StateGraph, START, END class State(TypedDict): value: int result: str def route_value(state: State) - Literal[low, high]: return low if state[value] 10 else high def low_handler(state: State): return {result: flow branch, value{state[value]}} def high_handler(state: State): return {result: fhigh branch, value{state[value]}} builder StateGraph(State) builder.add_node(low, low_handler) builder.add_node(high, high_handler) builder.add_conditional_edges( START, route_value, { low: low, high: high, } ) graph builder.compile() print(graph.invoke({value: 3})) print(graph.invoke({value: 20}))这段代码虽然简单但包含了所有分支执行的核心元素。route_value从State里读取value返回low或high然后通过路径映射表指向对应节点。调用两次分别得到两个不同节点的输出分支逻辑完全生效。执行过程中你会发现route_value其实挂在START入口之后这意味着图一开始就面临一个岔路口。如果想先经过某个处理节点再做分支只需要在START和条件边之间加一个普通节点把add_conditional_edges挂到那个节点上即可。2.3 条件路由容易踩的两个坑第一个坑是返回的路径名必须在映射表里存在。如果你在route_value里返回了medium但映射表里只有low和high运行时会直接抛错。这种错误很隐蔽因为不是语法错误只有执行到那个分支才会暴露。所以我习惯把条件函数的返回值类型写成Literal[...]让类型检查提前拦住低级错误。第二个坑是没有兜底路径。真实业务里State里的字段可能为空也可能出现预期之外的内容。比如用户提供的数值恰好等于10你写的判断是“大于10才走high”结果等于10的情况没被覆盖路由函数就得返回一个合理分支或者专门留一个unknown分支接管异常情况。条件边虽然灵活但它要求开发者把决策规则想得非常完备否则边界状态就是你系统里的定时炸弹。3. 真实场景拆解工具调用的“循环分支”3.1 模型自决继续调用工具还是直接回答分支最简单的形态是“二选一”但Agent开发里更常见的是“循环”形态。标准工具调用Tool Calling的流程是这样的模型先决定“我需不需要调工具”如果决定调就输出一个结构化的工具调用指令脚本识别到指令后执行工具把结果返回给模型模型再看一眼结果决定继续调用下一个工具还是给出最终答案。这就是一个典型的循环分支模型每输出一轮系统都要重新判断走“继续工具调用”还是“结束”。这个判断如果写死在代码里哎呀每次增删工具都要改主流程烦不胜烦。LangGraph里的写法就清爽多了from langgraph.prebuilt import ToolNode, tools_condition class State(TypedDict): messages: Annotated[list, add_messages] def chatbot(state: State): return {messages: [llm_with_tools.invoke(state[messages])]} builder StateGraph(State) builder.add_node(chatbot, chatbot) builder.add_node(tools, ToolNode(tools)) builder.add_edge(START, chatbot) builder.add_conditional_edges( chatbot, tools_condition, { tools: tools, __end__: END, } ) builder.add_edge(tools, chatbot)这段图看起来像是绕成了一个圈chatbot节点生成结果tools_condition检查结果如果里面有tool_calls就走tools节点执行工具执行完回到chatbot继续让它生成如果结果里没有tool_calls说明模型准备直接回答用户就走__end__终止整个流程。这里面有个细节很多人疑惑tools_condition为什么返回的是字符串tools或__end__而不是布尔值和节点对象答案和我们前面讲的机制一致——条件函数返回路径名映射表再把它指向真实节点。__end__是LangGraph预置的结束标记代表流程结束这是写死的规定有点类似shell脚本里退出码为0代表正常结束。3.2 循环终止条件与最大迭代次数循环结构最怕什么死循环。模型一旦在“调用工具 - 看到结果 - 继续调用工具”之间卡住你的服务就会被白白消耗的token和延迟拖垮。我遇到过最夸张的一次模型在尝试调用同一个查询工具整整8次之后才放弃期间调用了8次API耗时超过40秒——用户早走了。LangGraph本身提供了一个recursion_limit参数限制整个图的最大执行步数。用法是result graph.invoke( {messages: [(user, 请帮我把这份数据处理一下)]}, config{recursion_limit: 10} )但这只是兜底方案超过限制会抛异常而且不会自动保存中间结果。更稳妥的做法是在图里自己实现一个计数器维护一个steps字段每次循环都自增1在路由函数里检查达到上限就直接走结束分支并临时拼一条提示消息用做超时说明。这样既不会让循环无限跑下去也不会因为异常导致整个请求失败。3.3 并行分支用Send实现多个任务真正同时跑工具调用循环解决的是“一个任务如何动态延展”的问题但还有些场景需要“同时做很多件事”。比如用户一次性问“帮我对比北京、上海、广州今天的天气”如果串行跑三遍搜索工具慢不说模型还要等三轮才拿到全部数据。LangGraph对这种需求提供了Send机制专门用于动态并行分支。Send的思路可以这么理解你要启动N份“任务副本”每份副本带着自己的参数在各自动态启动的新路径上独立运行最后汇总到同一个节点做结果合并。示例很像经典的map-reducefrom langgraph.types import Send class State(TypedDict): tasks: list[str] results: Annotated[list[str], operator.add] def start_node(state: State): return [Send(worker, {task: t}) for t in state[tasks]] def worker(state: State): return {results: [analyze(state[task])]} def reducer(state: State): return {summary: summarize(state[results])} builder StateGraph(State) builder.add_node(start, start_node) builder.add_node(worker, worker) builder.add_node(reducer, reducer) builder.add_edge(START, start) builder.add_edge(worker, reducer) builder.add_edge(reducer, END)注意这里的start_node返回的不是状态更新而是一个Send对象列表。Send(worker, {...})表示“启动一个执行到worker节点的新分支并使用后面的字典作为该分支的初始状态”。有多少个任务就会并行启动多少个worker分支。真正踩过坑之后我才意识到并行分支最麻烦的不是启动而是汇总时的状态合并。比如results字段如果用普通列表各分支计算完回到合并节点时会互相覆盖或者出现顺序错乱。解决办法就是上面代码里的Annotated[list[str], operator.add]也就是给results字段挂一个合并函数让LangGraph把所有并行分支的返回值按列表拼接的方式合并到一起。这里Annotated和operator.add的组合正是状态Reducer机制的基础用法。4. 分支越多越要管好状态传递4.1 状态合并与Reducer机制Node运行完后返回一个字典这个字典会被合并到全局State里。合并的逻辑表面上看很简单同一个键就用新的覆盖旧的。可一旦你的图里有并行分支“覆盖”这个规则就会出问题。多个分支同时返回同一个键谁最后写到State里取决于谁先结束完全不可控。Reducer的作用就是让你自定义“同一个键的值冲突时怎么办”。LangGraph里最常用的Reducer有两个。一个是operator.add适合数值叠加和列表拼接另一个是LangChain生态里的add_messages专门用于合并消息列表它会按消息ID去重并保持顺序。我给Agent状态里的messages字段挂add_messages这样并行调用工具时返回的多轮消息不会因为覆盖而丢失。判断一个字段是否需要Reducer有个简单标准如果这个字段可能被多个分支写入且写入结果希望全部保留就挂Reducer。只让单个节点独占写入的字段普通覆盖就够了。无脑给所有字段都加Reducer反而会让状态更新变得不容易预测。4.2 分支节点内异常与兜底分支越多某一个分支崩溃导致整个流程失败的概率就越大。假设你的Agent有两个工具分支一个走天气查询一个走计算器天气服务临时超时你总不想让整个图跟着报废吧。所以每个分支节点内部要有异常兜底逻辑。我的习惯是在写节点函数时统一包一层try-except工具调用失败的返回内容直接写进消息里让模型知道自己“尝试过了但失败了”然后自主决定是换个工具还是向用户说明。这种做法等于把容错交给模型判断而不是让程序直接报错终止。图的分支逻辑依然保持简洁容错逻辑收敛在节点内部。另外LangGraph支持给节点设置重试次数在编译图时传入retry_policy比如某个分支节点最多重试3次、每次退避间隔递增。对那种偶发抖动的外部API请求重试策略比在节点内部写循环更省心因为重试状态由图框架统一管理不会污染业务State。4.3 嵌套分支与子图当你的主图有七八个分支节点每个分支节点还要分别往下走三四个小分支时整张图就成了一团乱麻。所有决策都放在一层图上不仅画出来难以阅读逻辑上也会出现交叉影响。LangGraph提供了子图的思路把一组关联紧密的节点封装成一个独立的StateGraph编译后作为“节点”嵌入到更大的主图里。外部看它只是一个平级的普通节点内部它有自己的状态流转和分支逻辑。我一般这样划分主图管“策略”——要不要工具、要不要人工、要不要结束子图管“执行细节”——比如某个复杂工具的多次调用内部如何决策。这样分支执行逻辑被明确分层条件边数量被控制在一定范围内排查问题的时候可以快速定位是在策略层出错还是执行层出错。5. 分支执行逻辑的调试与可视化5.1 让图自己把流程图画出来LangGraph最讨喜的功能之一就是图对象可以自动生成可视化布局。代码里加两行就能导出一张完整的流程图png_data graph.get_graph().draw_mermaid_png() with open(graph.png, wb) as f: f.write(png_data)这张图会把所有节点、普通边、条件边、结束标记都画清楚条件边的分支名称也会标注在线上。调试分支逻辑时我就是先跑一次:wq然后再对着这张图检查每条路径是否是我心中设想的走向比翻代码高效得多。如果模型没法连外网生成Mermaid图还可以用draw_ascii()在终端输出文字版结构图效果虽然朴素但同样直观。5.2 从日志回放每次路由走向图像是静态视图动态的调试还是要看日志。分支执行的过程可以通过stream方法逐段观察for event in graph.stream( {messages: [(user, 今天上海天气怎么样)]}, config{recursion_limit: 10}, stream_modeupdates ): print(event)stream_modeupdates会输出每一步是哪个节点更新了什么状态条件边决定走了哪条分支下一跳是哪个节点全部按时间顺序展示出来。排查“为什么模型明明说了要调工具却还是直接结束了”这类问题时这套日志几乎是唯一的线索来源。条件函数内部的打印也有另一种巧用。因为路由函数本身是纯函数你完全可以把简要的决策理由打印出来例如print(fstate value{value}, route to {path}),这样回放日志时每个岔路口的决策依据和结果都能对上号。5.3 常见问题速查表我把实际调试过程中遇到频率最高的分支执行问题整理成一个速查表。覆盖的场景不多但每个都是真实踩过的坑问题现象可能原因解决方式条件边执行时报错说返回的路径不在映射表里路由函数返回了一个未被声明的字符串把所有路径名抽成常量用映射表统一字典给返回类型加上Literal标注。检查大小写和空格。并行分支结束后结果字段只有最后一个分支的值状态字段没有挂Reducer多个分支互相覆盖给需要合并的字段挂operator.add或add_messagesAgent不停调用同一个工具始终不结束模型的工具调用结果一直符合“继续调工具”的路由条件加循环计数器到达阈值强制走结束分支检查路由函数对非工具决策的结束判断逻辑条件边返回结果正确但流程跳到了预期外的节点映射表里的路径名和目标节点名不一致用draw_mermaid_png导出图对照映射表检查每条箭头线启动图时报“遇到过大的递归限制”图里有循环分支但没有可靠的终止条件检查所有循环路径的路由函数务必有收敛到__end__的出口并行分支里某个任务失败整个图中断节点内缺少异常处理在节点内部包try-except失败信息写入状态让模型或后续节点接管处理6. 从分支执行到完整Agent的实际经验6.1 不是所有逻辑都要画成图分支学清楚了图分支的写法最容易犯的另一个错误是“到处都用条件边”。实际上很多决策放在节点内部用Python的if-else就够了。比如模型生成的结果里是否需要清洗一下文本——这种小逻辑塞进条件边完全是在增加路由函数的复杂度没有任何收益。我的判断标准是如果决策只影响当前节点内部怎么处理数据写在函数里如果决策会改变下一步执行哪个节点才用条件边。图分支的价值在于改变控制流而不在于做数据变换。数据变换逻辑放节点内控制流决策放边上面图的复杂度和代码可读性都能保持在一个健康区间。6.2 在实际Agent里管理工具调用的分支策略把前面这些机制组合起来就是我们常说的“让AI真的下地干活”的Agent底座。我在实际项目中把整套流程包在FastAPI的服务里每一个请求过来就graph.invoke()一次底层是条件边、循环分支还是并行分支上层调用方完全不感知。工具增多之后分支策略要比“模型输出tool_calls就调用工具”稍微复杂一点。比如有些工具很贵调用需要人批准这时候条件边在“是否调用工具”之外还得再判断工具类型决定走自动执行分支还是人工审批分支。这部分就是典型的图上条件分支把工具按“需要人工介入”和“自动执行”分成两类路由函数检查工具名是否在黑名单里然后返回对应的路径名。LangGraph的条件边天然契合这种多路策略不用像以前那样在工具调用函数里写一堆模式匹配。6.3 一点自己的体会分支执行逻辑说到底是在用代码重新表达人的决策流程什么条件下做什么什么状态下终止出了问题往哪兜底。LangGraph把这一整套东西变成了图上的显式设计和状态里的合并规则而不是散落在各种节点内部几十个if-else里。我实际使用中的最大体会是先别急着写节点先把状态字段设计好。状态是整个分支执行逻辑的地基字段覆盖还是合并每个键归谁写、谁读、会不会被并行更新这些想清楚了条件边几乎不会有逻辑bug。反过来如果状态设计没想清楚加再多的分支策略也只是让系统更早地暴露问题而已。