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(固定四步)
builder = StateGraph(State):用 State 类型创建一个图构造器(builder)。builder.add_node("chatbot", chatbot):把函数注册为名为chatbot的节点。builder.add_edge(START, "chatbot"):声明入口——从虚拟起点 START 指向 chatbot。builder.add_edge("chatbot", END):声明出口——chatbot 跑完到 END 结束。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 | 可变,未编译前随便改 |
| CompiledGraph | compile() 的产物,可执行引擎 | invoke / stream / get_graph / get_state | 不可变,编译后结构冻结 |
| START / END | 内置虚拟节点(哨兵) | 标记图的入口与出口,本身不执行代码 | 常量,不可改 |
第 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