LangGraph 的心智模型只有一句话:图就是「State 在节点间流动、每个节点返回的局部更新被 reducer 合并回 State」。你定义的不是控制流的每一步,而是「状态长什么样」+「哪个节点接哪个节点」+「每个字段如何合并」。

第 0 步:安装与准备 LLM

本页用 Anthropic Claude 作为 LLM(换成 OpenAI / 其它只需改一行 init 代码)。先装包并配置 API Key。langgraph 是核心,langchain 提供消息类型与统一的 chat model 接口。

# 安装核心包:langgraph 提供图引擎,langchain 提供 model 工厂与消息类型
pip install -U langgraph langchain "langchain[anthropic]"

# 配置 Anthropic API Key(换 OpenAI 则 export OPENAI_API_KEY=...)
export ANTHROPIC_API_KEY="sk-ant-..."

第 1 步:定义 State(带 add_messages 的 messages 字段)

State 是一个 TypedDict。普通字段「后写覆盖先写」,但聊天场景我们希望消息累积而不是被覆盖——这正是 reducer 的作用。给字段加 Annotated[list, add_messages],节点返回的新消息就会被 追加 进已有列表,而不是替换整个列表。

from typing import Annotated
from typing_extensions import TypedDict   # Python 3.9 用 typing_extensions 更稳
from langgraph.graph.message import add_messages

class State(TypedDict):
    # messages 字段:list 是类型,add_messages 是 reducer
    # reducer 决定「节点返回的新 messages 如何并入旧 messages」
    # add_messages = 按 id 去重的「追加」,而不是覆盖
    messages: Annotated[list, add_messages]

第 2 步:写一个调用 LLM 的 chatbot 节点

节点就是一个普通函数:签名 (state: State) -> dict。它读取当前 State,做点事(这里是把历史消息丢给 LLM),然后 只返回要更新的字段。注意节点返回 {"messages": [response]}——只是一条新消息,框架会通过 add_messages reducer 把它追加到完整列表里,你不用自己拼接历史。

from langchain.chat_models import init_chat_model

# 初始化 LLM:一行切换 provider/model
llm = init_chat_model("anthropic:claude-sonnet-4-5")

def chatbot(state: State) -> dict:
    # state["messages"] 是到目前为止的完整对话历史(list[BaseMessage])
    # llm.invoke 接受消息列表,返回一条 AIMessage
    response = llm.invoke(state["messages"])
    # 关键:只返回新增的那一条,reducer 负责追加,不要手动拼 state["messages"] + [response]
    return {"messages": [response]}

第 3 步:组装图 —— add_node / add_edge / compile(固定四步)

  1. builder = StateGraph(State):用 State 类型创建一个图构造器(builder)。
  2. builder.add_node("chatbot", chatbot):把函数注册为名为 chatbot 的节点。
  3. builder.add_edge(START, "chatbot"):声明入口——从虚拟起点 START 指向 chatbot。builder.add_edge("chatbot", END):声明出口——chatbot 跑完到 END 结束。
  4. graph = builder.compile():把可变的 builder 冻结成一个 不可变、可执行的 CompiledGraph
from langgraph.graph import StateGraph, START, END

# 1) 用 State 创建图构造器
builder = StateGraph(State)

# 2) 注册节点:第一个参数是节点名(字符串),第二个是可调用对象
builder.add_node("chatbot", chatbot)

# 3) 连边:START 是内置虚拟入口,END 是内置虚拟出口
builder.add_edge(START, "chatbot")   # 入口 -> chatbot
builder.add_edge("chatbot", END)     # chatbot -> 结束

# 4) 编译:builder 是「设计图」,compile() 产出「可运行的引擎」CompiledGraph
graph = builder.compile()
对象是什么能做什么可变吗
StateGraph (builder)图的「设计蓝图」add_node / add_edge / add_conditional_edges可变,未编译前随便改
CompiledGraphcompile() 的产物,可执行引擎invoke / stream / get_graph / get_state不可变,编译后结构冻结
START / END内置虚拟节点(哨兵)标记图的入口与出口,本身不执行代码常量,不可改
口诀Builder 是图纸,Compile 是开工,CompiledGraph 是建好的工厂——图纸能改,工厂只能用。

第 4 步:invoke —— 输入 dict,输出合并后的完整 State

graph.invoke(input) 的输入是一个 符合 State 结构的 dict。这里我们传初始 messages,可以是元组列表 [("user", "...")](LangChain 会自动转成 HumanMessage),也可以直接传消息对象。返回值同样是一个 dict——但它是 所有节点跑完后、被 reducer 合并好的完整 State,里面的 messages 已经包含「你的输入 + LLM 的回复」。

# invoke 输入:一个 dict,键必须是 State 里定义的字段
result = graph.invoke(
    {"messages": [("user", "用一句话解释什么是 LangGraph")]}
)

# invoke 输出:合并后的完整 State(也是一个 dict)
# result["messages"] = [HumanMessage(你的输入), AIMessage(LLM 回复)]
print(type(result))              # <class 'dict'>
print(len(result["messages"]))   # 2:输入1条 + 回复1条

# 取最后一条(LLM 的回复)
print(result["messages"][-1].content)

# 也可以用 pretty_print 看完整对话
for m in result["messages"]:
    m.pretty_print()

第 5 步:可视化 —— draw_mermaid() 零依赖看图结构

graph.get_graph() 返回图的底层结构对象,.draw_mermaid() 把它渲染成 Mermaid 文本(不需要装任何画图库 / 浏览器)。把输出粘到任意支持 Mermaid 的地方(Markdown 预览、mermaid.live)就能看到节点与边。这是排查「边连错了 / 节点没接上」最快的手段。

# 方式一:打印 Mermaid 文本(零依赖,终端直接可用)
print(graph.get_graph().draw_mermaid())

# 输出大致如下(START/END 为虚拟节点):
# ---
# graph TD;
#     __start__([__start__]) --> chatbot;
#     chatbot --> __end__([__end__]);
# ---

# 方式二:渲染成 PNG 字节(需要联网,调用 mermaid.ink 服务)
# png_bytes = graph.get_graph().draw_mermaid_png()
# with open("graph.png", "wb") as f:
#     f.write(png_bytes)

# 在 Jupyter 里可直接显示:
# from IPython.display import Image
# Image(graph.get_graph().draw_mermaid_png())

完整可运行脚本(粘贴即跑)

# === 完整最小 StateGraph:定义 State -> 写节点 -> 组装 -> invoke -> 可视化 ===
# 运行前:pip install -U langgraph langchain "langchain[anthropic]"
#         export ANTHROPIC_API_KEY=sk-ant-...

from typing import Annotated
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1) State:messages 字段用 add_messages reducer 实现「追加而非覆盖」
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2) LLM 与节点
llm = init_chat_model("anthropic:claude-sonnet-4-5")

def chatbot(state: State) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}

# 3) 组装图(固定四步)
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile()

# 4) 可视化(零依赖)
print(graph.get_graph().draw_mermaid())

# 5) 运行
result = graph.invoke({"messages": [("user", "用一句话解释什么是 LangGraph")]})
print("消息条数:", len(result["messages"]))
print("LLM 回复:", result["messages"][-1].content)
推荐做法
  • 节点只返回「要更新的字段」的增量,例如 {"messages": [response]},让 reducer 负责合并
  • 用内置 START / END 常量连边,不要自己造字符串入口名
  • 改完图结构后第一时间 print(graph.get_graph().draw_mermaid()) 核对连线
  • invoke 的输入 dict 只填初始字段,其余交给节点产出
不推荐
  • 不要在节点里返回 state["messages"] + [response](会让历史翻倍)
  • 不要忘了 compile() 就直接 invoke——builder 本身不可执行
  • 不要把整个 State 字段都在节点里重写一遍,只返回变化的部分
  • 不要给 messages 字段漏掉 Annotated[..., add_messages],否则消息会被覆盖只剩最后一条
常见误区
  • 漏配 ANTHROPIC_API_KEY:init_chat_model 会在第一次 invoke 时报认证错误
  • Python 3.9 直接从 typing 导 TypedDict 在某些组合下行为不一致,优先用 typing_extensions
  • draw_mermaid_png() 需要联网(走 mermaid.ink),离线环境用纯文本 draw_mermaid()

运行脚本后终端打印出 Mermaid 文本 + 「消息条数: 2」+ 一句 LLM 回复,即视为图端到端跑通。

invoke 返回的 messages 越跑越多 / 历史翻倍

典型表现
每次 invoke 后 messages 条数异常增长,出现重复消息
判断标准
节点返回值里是否手动拼接了 state["messages"]
解决方向
节点只返回增量 {"messages": [response]};合并交给 add_messages reducer,删掉手动拼接

messages 只剩最后一条,历史丢失

典型表现
对话没有上下文,LLM 看不到之前的消息
判断标准
State 的 messages 字段是否带 Annotated[..., add_messages]
解决方向
给字段加 reducer:messages: Annotated[list, add_messages];不加则默认覆盖

'CompiledGraph' object has no attribute 'add_node'

典型表现
在 compile() 之后还想 add_node / add_edge
判断标准
是否对 builder 还是对 graph 调用了构造方法
解决方向
结构修改都在 compile() 之前对 builder 做;CompiledGraph 是冻结的,只能 invoke/stream