LangGraph · 用图来编排 LLM 应用

— 00

为什么需要它

先建立心智模型：LangGraph 解决的是普通「链」解决不了的事。理解这一点，后面所有 API 都水到渠成。

普通链 / Chain

线性单向：A→B→C 一条道走到黑。
不能回头：第 3 步要补信息，跳不回第 1 步。
不能循环：Agent「想→做→看→再想」做不了。
难分支：想按中间结果二选一，得写一堆 if 拆链。
状态散落：中间数据靠手动一层层传。

VS

LangGraph / 图

图结构：节点 + 边，流转关系显式声明。
可成环：边能指回前面节点 → 天然支持循环。
条件路由：一个函数看状态，动态决定下一步。
共享状态：一个 State 贯穿全程，人人读写。
可观测/可中断：每步状态可见，能停、能恢复。

"

定义

LangGraph = 状态机 + 有向图的 LLM 编排框架。它不替你写提示词、不替你调模型，只管「多步流程怎么流转、状态怎么传、何时循环、何时分支」——给 LLM 应用用的流程图引擎。

— 01

五个核心构件

整个框架就这几样东西。每张卡左栏讲「通用概念」，右栏给「最小可运行代码」——照着就能写出来。

①

State · 共享状态

TypedDict + Reducer

概念是什么

State 是贯穿整张图的数据对象，每个节点都读它、往里写。通常用 TypedDict 声明字段。

关键在 Reducer（归并函数）：它定义「节点返回的新值，如何合并进旧状态」。默认是覆盖；聊天消息要的是追加，就给字段挂一个 reducer。

add_messages 就是官方给消息列表用的追加 reducer——新消息接到历史后面，而不是冲掉历史。

最小代码

state.pyPython

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    # 挂了 reducer：新消息“追加”，不覆盖历史
    messages: Annotated[list, add_messages]
    # 没挂 reducer 的字段 → 默认"覆盖"
    user_name: str

记法 · Annotated[类型， reducer] = 这个字段更新时按 reducer 合并

②

Node · 节点

就是一个普通函数

概念是什么

节点就是一个函数：接收当前 state，干点活（调 LLM、查库、算东西），返回「要更新哪些字段」的字典。

最易踩的点：节点不返回整个新 state，只返回变化的部分，剩下交给 reducer 合并。

最小代码

node.pyPython

def chatbot(state: State) -> dict:
    reply = llm.invoke(state["messages"])
    # 只返回“要更新的部分”，不是整个 state
    return {"messages": [reply]}

心法 · 节点签名永远是 (state) → 局部更新 dict

③

Edge · 普通边

add_edge · 写死的流转

概念是什么

边定义「这个节点跑完接着跑哪个」。add_edge("a","b") 固定：a 完了一定去 b。

两个特殊端点：START（入口，第一条边从它出发）和 END（终点，走到就结束）。

最小代码

edges.pyPython

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")   # 入口 → chatbot
builder.add_edge("chatbot", END)     # chatbot → 结束

④

Conditional Edge · 条件边

add_conditional_edges · 动态路由

概念是什么

当「下一步走哪」要看运行时状态决定时，用条件边。给一个路由函数：它读 state，返回一个标签；再给一张「标签 → 目标节点」的映射表。

这是 Agent 的灵魂：同一个节点，模型说「要调工具」就去 tools，说「答完了」就去 END。

最小代码

route.pyPython

def route(state: State) -> str:
    last = state["messages"][-1]
    if last.tool_calls:        # 模型想调工具
        return "tools"
    return "end"             # 否则收尾

builder.add_conditional_edges(
    "agent", route,
    {"tools": "tools", "end": END},  # 标签→节点
)

⑤

compile + invoke · 编译并运行

把蓝图变成可执行图

概念是什么

前面 builder 只是蓝图。compile() 把它编译成可执行的 graph，再像普通函数一样 invoke 一个初始状态。

返回的是跑完后的最终 state。想看每一步可以用 stream() 流式拿中间状态。

最小代码

run.pyPython

graph = builder.compile()

result = graph.invoke(
    {"messages": [{"role":"user",
                  "content":"你好"}]}
)
print(result["messages"][-1])   # 最终状态里的回复

调试 · 用 graph.stream(...) 可逐节点观察状态变化

— 02

跑通一个带工具的 Agent

把上面五件东西拼起来，就是 LangGraph 最经典的形态：一个会自己决定「要不要查工具」的 Agent。关键是那条指回去的边——它让图成了环。

Agent 循环图

整张图从左向右阅读

只有调用工具的分支会返回 agent；不需要工具时，流程直接结束。

▶

完整骨架

把 ① 至 ⑤ 拼起来

agent_app.pyPython

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode   # 官方现成的工具执行节点

builder = StateGraph(State)
builder.add_node("agent", agent)            # ② 调 LLM(已 bind_tools)
builder.add_node("tools", ToolNode(tools))   # ② 执行工具

builder.add_edge(START, "agent")          # ③ 入口
builder.add_conditional_edges(            # ④ agent 后分叉
    "agent", route, {"tools": "tools", "end": END})
builder.add_edge("tools", "agent")        # ③ 关键：工具→agent，成环！

graph = builder.compile()               # ⑤
graph.invoke({"messages": [{"role":"user","content":"北京今天天气?"}]})

它怎么转 · 一次问答循环了两轮 agent

agent 想→ 路由：要查天气→ tools 查→ 回 agent→ 看到结果→ END

轮数不写死，由模型自己决定何时停——这就是 Agent。

TIP只想要标准 Agent，官方有现成的 create_react_agent(model, tools) 一行生成同款图。但先手搓一遍上面这套，你才真懂内部在干嘛——之后要定制（加审核节点、改路由）就有的放矢。

— 03

记忆与中断

到这里图能跑，但跑完即忘、也不能中途停下等人。加一个 Checkpointer 就解锁多轮记忆、暂停恢复、人工介入。

⑥

Checkpointer · 检查点

持久化 + thread_id

概念是什么

Checkpointer 在每一步之后把 state 存下来，用 thread_id 区分是哪次会话。于是：

多轮记忆：同一 thread_id 再次 invoke，自动接着上次状态。
暂停/恢复：可在某节点前 interrupt，等人确认再继续。
回放/纠错：能回到任一历史检查点重跑（time travel）。

最小代码

memory.pyPython

from langgraph.checkpoint.memory import MemorySaver

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

cfg = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages":[u("我叫小明")]}, cfg)
graph.invoke({"messages":[u("我叫啥?")]}, cfg)
# 第二次能答"小明"——同一 thread_id 记住了历史

生产环境 · 把 MemorySaver 换成 SqliteSaver / PostgresSaver 即可落库

human-in-the-loop · 编译时加 interrupt_before=["tools"]，图会在执行工具前停住；你检查/修改后再 graph.invoke(None, cfg) 让它接着跑。这就是「危险操作先让人点确认」的实现方式——靠的正是 checkpointer 把状态存住了才能停。

— 04

易混点与上手清单

新手最容易栽的几个坑，以及一条最短上手路径。

常见误区

常见误解	实际情况
节点要返回一个完整的新 state。	节点只返回变化的字段（局部 dict），框架按 reducer 合并进总状态。
想让 messages 累积，自己在节点里手动拼接历史。	给字段挂 add_messages reducer，返回 `{"messages":[新消息]}` 就自动追加。手动拼反而和 reducer 打架。
条件边的路由函数要返回目标节点名。	它返回的是标签，再经第三个参数那张映射表翻译成节点。标签和节点名可不同。
图能跑就自带多轮记忆。	不加 checkpointer 就是无状态的，跑完即忘。多轮记忆 = checkpointer + 固定 `thread_id`。
有环就会死循环。	环是特性不是 bug。靠条件边的退出分支（返回 END）收敛；另可设 `recursion_limit` 兜底。

三步上手

第 1 步

装与跑通线性图：pip install langgraph，照①②③⑤写 START→chatbot→END，先让 invoke 出结果。

第 2 步

加条件边成环：接上 tools 节点 + route 路由，做出 §02 那个会循环的 Agent，体会「轮数由模型决定」。

第 3 步

加 checkpointer：挂 MemorySaver + thread_id 试多轮记忆，再试 interrupt_before 做一次人工确认。

该不该用 LangGraph

✓ 该用 LangGraph

需要循环 / 分支 / 多 Agent 协作 / 中途等人 / 长任务断点续跑——这才是它的主场。

✗ 不必用

流程线性、跑完就完（纯检索问答、单次总结）→ 普通链就行，上 LangGraph 是过度设计。