LangGraph StateGraph 完整详解一、核心概念StateGraph是 LangGraph 最核心的图构建器专门用来搭建带共享全局状态的 Agent 工作流。Node节点独立计算函数只能读写统一 StateEdge边控制节点流转顺序State整张图全局唯一的数据容器所有节点通信只靠它END流程终止标识。区别于基础Graph普通 Graph 无统一状态StateGraph 是业务开发标准。二、State 状态定义必须先定义状态用TypedDict/Pydantic描述规定流程中会流转哪些字段。python运行from typing import TypedDict, List # 自定义全局状态结构 class RAGState(TypedDict): question: str # 用户问题 chunks: List[str] # 检索文档片段 prompt: str # 拼接后的上下文Prompt answer: str # LLM生成回答三、StateGraph 基础初始化python运行from langgraph.graph import StateGraph, END # 传入状态类型创建图构造器 builder StateGraph(RAGState)四、核心常用方法1. add_node (name, function) 注册节点把计算函数绑定为命名节点python运行def retrieve(state: RAGState): # 读取state里的question做检索 q state[question] docs [文档1内容, 文档2内容] # 只返回需要更新的字段增量更新 return {chunks: docs} # 注册节点命名为 retrieve builder.add_node(retrieve, retrieve)2. add_edge (起点节点终点节点) 普通直线边固定顺序流转python运行builder.add_edge(retrieve, build_prompt) builder.add_edge(build_prompt, generate) builder.add_edge(generate, END) # 走到END代表流程结束3. add_conditional_edges 条件分支边根据状态动态选择下一个节点支持循环、判断、提前结束python运行def route_after_gen(state: RAGState): # 路由函数返回下一跳名称 / END if len(state[chunks]) 0: return retrieve # 无文档重新检索循环 else: return END builder.add_conditional_edges( sourcebuild_prompt, pathroute_after_gen )4. set_entry_point (node_name) 设置入口节点指定流程从哪个节点开始执行python运行builder.set_entry_point(retrieve)5. compile () 编译成可执行图定义完所有节点、边后编译得到可调用对象python运行graph builder.compile()6. invoke (初始状态字典) 运行工作流传入初始数据自动流转所有节点python运行result graph.invoke({question: 公司数据安全政策是什么}) # result 是完整更新后的 State 字典 print(result[answer])五、State 关键特性只读不覆写节点不能直接修改传入的 state 对象只能返回{字段: 新值}框架自动合并更新。增量更新节点只需返回发生变化的字段不用返回完整状态节省开销。全局共享 所有节点读取同一份状态天然实现跨节点数据传递不用手动传参。六、完整最小 RAG 示例python运行from typing import TypedDict, List from langgraph.graph import StateGraph, END # 1. 定义状态 class RAGState(TypedDict): question: str chunks: List[str] answer: str # 2. 定义节点函数 def retrieve(state: RAGState): return {chunks: [TechCorp 数据安全规范文档内容]} def generate(state: RAGState): ctx \n.join(state[chunks]) ans f基于文档回答{ctx} return {answer: ans} # 3. 构建StateGraph builder StateGraph(RAGState) # 注册节点 builder.add_node(retrieve, retrieve) builder.add_node(generate, generate) # 流转规则 builder.set_entry_point(retrieve) builder.add_edge(retrieve, generate) builder.add_edge(generate, END) # 编译执行 graph builder.compile() output graph.invoke({question: 安全政策}) print(output[answer])七、适用场景优势支持循环多轮 Agent工具调用、反思校验支持分支判断、多路径汇聚每个节点独立可拆分、调试、替换全流程状态可观测、可持久化、支持断点续跑完美适配 RAG、多工具智能体、反思型问答复杂业务。