LangGraph StateGraph 完整详解
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"])七、适用场景优势
- 支持循环多轮 Agent（工具调用、反思校验）；
- 支持分支判断、多路径汇聚；
- 每个节点独立可拆分、调试、替换；
- 全流程状态可观测、可持久化、支持断点续跑；
- 完美适配 RAG、多工具智能体、反思型问答复杂业务。