LangGraph分支执行逻辑解析

读懂LangGraph 的分支执行逻辑

一个 AI 请求只有一个答案时，流程很直；一旦同一输入要走几个方向，代码就像突然遇到一组路口：选哪条路、哪些路能同时走、最后在哪里汇合。LangGraph 把这些路口变成可以运行的图。 先分清两种分支 ① 静态分支 编译图时，所有候选节点已经注册完成。运行时只是在这些固定节点中选择一条或几条。 ② 动态分支 图中只准备通用 Worker，运行时才决定创建几个任务、每个任务带什么参数。 Send 属于动态分支，适合批量处理和数量不确定的任务。 ⭐ 判断口诀： 节点集合提前知道，是静态分支；任务数量运行时才知道，是动态分支。 1️⃣ 静态分支 条件路由 下面的代码可以直接运行。 add_conditional_edges() 根据路由函数返回的标签，选择已经注册好的节点。 from typing import TypedDict, Literal from dotenv import load_dotenv from langchain.messages import HumanMessage from langchain_deepseek import ChatDeepSeek from langgraph.graph import StateGraph, START, END load_dotenv(override= True ) model = ChatDeepSeek(model= "deepseek-v4-flash" ) class ContentState (TypedDict, total= False ): topic: str content_type: str poem: str joke: str def poem_node ( state: ContentState ) -> dict : # 诗歌节点只写入 poem prompt = f"请写一首关于 {state[ 'topic' ]} 的七言绝句" return { "poem" : model.invoke([HumanMessage(prompt)]).content} def joke_node ( state: ContentState ) -> dict : # 笑话节点只写入 joke prompt = f"请写一个关于 {state[ 'topic' ]} 的笑话" return { "joke" : model.invoke([HumanMessage(prompt)]).content} def router ( state: ContentState ) -> Literal [ "poem" , "joke" ]: # 返回业务标签，不直接暴露节点名称 return "poem" if state[ "content_type" ] == "poem" else "joke" builder = StateGraph(ContentState) builder.add_node( "poem_node" , poem_node) builder.add_node( "joke_node" , joke_node) # path_map 将业务标签映射到真实节点 builder.add_conditional_edges( START, router, path_map={ "poem" : "poem_node" , "joke" : "joke_node" }, ) builder.add_edge( "poem_node" , END) builder.add_edge( "joke_node" , END) graph = builder. compile () result = graph.invoke({ "topic" : "布偶猫" , "content_type" : "poem" }) print (result[ "poem" ]) flowchart LR S([START]) --> P[poem_node] S --> J[joke_node] P --> E([END]) J --> E ⭐ 重点： 图中只有 poem_node 和 joke_node 两个候选节点，运行时不会凭空增加第三个节点。 2️⃣ 动态分支 Command 路由 Command 把“判断状态”和“跳转位置”放在一个节点里。下面是完整写法，未知类型会直接结束： 典型用法： Send （动态扇出任务/数据） Command(goto=...) （运行时跳转到下游节点） from typing import TypedDict, Literal from dotenv import load_dotenv from langchain.messages import HumanMessage from langchain_deepseek import ChatDeepSeek from langgraph.graph import StateGraph, START, END from langgraph.types import Command load_dotenv(override= True ) model = ChatDeepSeek(model= "deepseek-v4-flash" ) class ContentState (TypedDict, total= False ): topic: str content_type: str poem: str joke: str def poem_node ( state: ContentState ) -> dict : prompt = f"请写一首关于 {state[ 'topic' ]} 的七言绝句" return { "poem" : model.invoke([HumanMessage(prompt)]).content} def joke_node ( state: ContentState ) -> dict : prompt = f"请写一个关于 {state[ 'topic' ]} 的笑话" return { "joke" : model.invoke([HumanMessage(prompt)]).content} def command_router ( state: ContentState ) -> Command[ Literal [ "poem_node" , "joke_node" , END] ]: # Command 的 goto 就是运行时跳转到下游节点 if state[ "content_type" ] == "poem" : return Command(goto= "poem_node" ) if state[ "content_type" ] == "joke" : return Command(goto= "joke_node" ) return Command(goto=END) # 兜底出口 builder = StateGraph(ContentState) builder.add_node( "router" , command_router) builder.add_node( "poem_node" , poem_node) builder.add_node( "joke_node" , joke_node) builder.add_edge(START, "router" ) builder.add_edge( "poem_node" , END) builder.add_edge( "joke_node" , END) graph = builder. compile () print (graph.invoke({ "topic" : "布偶猫" , "content_type" : "poem" })) print (graph.invoke({ "topic" : "布偶猫" , "content_type" : "joke" })) flowchart LR S([START]) --> R[router] R -- poem --> P[poem_node] R -- joke --> J[joke_node] P --> E([END]) J --> E R -- 未知类型 --> E 有效任务会先经过对应 Worker，再到 END ；只有未知类型才从 Router 直接结束。 add_conditional_edges() 是“返回标签后查映射表”，属于静态分支； Command 是“函数在运行时直接指定下一站”，属于动态分支。两者选一种即可，不要给同一个 Router 再接一套普通下游边。 ⭐ 重点： Command 的“动态”指运行时决定跳转路径，目标节点本身仍然要提前注册到图中。 3️⃣ 扇入与扇出 扇出：用 Send 拆分任务 动态分支的另一种常见形式是运行时拆出多个任务。 扇出 就是“一变多”：一个 Router 把一份输入拆成多个独立任务。下面的代码从导入到调用都是完整的： from typing import TypedDict from dotenv import load_dotenv from langchain.messages import HumanMessage from langchain_deepseek import ChatDeepSeek from langgraph.graph import StateGraph, START, END from langgraph.types import Send load_dotenv(override= True ) model = ChatDeepSeek(model= "deepseek-v4-flash" ) CONTENT_TYPES = [ "poem" , "ci_poem" , "joke" ] class InputState ( TypedDict ): topic: str class WorkerState ( TypedDict ): content_type: str prompt: str class GraphState (TypedDict, total= False ): topic: str poem: str ci_poem: str joke: str def router ( state: InputState ) -> list [Send]: # 列表有几个元素，就发出几张任务单 return [ Send( "worker_node" , { "content_type" : kind, "prompt" : f"请生成关于 {state[ 'topic' ]} 的 {kind} " , }, ) for kind in CONTENT_TYPES ] def worker_node ( state: WorkerState ) -> dict : # 同一个 Worker 被多次调度，每次拿到独立参数 content = model.invoke([HumanMessage(state[ "prompt" ])]) return {state[ "content_type" ]: content.content} builder = StateGraph(GraphState, input_schema=InputState) builder.add_node( "worker_node" , worker_node) builder.add_conditional_edges( START, router, path_map=[ "worker_node" ] ) builder.add_edge( "worker_node" , END) graph = builder. compile () result = graph.invoke({ "topic" : "布偶猫" }) print (result) router 发出三张任务单，三个 worker_node 可以同时执行；Worker 不需要知道自己是第几个任务，只处理收到的 prompt 。这就是动态分支：任务数量和参数由运行时数据决定。 flowchart LR S([__start__]) -.-> W[worker_node] W --> E([__end__]) 扇入：等待多个分支汇合 扇入 就是“多变一”：多个分支完成后进入同一个汇总节点。重点只看下面两种连接方式。 完整示例 from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_core.runnables import RunnableConfig from loguru import logger class EmptyState ( TypedDict ): pass def node_a ( state: EmptyState, config: RunnableConfig ) -> EmptyState: cur_step = config[ "metadata" ][ "langgraph_step" ] logger.info( "cur_step: {}, node_a 被触发" , cur_step) return {} def node_b ( state: EmptyState, config: RunnableConfig ) -> EmptyState: cur_step = config[ "metadata" ][ "langgraph_step" ] logger.info( "cur_step: {}, node_b 被触发" , cur_step) return {} def node_c ( state: EmptyState, config: RunnableConfig ) -> EmptyState: cur_step = config[ "metadata" ][ "langgraph_step" ] logger.info( "cur_step: {}, node_c 被触发" , cur_step) return {} def node_d ( state: EmptyState, config: RunnableConfig ) -> EmptyState: cur_step = config[ "metadata" ][ "langgraph_step" ] logger.info( "cur_step: {}, node_d 被触发" , cur_step) return {} def node_e ( state: EmptyState, config: RunnableConfig ) -> EmptyState: cur_step = config[ "metadata" ][ "langgraph_step" ] logger.info( "cur_step: {}, node_e 被触发" , cur_step) return {} builder = StateGraph(state_schema=EmptyState) builder.add_node( "node_a" , node_a) builder.add_node( "node_b" , node_b) builder.add_node( "node_c" , node_c) builder.add_node( "node_d" , node_d) builder.add_node( "node_e" , node_e) builder.add_edge(START, "node_a" ) builder.add_edge( "node_a" , "node_b" ) builder.add_edge( "node_a" , "node_c" ) builder.add_edge( "node_b" , "node_d" ) # 独立触发：node_c 完成后可以触发 node_e builder.add_edge( "node_c" , "node_e" ) # 独立触发：node_d 完成后也可以触发 node_e builder.add_edge( "node_d" , "node_e" ) # 改成等待全部前置节点 builder.add_edge([ "node_c" , "node_d" ], "node_e" ) graph = builder. compile () graph.invoke({}) from IPython.display import display display(graph) 独立边的执行逻辑 这是两条独立触发条件： node_c 完成可以触发一次， node_d 完成也可以触发一次。因此 node_e 会执行两次，适合“任意一个结果到达就处理”的场景，例如先到先返回、逐条通知。 列表边的执行逻辑 完整示例中的其他节点和边都不变，只把两条独立边替换成一条列表形式的多前置边： # 删除这两条独立边 # builder.add_edge("node_c", "node_e") # builder.add_edge("node_d", "node_e") # 改成等待全部前置节点 builder.add_edge([ "node_c" , "node_d" ], "node_e" ) 这是一个同步条件： node_c 和 node_d 都完成后， node_e 只执行一次。它适合汇总多个接口结果、统一审核、拼装最终答案等场景。 独立边的日志 使用两条独立边： 2026-09-27 11:02:13.556 | INFO | __main__:node_a:14 - cur_step: 1, node_a 被触发 2026-09-27 11:02:13.559 | INFO | __main__:node_b:20 - cur_step: 2, node_b 被触发 2026-09-27 11:02:13.561 | INFO | __main__:node_c:26 - cur_step: 2, node_c 被触发 2026-09-27 11:02:13.563 | INFO | __main__:node_e:38 - cur_step: 3, node_e 被触发 2026-09-27 11:02:13.563 | INFO | __main__:node_d:32 - cur_step: 3, node_d 被触发 2026-09-27 11:02:13.566 | INFO | __main__:node_e:38 - cur_step: 4, node_e 被触发 可以看到 node_e 出现两次：第一次由 node_c 触发，第二次由 node_d 触发。 列表边的日志 使用 builder.add_edge(["node_c", "node_d"], "node_e") ： 2026-06-05 14:55:01.705 | INFO | __main__:node_a:12 - cur_step: 1, node_a 被触发 2026-06-05 14:55:01.708 | INFO | __main__:node_b:17 - cur_step: 2, node_b 被触发 2026-06-05 14:55:01.710 | INFO | __main__:node_c:22 - cur_step: 2, node_c 被触发 2026-06-05 14:55:01.713 | INFO | __main__:node_d:27 - cur_step: 3, node_d 被触发 2026-06-05 14:55:01.714 | INFO | __main__:node_e:32 - cur_step: 4, node_e 被触发 这里只有一次 node_e 日志，说明它等待两个前置节点都完成后才执行。 ⭐ 重点： 独立边是“谁先完成谁触发”，列表边是“全部完成后触发一次”。 flowchart LR A([START]) --> B[node_a] B --> C[node_b] B --> D[node_c] C --> E[node_d] D --> F[node_e] E --> F F --> G([END]) 4️⃣ 总结 defer 做最后检查 主任务完成后，可以让延迟节点负责日志和完整性检查： from typing import TypedDict from langgraph.graph import StateGraph, START, END from loguru import logger class AuditState (TypedDict, total= False ): result: str def work_node ( state: AuditState ) -> dict : return { "result" : "任务已完成" } def audit_node ( state: AuditState ) -> dict : # 不参与业务处理，只负责最后检查和记录 logger.info( "最终结果：{}" , state.get( "result" )) return {} builder = StateGraph(AuditState) builder.add_node( "work_node" , work_node) # defer=True 让审计节点延迟到流程末尾执行 builder.add_node( "audit_node" , audit_node, defer= True ) builder.add_edge(START, "work_node" ) builder.add_edge( "work_node" , END) builder.add_edge(START, "audit_node" ) graph = builder. compile () print (graph.invoke({})) 它适合统计、质量检查和日志记录。 defer 不是新的分支类型，而是一个收尾时机控制器。 日常使用场景 固定类型选择一个处理器： add_conditional_edges 或 Command 。 数量不固定的批量任务： Send 动态创建 Worker。 多来源同时查询后汇总：并行扇出，再用列表形式的多前置边扇入。 任务结束后检查和记录： defer=True 配合日志节点。 ✅ 一句话记忆： 静态分支是在固定节点中选路，动态分支是在运行时发任务；扇出负责拆开任务，扇入负责等齐结果。