LangGraph：概述与安装

线性 Chain 能表达「先 A 后 B」；真实 Agent 常是：调工具 → 失败 → 再让 LLM 重试 → 某步等人点批准 → 进程重启后接着跑。这些都需要环、共享状态和持久化——LangGraph 用有向状态图（节点 + 边 + State）把控制流写进代码，而不是藏在 prompt 或手写 while True 里。

段末注释：LangGraph 构建在 LangChain 消息/模型抽象之上，负责编排层；ChatModel 参数、Prompt 细节见 LangChain 概述。

1. 一句话定位

维度	内容
角色	编排层：StateGraph、条件路由、检查点、中断与恢复
输入 → 输出	`invoke(initial_state, config)` → 最终 `State` 快照
核心 API	`StateGraph`、`add_node`、`add_edge`、`add_conditional_edges`、`compile`
依赖 LangChain	节点内常用 `BaseMessage`、`ChatModel`；图结构本身不替代模型调用

出现背景：LangGraph 约 2024 年由 LangChain 团队发布，补「AgentExecutor 黑盒循环 + 难持久化」的短板；与 LangChain v1 并列维护。本目录为可跟练的 API 专篇；跨框架编排选型见 Agent-编排-主流框架综述与选型；本仓库落地见 Agent-10-09。

2. LangGraph vs LangChain vs 手写循环

2.1 与线性 Chain / LCEL 的差异

维度	线性 Chain / LCEL	LangGraph
控制流	多为单向管道	任意图：分支、合并、自环
状态	常隐式在上下文对象里	显式 `State`，节点读写契约清晰
持久化	需自行拼	checkpointer + `thread_id`
人在回路	要自己阻塞与恢复	interrupt、从检查点 resume

若业务只是「一次 RAG 问答」，用轻量 Chain 或 LlamaIndex QueryEngine 即可；若需要「多轮工具 + 失败重试 + 人工审批某一步」，LangGraph 更贴切。

2.2 与其他实现方式

方式	控制流可见性	持久化 / HITL	适用
手写 `while` + OpenAI SDK	中	需自研	极简原型
LangChain LCEL / `create_agent`	中	有限	线性或短循环 Agent
LangGraph	高（图即文档）	checkpointer 一等公民	多步工具、审批、可恢复长任务

LangGraph 图
  ├─ 节点 A：调用 LangChain ChatModel
  ├─ 条件边：router(state) → "tool" | "end"
  ├─ 节点 B：执行 Tool → 写回 messages
  └─ 边：B → A（形成环，直到 router 结束）

3. 安装

3.1 要求

Python 3.10+
已装或同时安装 langchain-core（节点示例常用消息类型）

版本锚点（检索于 2026-08-18 PyPI）：

包	当时最新稳定版	备注
`langgraph`	1.2.11	依赖 `langchain-core>=1.4.7,<2`；与 `langchain` 1.3.x 配套区间为 `>=1.2.11,<1.3.0`
`langchain-core`	1.5.6	消息类型、Runnable
`langchain`	1.3.15	预构建 Agent 用 `langchain.agents.create_agent`（替代已弃用的 `create_react_agent`）

3.2 基础安装

1	pip install -U langgraph langchain-core

需要真实 LLM 的专篇会额外要求 langchain-openai 等 partner 包。

3.3 持久化扩展（按需）

包	用途
`langgraph`（内置）	`InMemorySaver`，开发/单测
`langgraph-checkpoint-sqlite`	本地 SQLite 持久化
`langgraph-checkpoint-postgres`	生产 Postgres

1 2	pip install langgraph-checkpoint-sqlite # 本地实验 # pip install langgraph-checkpoint-postgres # 生产

生产环境请使用已修复安全问题的版本（如 sqlite checkpointer ≥ 3.0.1，langgraph ≥ 1.0.10，以官方安全公告为准）。

3.4 验证

1	python -c "import langgraph; print('ok')"

4. 核心概念（读专篇前）

4.1 State（状态）

用 TypedDict（或 Pydantic）描述图内共享字段，例如 messages: list[BaseMessage]。
多节点写入同一列表字段时，需 Annotated[..., reducer]（如 add_messages）定义合并规则，否则后写覆盖先写。

4.2 节点（Node）

Python 函数：def node(state: State) -> dict，返回部分 state 更新（不是全量 state）。

4.3 边（Edge）

固定边：add_edge("a", "b")
条件边：add_conditional_edges(source, router_fn, path_map)
- router_fn(state) 返回字符串键
- path_map 把键映射到目标节点名或 END

4.4 compile 与 invoke

builder = StateGraph(State)
... add_node / add_edge ...
graph = builder.compile(checkpointer=...)   # 可选持久化
result = graph.invoke(input_state, config)

4.5 checkpoint 与 thread_id

配置 checkpointer 后，用 config={"configurable": {"thread_id": "会话-1"}} 隔离会话；同一 thread_id 的多次 invoke 可接续历史状态。

5. 最小可运行示例（无 LLM）

5.1 线性图：START → echo → END

from typing import Annotated, TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage


class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


def echo_node(state: State) -> dict:
    last = state["messages"][-1].content
    return {"messages": [AIMessage(content=f"echo: {last}")]}


builder = StateGraph(State)
builder.add_node("echo", echo_node)
builder.add_edge(START, "echo")
builder.add_edge("echo", END)

graph = builder.compile()
out = graph.invoke({"messages": [HumanMessage(content="hello")]})
print(out["messages"][-1].content)
# 预期：echo: hello

节点表

节点名	职责	读	写
`echo`	回显最后一条用户消息	`messages[-1]`	`messages`（追加 AIMessage）

边表

源	目标	类型
START	echo	固定
echo	END	固定

5.2 条件边：按关键词分支

from typing import Literal

def router(state: State) -> Literal["search", "direct"]:
    text = state["messages"][-1].content
    return "search" if "搜索" in text else "direct"


def search_node(state: State) -> dict:
    return {"messages": [AIMessage(content="[检索分支]")]}


def direct_node(state: State) -> dict:
    return {"messages": [AIMessage(content="[直接回答分支]")]}


builder = StateGraph(State)
builder.add_node("search", search_node)
builder.add_node("direct", direct_node)
builder.add_conditional_edges(
    START,
    router,
    {"search": "search", "direct": "direct"},
)
builder.add_edge("search", END)
builder.add_edge("direct", END)

graph = builder.compile()
print(graph.invoke({"messages": [HumanMessage(content="请搜索天气")]}))

要点：router 返回值必须是 path_map 的键。

5.3 checkpoint + thread_id：两轮 invoke

在 §5.1 的 echo 图上加 checkpointer（完整可运行）：

from typing import Annotated, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage


class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


def echo_node(state: State) -> dict:
    n = len(state["messages"])
    return {"messages": [AIMessage(content=f"已收到第 {n} 条用户消息")]}


builder = StateGraph(State)
builder.add_node("echo", echo_node)
builder.add_edge(START, "echo")
builder.add_edge("echo", END)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "demo-1"}}
graph.invoke({"messages": [HumanMessage(content="第一轮")]}, config)
out = graph.invoke({"messages": [HumanMessage(content="第二轮")]}, config)
print(len(out["messages"]))  # 预期：4（两轮各 1 Human + 1 AI）

重要配置参数（compile / invoke 入门）

参数	作用与影响	参考起点	配置指导
`compile(checkpointer=...)`	挂上 Saver；invoke 经 get_tuple/put	`InMemorySaver()` 开发	生产换 Postgres/Sqlite Saver
`configurable.thread_id`	Saver 第一层键，会话隔离	UUID 或 `user-{id}`	多租户必须唯一
`compile(interrupt_before=[...])`	指定节点前暂停	高风险 tool 节点名	配合 HITL 专篇
`invoke(..., stream_mode=...)`	流式事件粒度	`"updates"`	SSE 对接见专篇
`compile(debug=...)`	调试输出	开发 `True`	生产关闭

6. 一次 invoke 的生命周期（实现逻辑）

1. 传入 initial_state（字段符合 State schema）
2. 若存在 checkpointer：Saver.get_tuple(thread_id) 加载最近 checkpoint
3. 调度 START 的后继节点
4. 节点读 state → 返回 partial update
5. reducer 合并（如 add_messages）
6. 沿固定边或条件边进入下一节点；可形成环
7. 每 superstep 结束 Saver.put；到达 END 返回最终 state

图不直接读写 InMemorySaver 的 dict；只调用 get_tuple / put。thread_id 是第一层键，换 ID 等于新会话。

7. 推荐阅读顺序

配合 LangGraph-00.知识点索引。

阶段	主题	核心问题
1	StateGraph 与节点	怎么建图、怎么 invoke？
2	TypedDict 与 Reducer	多节点为何不能覆盖 messages？
3	conditional_edges	谁决定走哪条边？
4	checkpoint / thread_id	如何多轮续跑？
5	interrupt / HITL	如何等人批准？
6	stream	如何推 SSE？
7	create_react_agent（弃用）	对照旧预构建拓扑；新代码用 `create_agent`
8	Subgraph / Store	多 Agent、长期记忆

前置：建议先读 LangChain 目录的 Messages、Tool（或 Agent-10-02 原理篇），再进入本目录。

8. 优势与劣势

优势

显式控制流：图即设计文档，便于 Code Review 与合规审计。
原生循环与重试：工具失败 → 回到 LLM 再规划，无需用递归提示词 hack。
检查点一等公民：HITL、容错、多轮会话与「从某步重放」有统一抽象。
流式与部分输出：stream_mode 可与 SSE / WebSocket 对齐，利于产品化。

劣势

概念与样板偏多：State / reducer / checkpointer / interrupt 学习曲线陡于单文件 Agent。
生态绑定：与 LangChain 版本、包拆分强相关，升级需跟 release notes。
过度设计风险：简单问答硬上图会增加文件数与心智负担。
分布式与并发：单机图清晰；多机多实例需自行处理 thread_id、外部存储一致性与幂等。

9. 何时优先选 LangGraph

多工具、多步、可失败重试的 Agent（客服、运维 SOP、研究助理）。
需要 HITL：支付、删库、对外发信等高风险动作前必须人工确认。
需要 可恢复的长任务：任一步崩溃可从检查点续跑。
多分支 RAG：先路由到不同知识库 / prompt，再合并答案。
要把 LLM 应用做成可观测服务：节点级日志、指标与状态快照对齐。

何时不必用

单次 RAG 问答、无工具环、无审批 → LangChain Chain 或 LlamaIndex QueryEngine 更简单。
极短脚本验证 prompt → 裸 SDK 即可。
跨小时、跨服务的长事务 → 可能还需 Temporal 等外层编排（LangGraph 管单进程内图）。

10. 落地建议

先画状态与边，再写节点；State 字段越少越好，避免「万能字典」。
路由函数保持纯：尽量只读 state 与轻量规则；重逻辑放独立节点，便于单测。
生产持久化优先选官方支持的 Saver；InMemorySaver 仅用于开发与单测。
节点内仍调用 LangChain 的 ChatModel、Tool、Retriever；LangGraph 管何时调用、状态如何累积、如何持久化与中断。

11. 易踩坑

列表字段无 reducer：第二个节点写 messages 会盖掉第一个。
router 返回值未出现在 path_map：运行时报路由错误。
生产用 InMemorySaver：进程退出状态即失。
换 thread_id 却期待历史：新 ID = 新会话。
与 LangChain 版本不兼容：同环境升级 langgraph 与 langchain-core。

小结

LangGraph = 显式状态图 + 可选 checkpoint，解决环、重试、HITL、恢复。
最小路径：StateGraph → add_node → add_edge → compile → invoke。
持久化记得 checkpointer + thread_id。
节点内调 LangChain；图结构在本目录专篇深入。
跨框架选型见 Agent开发目录编排综述；工程落地见 Agent-10-09。