LangGraph 四种核心状态详解

发布时间:2026/8/31 23:47:06
LangGraph 四种核心状态详解 LangGraph 四种核心状态详解LangGraph 是 LangChain 生态中用于构建有状态、多步骤工作流的框架其核心是StateGraph状态图。StateGraph 由四个核心概念组成这四个概念协同工作构建出强大且灵活的 Agent 工作流。四种核心状态LangGraph 的 StateGraph 核心四件套包括State状态共享的数据结构是图的“血液”Node节点执行逻辑的函数是图的“背肌”Edge边节点之间的连接决定流程走向Conditional Edge条件边动态路由函数是 Chain 与 Graph 的关键区别1. State状态定义一个共享的 TypedDict所有节点都可以读取和写入它。State 代表了应用程序的当前快照。作用存储图执行过程中的所有数据作为节点间通信的桥梁支持状态持久化和恢复特点所有节点共享同一个 State节点不需要显式传递参数直接从 State 中获取支持 reducer 函数控制状态更新逻辑代码示例fromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassGraphState(TypedDict):messages:Annotated[list,add]# 使用 reducer 自动追加消息context:strresult:str2. Node节点定义一个 Python 函数接收当前 State 作为输入执行计算或副作用返回 State 的部分更新。作用封装具体的业务逻辑处理数据并更新状态可以是 LLM 调用、工具使用或普通计算特点节点函数是纯函数不直接修改原 State只返回需要更新的字段LangGraph 自动合并支持同步和异步函数代码示例defprocess_document(state:GraphState)-dict:处理文档的节点# 从 State 中获取数据messagesstate[messages]# 执行业务逻辑processed_result处理后的文档内容# 返回部分更新return{result:processed_result}3. Edge边定义节点之间的连接决定流程的走向。可以是固定连接或条件连接。作用定义节点的执行顺序控制工作流的流程决定何时结束执行类型普通边固定的节点连接A → B条件边根据 State 动态决定下一个节点代码示例fromlanggraph.graphimportStateGraph,START,END# 创建状态图graphStateGraph(GraphState)# 添加节点graph.add_node(process,process_document)graph.add_node(analyze,analyze_result)# 添加普通边graph.add_edge(START,process)# 起始边graph.add_edge(process,analyze)# 节点间边graph.add_edge(analyze,END)# 结束边4. Conditional Edge条件边定义一个路由函数读取当前 State决定下一个执行哪个节点。这是 Chain 与 Graph 拉开差距的关键结构。作用实现动态分支逻辑支持运行时决策让 LLM 能够决定流程走向真正的 Agent 行为特点路由函数返回节点名称字符串支持多分支路由路径在运行时决定而非定义时硬编码代码示例defroute_based_on_confidence(state:GraphState)-str:根据置信度路由confidencestate.get(confidence,0)ifconfidence0.8:returnauto_approve# 高置信度自动审批elifconfidence0.5:returnhuman_review# 中等置信度人工审核else:returnreject# 低置信度拒绝# 添加条件边graph.add_conditional_edges(analyze,# 源节点route_based_on_confidence,# 路由函数{auto_approve:save_document,human_review:human_approval,reject:notify_user})完整示例文档处理工作流fromtypingimportTypedDict,Annotatedfromoperatorimportaddfromlanggraph.graphimportStateGraph,START,END# 定义状态classDocumentState(TypedDict):documents:Annotated[list,add]processed_docs:listconfidence:floatstatus:str# 定义节点defextract_info(state:DocumentState)-dict:提取文档信息docsstate[documents]processed[f处理:{doc}fordocindocs]return{processed_docs:processed}defcheck_quality(state:DocumentState)-dict:检查质量并计算置信度# 模拟质量检查confidence0.85return{confidence:confidence}defroute_decision(state:DocumentState)-str:路由决策ifstate[confidence]0.8:returnsaveelse:returnreview# 构建状态图graphStateGraph(DocumentState)# 添加节点graph.add_node(extract,extract_info)graph.add_node(quality_check,check_quality)graph.add_node(save,lambdas:{status:saved})graph.add_node(review,lambdas:{status:under_review})# 添加边graph.add_edge(START,extract)graph.add_edge(extract,quality_check)# 添加条件边graph.add_conditional_edges(quality_check,route_decision,{save:save,review:review})graph.add_edge(save,END)graph.add_edge(review,END)# 编译图compiled_graphgraph.compile()# 执行resultcompiled_graph.invoke({documents:[文档1,文档2],processed_docs:[],confidence:0.0,status:})print(result)总结LangGraph 的四种核心状态概念相互配合State提供共享数据空间Node封装具体业务逻辑Edge定义固定流程Conditional Edge实现动态路由这种设计使得 LangGraph 能够构建复杂、灵活且可维护的 Agent 工作流特别适合需要循环、分支、持久化和人工干预的生产级应用场景。关键优势状态管理清晰避免参数传递混乱支持动态决策真正实现 Agent 智能易于扩展和维护支持持久化和故障恢复