1. 引言在构建复杂的 AI Agent 应用时仅仅调用大模型 API 往往不够我们需要一种能够精细控制执行流程、状态管理和分支逻辑的框架。LangGraph 正是为此而生。它借鉴了图计算的思想将应用逻辑抽象为一张有向图其中Node节点负责执行具体的计算或操作而Edge边则定义了节点之间的流转关系与控制流。本文将深入剖析 LangGraph 中 Node 与 Edge 的核心概念、定义方式以及它们如何协同工作帮助你从零开始掌握这一强大的编排工具。2. 为什么需要图结构在传统的链式调用Chain中流程是线性的A 完成后必然执行 B。然而现实中的 Agent 应用往往需要条件分支根据模型输出或工具结果决定下一步走向。循环当结果不满足要求时需要重试或回到上一步。并行执行多个独立任务同时进行提升效率。图结构天然支持上述复杂逻辑。LangGraph 将应用状态State在节点之间传递每个节点可以读取、修改状态而边则决定了状态流转的路径。3. Node图的基本执行单元3.1 什么是 NodeNode 是 LangGraph 图中的基本工作单元它本质上是一个函数或可调用对象接收当前状态作为输入处理后返回更新后的状态或其部分内容。一个典型的 Node 函数签名如下defmy_node(state:dict)-dict:# 读取状态current_valuestate.get(key)# 处理逻辑new_valuecurrent_value1# 返回状态更新可以是部分状态return{key:new_value}3.2 如何添加 Node在 LangGraph 中我们通过StateGraph来构建图并使用add_node方法注册节点fromlanggraph.graphimportStateGraph# 定义状态结构TypedDict 或 Pydantic 模型fromtypingimportTypedDictclassAgentState(TypedDict):messages:listnext_step:str# 创建状态图graphStateGraph(AgentState)# 定义节点函数defstart_node(state:AgentState)-dict:return{next_step:process}defprocess_node(state:AgentState)-dict:# 模拟处理return{messages:state[messages][processed]}# 注册节点graph.add_node(start,start_node)graph.add_node(process,process_node)3.3 Node 的返回值Node 返回的字典会与当前状态进行合并类似dict.update因此你只需返回需要更新的字段而不必返回整个状态。这大大简化了代码编写。4. Edge连接与流转4.1 什么是 EdgeEdge 定义了节点之间的连接关系即从一个节点执行完毕后下一个应该执行哪个节点。LangGraph 提供了多种边类型以满足不同的控制流需求。4.2 普通边Normal Edge普通边表示无条件转移源节点执行完毕后直接进入目标节点。# 添加普通边start 执行完后进入 processgraph.add_edge(start,process)4.3 条件边Conditional Edge条件边是 LangGraph 的核心特性之一它允许根据当前状态动态决定下一个节点。条件边通过一个路由函数实现defroute_after_process(state:AgentState)-str:# 根据状态决定下一步iferrorinstate[messages]:returnhandle_errorelse:returnfinish# 添加条件边graph.add_conditional_edges(process,route_after_process,{handle_error:error_handler,finish:end_node,})路由函数返回一个字符串键该键在映射字典中对应实际的目标节点名。4.4 起始边Entry Point我们需要指定图的入口节点即整个图从哪个节点开始执行graph.set_entry_point(start)4.5 结束边Finish当流程到达终点时我们需要显式地标记结束。LangGraph 使用特殊值END来表示fromlanggraph.graphimportEND graph.add_edge(end_node,END)5. Command在节点内部控制流转前面我们通过add_conditional_edges在边上做路由LangGraph 还提供了一种更灵活的方式——Command。它允许你在节点内部同时返回状态更新和下一步去向从而把「做什么」和「下一步去哪」封装在同一个返回值里。5.1 什么是 CommandCommand是一个特殊的数据结构它包含两个核心字段update要合并进状态的内容等价于普通 Node 的返回值。goto下一步要执行的节点名或END。使用Command后节点函数不再返回普通字典而是返回一个Command对象fromlanggraph.graphimportCommanddefprocess_node(state:AgentState)-Command:# 更新状态new_state{messages:state[messages][processed]}# 同时决定下一步returnCommand(updatenew_state,gotofinish)5.2 用 Command 替代条件边回到第 4 节的情感分析示例我们可以把「路由」逻辑直接放进节点内部省去单独的route_by_sentiment函数fromlanggraph.graphimportCommanddefanalyze_and_route(state:ChatState)-Command:textstate[user_input]if开心intextor棒intext:returnCommand(update{sentiment:positive},gotopositive)else:returnCommand(update{sentiment:negative},gotonegative)这样analyze节点既完成了情感分析又决定了下一步走向图结构更加紧凑。5.3 动态 gotogoto不仅可以是固定字符串还可以是动态计算的结果。例如根据外部配置或模型输出决定目标节点defdynamic_router(state:AgentState)-Command:targetstate[next_step]# 由模型或上游节点写入returnCommand(update{routed:True},gototarget)5.4 Command 与条件边的取舍5.5 子图Subgraph当图变得复杂时我们可以把一组相关的节点封装成一个子图Subgraph再作为单个节点嵌入到父图中。这样既能复用逻辑又能让整体结构保持清晰。什么是子图子图本质上就是一个独立的StateGraph它拥有自己的节点、边和状态。编译后子图可以像普通节点一样被添加到父图中fromlanggraph.graphimportStateGraph,END# 1. 构建子图sub_builderStateGraph(AgentState)sub_builder.add_node(fetch,fetch_node)sub_builder.add_node(parse,parse_node)sub_builder.set_entry_point(fetch)sub_builder.add_edge(fetch,parse)sub_builder.add_edge(parse,END)subgraphsub_builder.compile()# 2. 把子图作为节点加入父图parent_builderStateGraph(AgentState)parent_builder.add_node(preprocess,preprocess_node)parent_builder.add_node(sub,subgraph)# 子图当作普通节点parent_builder.add_node(postprocess,postprocess_node)在 Command 中 goto 子图子图在父图中就是一个普通节点因此Command的goto可以直接指向子图节点名defroute_to_subgraph(state:AgentState)-Command:# 更新状态后跳转到子图节点returnCommand(update{stage:sub},gotosub)子图与 Command 的组合同一个 State 会被更新两次当子图与父图共享同一个状态结构即使用同一个State时需要注意状态会被更新两次——一次在子图内部一次在子图返回父图时。具体来说子图作为一个节点被调用时LangGraph 会把父图的当前状态传入子图子图内部各节点依次执行并更新自己的状态副本子图执行完毕后把整个子图的状态合并回父图。因此如果子图内部某个节点更新了字段A而父图在子图之后又更新了同一个字段A那么A会被先后写入两次。看下面的例子fromlanggraph.graphimportStateGraph,ENDclassSharedState(TypedDict):count:intnote:str# 子图内部节点把 count 加 1defsub_increment(state:SharedState)-dict:return{count:state[count]1}# 父图节点把 count 再加 1defparent_increment(state:SharedState)-dict:return{count:state[count]1}# 构建子图sub_builderStateGraph(SharedState)sub_builder.add_node(inc,sub_increment)sub_builder.set_entry_point(inc)sub_builder.add_edge(inc,END)subgraphsub_builder.compile()# 构建父图parent_builderStateGraph(SharedState)parent_builder.add_node(sub,subgraph)parent_builder.add_node(parent_inc,parent_increment)parent_builder.set_entry_point(sub)parent_builder.add_edge(sub,parent_inc)parent_builder.add_edge(parent_inc,END)appparent_builder.compile()resultapp.invoke({count:0,note:})print(result[count])# 输出2子图 1父图再 1提示如果不想让子图把内部状态合并回父图可以给子图定义独立的状态结构与父图不同的TypedDict这样子图内部的状态变化就不会污染父图。只有当确实需要共享状态时才使用同一个State并留意「更新两次」带来的叠加效果。子图内部同样可以使用Command来控制自己的流转甚至可以在子图内部通过Command返回goto到父图的节点LangGraph 支持跨层级跳转# 子图内部节点处理完成后直接跳回父图的 postprocessdefsub_inner_node(state:AgentState)-Command:returnCommand(update{parsed:True},gotopostprocess)注意跨层级goto需要子图与父图共享状态结构且目标节点名在父图中存在。合理使用子图 Command可以把大型 Agent 拆分成多个可独立测试的小模块大幅提升可维护性。方式优点适用场景条件边add_conditional_edges路由逻辑集中在图定义处结构一目了然路由规则相对固定、便于统一查看Command路由与状态更新封装在节点内代码更内聚节点需要同时更新状态并动态决定去向两者可以混用你可以在部分节点用Command返回部分节点用普通字典返回LangGraph 都能正确处理。5. 完整示例一个带条件分支的 Agent下面我们构建一个简单的示例根据用户输入的情感倾向决定走「积极回复」还是「消极安抚」分支。fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,ENDclassChatState(TypedDict):user_input:strsentiment:strreply:strdefanalyze_sentiment(state:ChatState)-dict:# 模拟情感分析textstate[user_input]if开心intextor棒intext:return{sentiment:positive}else:return{sentiment:negative}defpositive_reply(state:ChatState)-dict:return{reply:太棒了很高兴听到你这么说}defnegative_reply(state:ChatState)-dict:return{reply:别难过一切都会好起来的。}defroute_by_sentiment(state:ChatState)-str:returnstate[sentiment]# 构建图builderStateGraph(ChatState)builder.add_node(analyze,analyze_sentiment)builder.add_node(positive,positive_reply)builder.add_node(negative,negative_reply)builder.set_entry_point(analyze)builder.add_conditional_edges(analyze,route_by_sentiment,{positive:positive,negative:negative})builder.add_edge(positive,END)builder.add_edge(negative,END)# 编译并运行appbuilder.compile()resultapp.invoke({user_input:今天太开心了,sentiment:,reply:})print(result[reply])# 输出太棒了很高兴听到你这么说7. 进阶流式支持Streaming在实际的 Agent 应用中我们往往希望边执行边输出而不是等整个图跑完才拿到最终结果。LangGraph 提供了流式支持让你可以实时观察节点的执行过程与状态变化非常适合构建打字机式的对话体验。7.1 流式输出节点结果最常用的方式是stream方法它会按执行顺序逐个产出每个节点的输出# 使用 stream 逐节点输出forchunkinapp.stream({user_input:今天太开心了,sentiment:,reply:}):print(chunk)输出大致如下每个节点执行完都会产出一条记录{analyze: {sentiment: positive}} {positive: {reply: 太棒了很高兴听到你这么说}}可以看到stream会在每个节点执行完毕后立即返回该节点的状态更新而不是等整张图跑完。这样你就可以把中间结果实时推送给用户。7.2 流式输出 TokenStreaming Tokens对于大模型节点我们往往希望像 ChatGPT 一样逐字输出。LangGraph 提供了stream_modemessages可以流式产出 LLM 生成的 token# 流式输出 LLM 的 tokenasyncformessage,metadatainapp.astream({user_input:今天太开心了},stream_modemessages,):# message 是 AIMessageChunkmetadata 包含节点信息ifmessage.content:print(message.content,end)说明stream_modemessages需要节点内部使用 LangChain 的模型如ChatOpenAI并返回消息对象才能逐 token 输出。普通返回字典的节点不会产生 token 流。7.3 流式更新状态Streaming Updates如果你只关心状态的变化而不关心具体是哪个节点产生的可以使用stream_modeupdates# 只流式输出状态更新forupdateinapp.stream({user_input:今天太开心了},stream_modeupdates,):print(update)它与默认模式类似但更聚焦于「状态被改成了什么」适合做日志或调试。7.4 流式模式对比模式产出内容适用场景默认values每个节点执行后的完整状态观察整体状态演进updates每个节点返回的状态增量关注状态变化、调试messagesLLM 逐 token 输出打字机式对话体验提示流式支持与Command、子图完全兼容。子图作为节点执行时其内部节点的输出同样会被流式返回你可以据此实现「子图内部也逐节点展示」的效果。6. 进阶循环与并行6.1 循环Loop通过条件边我们可以轻松实现循环。例如一个「生成-评估-重试」的循环defshould_retry(state:ChatState)-str:ifstate[retry_count]3andnotstate[is_good]:returnretryreturnfinish# 在图中添加回边builder.add_conditional_edges(evaluate,should_retry,{retry:generate,finish:END})6.2 并行ParallelLangGraph 支持扇出fan-out与扇入fan-in模式实现并行执行# 从 start 同时分发到 task_a 和 task_bbuilder.add_edge(start,task_a)builder.add_edge(start,task_b)# 两个任务都完成后汇聚到 aggregatebuilder.add_edge(task_a,aggregate)builder.add_edge(task_b,aggregate)8. 总结概念作用关键 APINode执行具体逻辑读写状态add_node(name, fn)普通边无条件流转add_edge(src, dst)条件边按状态动态路由add_conditional_edges(src, router, mapping)Command节点内同时更新状态并决定去向Command(update..., goto...)子图把一组节点封装成可复用的子图节点add_node(name, subgraph)入口指定起始节点set_entry_point(name)结束标记流程终点add_edge(node, END)Node 与 Edge 是 LangGraph 的基石。Node 定义了「做什么」Edge 定义了「下一步做什么」。而Command则把两者优雅地融合在节点内部让路由逻辑更内聚子图则进一步帮你把复杂逻辑拆分成可复用的模块。掌握条件边、循环、Command与子图的用法你就能构建出灵活、健壮的 Agent 应用。希望本文能帮助你迈出 LangGraph 实战的第一步。