LangGraph StateGraph 完整详解

2026/7/5 10:34:48

LangGraph StateGraph 完整详解

一、核心概念

StateGraph是 LangGraph 最核心的图构建器，专门用来搭建带共享全局状态的 Agent 工作流。

Node（节点）：独立计算函数，只能读写统一 State；
Edge（边）：控制节点流转顺序；
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( source="build_prompt", path=route_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"])