前言

做LLM应用开发几乎都会踩这些坑：

1. 多分支、循环、并行的复杂流程，用LangChain线性Chain写出来逻辑混乱、难以维护；
2. 多轮对话上下文丢失，任务中途报错只能从头重新执行；
3. 资金操作、内容发布等高风险流程，没有人工介入审核的机制；
4. 多个节点并行更新数据，字段互相覆盖，出现数据错乱。

LangGraph 是 LangChain 官方推出的底层图编排运行时，专为有状态、长流程AI智能体设计。全文弱化冗余代码，只保留核心片段，理论通俗讲解。

一、LangGraph 基础认知

1.1 什么是 LangGraph

一句话概括：将Agent工作流建模为有向图的底层编排框架，三大核心要素：

1. Node 节点：最小计算单元，执行LLM调用、工具检索、自定义业务逻辑；
2. Edge 边：定义节点流转规则，支持顺序、并行、条件分支、循环；
3. State 状态：全局共享存储空间，全程承载所有上下文数据。

五大生产级核心能力：

1. 断点持久化：自动保存每一步状态快照，程序崩溃可从断点续跑；
2. 人机协同：任意节点暂停流程，人工审核修改数据后恢复执行；
3. 精细化状态管控：自定义数据合并规则、内外数据隔离、节点权限隔离；
4. 原生流式输出：实时推送LLM Token、节点执行日志、自定义中间信息；
5. 复杂控制流：天然支持并行、分支、循环，适配多智能体协作场景。

1.2 LangGraph vs LangChain 核心区别

二者不是替代关系，而是分层协作：LangChain是上层封装组件，LangGraph是底层执行引擎。

维度	LangChain	LangGraph
抽象层级	高层封装，开箱即用	底层运行时，细粒度完全可控
执行模型	线性顺序执行，无原生循环	基于Pregel超步，支持并行/分支/循环
状态能力	简易临时记忆，无持久化	内置全局状态+检查点持久化
故障恢复	需要自行实现存储逻辑	原生断点续跑、历史状态回溯
适用场景	简单问答、基础RAG、快速原型	多轮对话、工具循环、人工审核、多智能体

选型建议
简单固定流程用LangChain；需要循环、并行、人工审批、会话记忆的复杂Agent必须用LangGraph。

1.3 极简入门Demo（仅核心片段）

需求：用户提问，并行执行知识库检索+全网搜索，最后汇总输出答案。

# 1. 定义全局状态（TypedDict官方推荐）
class MyState(TypedDict):
    query: str
    rag_result: str
    web_search_result: str
    final_answer: str

# 2. 节点：只返回增量字段（核心规范）
def rag_search_node(state: MyState):
    return {"rag_result": "知识库检索结果"}

def web_search_node(state: MyState):
    return {"web_search_result": "全网搜索结果"}

def summary_node(state: MyState):
    ans = f"{state['rag_result']}\n{state['web_search_result']}"
    return {"final_answer": ans}

# 3. 构建图、配置流转
graph = StateGraph(MyState)
graph.add_node("rag", rag_search_node)
graph.add_node("web", web_search_node)
graph.add_node("summary", summary_node)
# START同时指向两个节点，自动并行执行
graph.add_edge(START, "rag")
graph.add_edge(START, "web")
graph.add_edge("rag", "summary")
graph.add_edge("web", "summary")
graph.add_edge("summary", END)

app = graph.compile()
# 执行流程
res = app.invoke({"query": "什么是大模型幻觉"})
print(res["final_answer"])

二、State 状态：LangGraph全局记忆大脑

State是整个图的统一数据仓库，是实现多轮对话、断点续跑的核心。

2.1 三种状态定义方案

1. TypedDict（推荐，绝大多数场景）
   带静态类型提示，IDE自动校验字段，轻量无额外依赖，适合大型Agent项目。

   from typing import TypedDict, List
   from typing_extensions import NotRequired
   class AgentState(TypedDict):
       query: str                # 必填字段
       history: List[str]        # 必填
       search_res: NotRequired[List[str]] # 可选字段
   

2. Pydantic BaseModel
   运行时强制数据校验、支持自定义校验器，适合金融、政务等高规范场景。
3. @dataclass
   自动生成初始化、打印方法，仅适合极简测试脚本。

2.2 三层数据隔离（企业开发必备）

创建StateGraph时可分别定义三类Schema，实现数据边界管控：

1. state_schema：完整全局状态，包含所有节点读写的全部字段；
2. input_schema：外部输入过滤器，仅允许传入指定字段，防止脏数据；
3. output_schema：对外输出过滤器，隐藏内部中间敏感数据，统一接口。

额外支持node_state：给单个节点定义子集状态，限制节点只能读写指定字段，实现权限隔离。

# 三层Schema初始化示例
graph = StateGraph(
    state_schema=FullState,
    input_schema=InputSchema,
    output_schema=OutputSchema
)

2.3 Reducer 状态合并规则（核心重难点）

节点返回增量数据后，由Reducer定义新旧状态的合并逻辑，解决并行更新冲突、对话历史追加问题。

1. 默认规则：覆盖
   不标注Reducer时，新值直接替换旧值，并行节点同时更新同一字段会报错。
2. 内置追加Reducer（最常用）
   • operator.add：通用列表、字符串追加；
   • add_messages：专门适配LLM对话消息列表。

   import operator
   from typing import Annotated, List
   from langgraph.graph import add_messages
   from langchain_core.messages import BaseMessage
   
   class State(TypedDict):
       search_log: Annotated[List[str], operator.add]
       messages: Annotated[List[BaseMessage], add_messages]
   

3. 自定义Reducer
   内置规则无法满足格式化、过滤需求时，自定义合并函数，入参为(旧值, 新值)。

2.4 Checkpointer 持久化、会话、断点恢复

默认每次调用invoke都会生成全新状态，无法保留上下文，Checkpointer解决该问题。

核心概念

• Checkpoint：每轮执行完成自动快照全局状态；
• thread_id：会话唯一标识，不同ID隔离独立上下文，支持多用户并发；
• 存储后端：内存（测试）、SQLite/Postgres（生产持久化）、Redis（分布式）。

核心代码片段：多轮对话记忆

from langgraph.checkpoint.memory import MemorySaver
# 初始化检查点
checkpointer = MemorySaver()
# 编译图时注入
app = graph.compile(checkpointer=checkpointer)
# 同一会话使用同一个thread_id
config = {"configurable": {"thread_id": "user_001"}}
# 多轮调用自动加载历史状态，无需手动传入上下文
app.invoke({"query": "问题1"}, config=config)
app.invoke({"query": "问题2"}, config=config)

断点恢复
使用SQLite等持久化存储，流程中途报错修复后，调用时传入None即可从断点继续执行：

# 从历史快照恢复，不传入新输入
app.invoke(None, config=config)

2.5 历史状态回溯

通过get_state_history()获取全流程每一步快照，用于调试复盘：

history = list(app.get_state_history(config))
for snapshot in history:
    print("当前状态数据：", snapshot.values)
    print("下一步待执行节点：", snapshot.next)

三、Node 节点：最小执行单元

节点本质是Python函数，遵循固定输入输出规范，同时提供缓存、重试、流式、人工中断等高级能力。

3.1 节点输入规范

可按需声明3个自动注入参数：

1. state：全局状态，第一个必传参数；
2. config: RunnableConfig：会话配置，存放thread_id、用户自定义参数；
3. runtime: Runtime：运行时上下文，提供自定义流式输出、依赖注入。

3.2 输出黄金规范（必须遵守）

只返回变更字段字典，禁止返回完整state

• 错误：return state 并行场景会出现字段冲突报错；
• 正确：return {"res": "检索结果"} 仅返回需要更新的字段。

3.3 内置特殊节点

START：流程入口，外部输入数据注入点；
END：流程终止标记，抵达后停止执行。

3.4 四大高级节点能力（核心片段）

1. 节点缓存：耗时LLM/接口调用缓存，相同输入复用结果

   from langgraph.types import CachePolicy
   graph.add_node("calc", calc_node, cache_policy=CachePolicy(ttl=10))
   

2. 自动重试：网络超时、API限流等临时故障自动重试

   from langgraph.types import RetryPolicy
   retry = RetryPolicy(max_attempts=3, retry_on=(ConnectionError, TimeoutError))
   graph.add_node("api", api_node, retry_policy=retry)
   

3. 流式输出：实时推送LLM Token、自定义中间日志

   # 实时打印大模型逐字输出
   for chunk, meta in app.stream(input, stream_mode="messages"):
       print(chunk.content, end="")
   

4. Interrupt人工中断（人机协同核心）
   流程暂停等待人工审核，必须搭配Checkpointer使用：

   from langgraph.types import Command, interrupt
   def audit_node(state):
       # 触发中断，向外抛出审核信息
       audit_data = interrupt({"金额": state["amount"]})
       return {"approve": audit_data["pass"]}
   # 恢复执行：传入人工决策
   app.invoke(Command(resume={"pass": True}), config=config)
   

四、Edge 边：控制流程流转逻辑

边分为普通边、条件边，依靠条件边实现分支、循环。

4.1 普通边 add_edge

无条件单向流转，支持多节点并行汇聚：

# 并行执行A、B，全部完成后执行C
graph.add_edge(START, "A")
graph.add_edge(START, "B")
graph.add_edge("A", "C")
graph.add_edge("B", "C")
graph.add_edge("C", END)

4.2 条件边 add_conditional_edges（分支/循环核心）

接收路由函数，根据当前状态动态选择下一个节点，实现if-else逻辑。

# 路由函数：根据数值奇偶返回分支标识
def route(state) -> Literal["even", "odd"]:
    return "even" if state["num"] % 2 == 0 else "odd"

# 绑定条件边与映射关系
graph.add_conditional_edges(
    source="node_a",
    path=route,
    path_map={"even": "even_node", "odd": "odd_node"}
)

4.3 可控循环与递归限制

条件边可构建闭环，实现LLM循环调用工具；配置recursion_limit防止死循环，超出限制抛出异常：

app.invoke(input, config={"recursion_limit": 60})

五、底层Pregel引擎通俗原理

LangGraph底层基于Google Pregel批量同步并行模型，以超步（Superstep） 为单位循环执行，每一轮超步分三阶段：

1. 规划Plan：筛选上一轮更新过数据、需要执行的节点；
2. 执行Execute：选中节点并行运行，同一轮内互相看不到对方写入的数据；
3. 更新Update：统一合并所有节点输出，更新全局状态，生成检查点快照。

核心优势：天然规避并行读写冲突，每轮执行完成自动持久化状态，为断点恢复提供底层支撑。

六、LangGraph 落地场景总结

适合使用LangGraph

1. 工具循环Agent（LLM判断是否需要联网检索，反复迭代）；
2. 人工审核流程（转账、内容发布等高风险操作）；
3. 多轮长会话机器人、多智能体分工协作；
4. 复杂RAG流水线（多路并行检索、结果反思重查）；
5. 后台异步AI任务，需要崩溃断点续跑。

优先使用LangChain

1. 流程固定的单次问答、简单基础RAG；
2. 无分支、无循环、无人工干预的极简流程。