一张图了解全貌
┌─────────────────────────────────────────────────────────────┐
│ LangGraph 全局视角 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ State │ │ Node │ │ Edge │ │
│ │ 共享数据 │───▶│ 处理步骤 │───▶│ 跳转逻辑 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ TypedDict def node() add_edge() │
│ MessagesState 接收state add_conditional_edges() │
│ Annotated+Reducer 返回更新 │
│ │
│ ┌──────────────────────────────────────────────┐ │
│ │ StateGraph → compile() → invoke() │ │
│ │ 定义图 编译图 运行图 │ │
│ └──────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
一、是什么
一句话定义
LangGraph = 用有向图构建有状态的 AI Agent 工作流。
拆开看三个关键词：
有向图：节点 + 边，数据沿边流动，不会乱跑
有状态：所有节点共享同一份 State，每一步都能读到之前的结果
Agent 工作流：不是简单的链式调用，支持循环、分支、条件路由
核心三要素
┌────────────┐ ┌────────────┐ ┌────────────┐
│ State │ │ Node │ │ Edge │
│ 状态 │ │ 节点 │ │ 边 │
├────────────┤ ├────────────┤ ├────────────┤
│ 图执行过程中 │ │ 一个Python │ │ 节点之间的 │
│ 的共享数据 │ │ 函数，接收 │ │ 跳转逻辑 │
│ │ │ state，返回 │ │ │
│ TypedDict │ │ 更新 │ │ 固定跳转 or │
│ 定义结构 │ │ │ │ 条件路由 │
└────────────┘ └────────────┘ └────────────┘
对比 LangChain
打个比方：Chain 是单行道，LangGraph 是立交桥。 单行道只能直走，立交桥有岔路、有环路、有条件转向。
Chain（链）： A ──▶ B ──▶ C ──▶ 输出
LangGraph（图）：
┌──▶ 退款处理 ──▶ END
│
START ──▶ 意图分类 ──▶ 查询处理 ──▶ END
│
└──▶ 转人工 ──▶ END
二、怎么用
安装
pip install langgraph langchain-openai
就这么简单。langgraph 是核心库，langchain-openai 提供 LLM 接入。
最低依赖说明：Python >= 3.10，无需数据库、无需 Docker，装完就能跑。
环境变量
export OPENAI_API_KEY="sk-你的密钥"如果用其他兼容 OpenAI 接口的模型（如 DeepSeek），也可以：
export OPENAI_API_KEY="sk-你的密钥"
export OPENAI_BASE_URL="https://api.deepseek.com"
第一个验证脚本
装好了没？跑一下这个，能输出就说明环境 OK：
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class State(TypedDict):
text: str
def node_a(state: State) -> dict:
return {"text": state["text"] + "Hello "}
def node_b(state: State) -> dict:
return {"text": state["text"] + "LangGraph!"}
graph = StateGraph(State)
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", END)
app = graph.compile()
result = app.invoke({"text": ""})
print(result["text"]) # Hello LangGraph!看到 Hello LangGraph! 就成功了！
三、核心概念
3.1 State（状态）
State 是什么？ 图执行过程中的共享数据。所有节点都能读，都能写。
你可以把它想象成一张"共享便签纸"——每个节点往上面写内容，下一个节点能看到前面写的所有东西。
TypedDict
from typing import TypedDict
class AgentState(TypedDict):
messages: list # 对话历史
intent: str # 意图分类结果
count: int # 计数器这是最基础的写法，每个字段就是一个普通类型。
MessagesState
如果你的 State 主要就是聊天消息，LangGraph 提供了内置的 MessagesState，省得自己写：
from langgraph.graph import MessagesState
# MessagesState 等价于：
# class MessagesState(TypedDict):
# messages: Annotated[list[AnyMessage], add_messages]
它已经帮你定义好了 messages 字段，并且内置了 add_messages 合并策略。
Annotated + Reducer
（add_messages）合并策略
这是 State 最关键的机制：当多个节点更新同一个字段时，怎么合并？
┌─────────────────────────────────────────────────────┐
│ Reducer 合并策略 │
├─────────────────────────────────────────────────────┤
│ │
│ 没有 Reducer（默认）：新值 直接覆盖 旧值 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ count=1 │ ──▶ │ count=2 │ ──▶ │ count=2 │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ 节点A设置 节点B覆盖 最终值=2 │
│ │
│ 有 Reducer（如 add_messages）：新值 追加到 旧值 │
│ ┌────────────┐ ┌──────────────┐ ┌───────────┐ │
│ │ msgs=[A] │─▶ │ msgs=[A,B] │─▶ │ msgs=[A,B]│ │
│ └────────────┘ └──────────────┘ └───────────┘ │
│ 节点A追加 节点B追加 最终=A+B │
│ │
└─────────────────────────────────────────────────────┘from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import add_messages
from langchain_core.messages import AnyMessage
class AgentState(TypedDict):
# messages 字段用 add_messages reducer：追加而非覆盖
messages: Annotated[list[AnyMessage], add_messages]
# 普通字段：新值覆盖旧值
intent: str
# 计数器用 operator.add 做累加
count: Annotated[int, operator.add]add_messages 做了什么？
新消息没有 ID → 直接追加到列表末尾
新消息有 ID 且已存在 → 替换同 ID 的旧消息（更新）
新消息有 ID 且不存在 → 追加
这个设计非常贴合聊天场景：对话历史是追加的，但如果 AI 修正了之前的回复，可以通过相同 ID 覆盖。
3.2 Node（节点）
Node 是什么？ 一个 Python 函数，接收当前 state，返回部分 state 更新。
节点函数
def my_node(state: AgentState) -> dict:
# 1. 读取 state
current_messages = state["messages"]
# 2. 做点什么
result = do_something(current_messages)
# 3. 返回要更新的字段（不需要返回全部字段！）
return {"messages": [result], "count": 1}关键点：
只返回你想更新的字段，LangGraph 会自动合并
如果字段有 Reducer，按 Reducer 规则合并
如果没有 Reducer，新值直接覆盖旧值
START 和 END
LangGraph 有两个特殊的"虚拟节点"，用来标记图的入口和出口：
START ──▶ 第一个节点
最后一个节点 ──▶ ENDfrom langgraph.graph import START, END
graph.add_edge(START, "first_node") # 图的入口
graph.add_edge("last_node", END) # 图的出口示例
"""
写 3 个节点函数
"""
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
llm = ChatOpenAI(model="gpt-4o-mini")
def greet_node(state: AgentState) -> dict:
"""打招呼节点：根据意图生成问候语"""
return {"messages": [AIMessage(content="你好！有什么可以帮你的？")]}
def chat_node(state: AgentState) -> dict:
"""聊天节点：调用 LLM 回复用户"""
response = llm.invoke(state["messages"])
return {"messages": [response]}
def goodbye_node(state: AgentState) -> dict:
"""告别节点：生成结束语"""
return {"messages": [AIMessage(content="再见！祝你好运！")]}3.3 Edge（边）
Edge 是什么？ 节点之间的跳转逻辑。分为两种：普通边和条件边。
普通边
add_edge（固定跳转）
A 完了永远去 B，没有商量余地：
graph.add_edge(START, "greet") # 开始 → 打招呼
graph.add_edge("greet", "chat") # 打招呼 → 聊天
graph.add_edge("chat", END) # 聊天 → 结束条件边
add_conditional_edges（根据 state 动态路由）
A 完了去哪？看情况！这是 LangGraph 最强大的地方：
┌──▶ 退款处理
│
意图分类 ───┼──▶ 查询处理
│
└──▶ 转人工路由函数的
路由函数接收 state，返回一个字符串，表示下一个要去的节点名：
def route_by_intent(state: AgentState) -> str:
"""根据意图分类结果路由到不同节点"""
if state["intent"] == "refund":
return "refund_node"
elif state["intent"] == "query":
return "query_node"
else:
return "human_node"示例：条件分支
from langgraph.graph import StateGraph, START, END
graph = StateGraph(AgentState)
# 注册节点
graph.add_node("classify", classify_node)
graph.add_node("refund", refund_node)
graph.add_node("query", query_node)
graph.add_node("human", human_node)
# 入口
graph.add_edge(START, "classify")
# 条件边：classify 完了根据意图走不同分支
graph.add_conditional_edges(
"classify", # 源节点
route_by_intent, # 路由函数
{ # 返回值 → 目标节点映射
"refund": "refund",
"query": "query",
"human": "human",
}
)
# 各分支都到终点
graph.add_edge("refund", END)
graph.add_edge("query", END)
graph.add_edge("human", END)提示：
path_map字典是可选的。如果不传，LangGraph 会把路由函数的返回值直接当作目标节点名。
3.4 Compile（编译）
为什么必须编译？ 你定义的图只是一个"蓝图"，compile() 会做这些事：
┌───────────────────────────────────────────────┐
│ compile() 做了什么 │
├───────────────────────────────────────────────┤
│ │
│ 1. 类型检查 │
│ └─ State 字段类型是否一致 │
│ │
│ 2. 连通性验证 │
│ └─ 从 START 能否到达所有节点 │
│ └─ 所有节点能否到达 END │
│ └─ 有没有孤立节点 │
│ │
│ 3. 构建 Pregel 执行引擎 │
│ └─ 编译后才能 invoke / stream │
│ │
└───────────────────────────────────────────────┘# 编译
app = graph.compile()
# 带持久化的编译（后面第39篇会讲）
from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())编译后你才能调用 invoke() 和 stream()。
四、实战案例
ChatAgent
5 步构建
┌─────────────────────────────────────────────────────┐
│ 构建 LangGraph 的 5 步 │
├─────────────────────────────────────────────────────┤
│ │
│ 1️⃣ 定义 State ── TypedDict / MessagesState │
│ 2️⃣ 写 Node ── def node(state) -> dict │
│ 3️⃣ 画 Edge ── add_edge / add_conditional │
│ 4️⃣ Compile ── graph.compile() │
│ 5️⃣ Run ── app.invoke() / app.stream() │
└─────────────────────────────────────────────────────┘完整代码
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END, MessagesState
# ── 1️⃣ 定义 State ──────────────────────────────
# MessagesState 自带 messages 字段和 add_messages reducer
# ── 2️⃣ 写 Node ─────────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini")
def chatbot(state: MessagesState) -> dict:
response = llm.invoke(state["messages"])
return {"messages": [response]}
# ── 3️⃣ 画 Edge + 构建图 ────────────────────────
graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
# ── 4️⃣ Compile ─────────────────────────────────
app = graph.compile()
# ── 5️⃣ Run ─────────────────────────────────────
result = app.invoke({"messages": [HumanMessage(content="你好，介绍下你自己")]})
print(result["messages"][-1].content)invoke vs stream
# invoke：等全部执行完，一次性返回结果
result = app.invoke({"messages": [HumanMessage(content="你好")]})
# stream：逐步返回每个节点的输出，适合实时展示
for chunk in app.stream({"messages": [HumanMessage(content="你好")]}):
print(chunk) # 每个节点的增量输出什么时候用哪个？
invoke：后台任务、批量处理stream：实时对话、前端逐字展示
Agent + Tools
真正的 Agent 不只是聊天，还得能调用工具。这一节我们构建一个能查天气 + 搜索的 Agent。
整体架构
┌─────────────────────────────────────────────────────┐
│ 带工具的 Agent 架构 │
├─────────────────────────────────────────────────────┤
│ │
│ START ──▶ agent ──┬──▶ tools ──▶ agent ──┬──▶ END │
│ (LLM决策) │ (执行工具) (再决策) │ │
│ │ │ │
│ └──▶ END │ │
│ (无需工具,直接回答) │ │
│ │
│ 关键：agent 节点通过条件边决定 │
│ "继续调工具" 还是 "结束" │
│ │
└─────────────────────────────────────────────────────┘
定义 Tool
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""查询指定城市的天气"""
# 模拟天气查询（实际项目可对接真实 API）
weather_data = {
"北京": "晴天，-2°C",
"上海": "多云，5°C",
"深圳": "小雨，18°C",
}
return weather_data.get(city, f"暂无{city}的天气数据")
@tool
def search_web(query: str) -> str:
"""搜索互联网信息"""
# 模拟搜索（实际项目可对接搜索 API）
return f"搜索结果：关于「{query}」的最新信息..."
完整代码
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
# ── 定义工具 ────────────────────────────────────
@tool
def get_weather(city: str) -> str:
"""查询指定城市的天气"""
weather_data = {
"北京": "晴天，-2°C",
"上海": "多云，5°C",
"深圳": "小雨，18°C",
}
return weather_data.get(city, f"暂无{city}的天气数据")
@tool
def search_web(query: str) -> str:
"""搜索互联网信息"""
return f"搜索结果：关于「{query}」的最新信息..."
tools = [get_weather, search_web]
# ── 配置 LLM ───────────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
# ── 定义节点 ────────────────────────────────────
def agent_node(state: MessagesState) -> dict:
"""LLM 决策节点：决定是调用工具还是直接回答"""
system = SystemMessage(
content="你是一个智能助手，可以查询天气和搜索信息。请用中文回答。"
)
messages = [system] + state["messages"]
response = llm.invoke(messages)
return {"messages": [response]}
# ToolNode 是 LangGraph 内置的工具执行节点
tool_node = ToolNode(tools)
# ── 构建图 ─────────────────────────────────────
graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
# 条件边：LLM 是否要调用工具？
# tools_condition 是内置的路由函数：
# - 如果 LLM 返回了 tool_calls → 走 "tools" 节点
# - 如果 LLM 没有返回 tool_calls → 走 END
graph.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})
# 工具执行完回到 agent，让 LLM 再决策
graph.add_edge("tools", "agent")
# ── 编译运行 ───────────────────────────────────
app = graph.compile()
# 测试：查天气
result = app.invoke({"messages": [HumanMessage(content="北京今天天气怎么样？")]})
print(result["messages"][-1].content)
# 测试：搜索
result = app.invoke(
{"messages": [HumanMessage(content="帮我搜一下 LangGraph 最新版本")]}
)
print(result["messages"][-1].content)执行流程
用户: "北京今天天气怎么样？"
START → agent(LLM决策: 需要调get_weather)
→ tools(执行: get_weather("北京") → "晴天，-2°C")
→ agent(LLM决策: 信息够了,直接回答)
→ END
输出: "北京今天是晴天，气温-2°C，比较冷，注意保暖哦！"
CustomerAgent
现在来个更贴近实际业务的场景：一个智能客服，能根据用户意图走不同处理路径。
架构图
┌──────────────────────────────────────────────────────┐
│ 客服 Agent 架构 │
├──────────────────────────────────────────────────────┤
│ │
│ START ──▶ classify ──┬──▶ refund ────▶ END │
│ (意图分类) ├──▶ query ────▶ END │
│ └──▶ human ────▶ END │
│ │
│ classify: LLM 判断意图 → refund / query / human │
│ refund: 生成退款流程提示 │
│ query: 回答查询问题 │
│ human: 生成转人工话术 │
│ │
└──────────────────────────────────────────────────────┘
完整代码
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage, AIMessage
from langgraph.graph import StateGraph, START, END, add_messages
# ── 1️⃣ 定义 State ──────────────────────────────
class ServiceState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
intent: str # 意图: refund / query / human
response: str # 最终回复
# ── 2️⃣ 写 Node ─────────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini")
def classify_node(state: ServiceState) -> dict:
"""意图分类节点：判断用户是退款、查询还是需要转人工"""
last_msg = state["messages"][-1].content
prompt = [
SystemMessage(
content=(
"你是一个意图分类器。根据用户消息判断意图，只回复以下之一：\n"
"- refund：用户要求退款、退货、退钱\n"
"- query：用户查询订单、物流、商品信息\n"
"- human：用户投诉、要求转人工、情绪激动\n"
"只回复一个词，不要解释。"
)
),
HumanMessage(content=last_msg),
]
intent = llm.invoke(prompt).content.strip().lower()
# 确保意图在合法范围内
if intent not in ("refund", "query", "human"):
intent = "human"
return {"intent": intent}
def refund_node(state: ServiceState) -> dict:
"""退款处理节点"""
last_msg = state["messages"][-1].content
prompt = [
SystemMessage(
content=(
"你是退款客服。根据用户描述，给出退款指引。格式：\n"
"1. 确认退款原因\n"
"2. 告知退款流程\n"
"3. 预计到账时间\n"
"语气友好、专业。"
)
),
HumanMessage(content=last_msg),
]
response = llm.invoke(prompt).content
return {"response": response, "messages": [AIMessage(content=response)]}
def query_node(state: ServiceState) -> dict:
"""查询处理节点"""
last_msg = state["messages"][-1].content
prompt = [
SystemMessage(
content=(
"你是查询客服。回答用户关于订单、物流、商品的问题。"
"如果无法确认具体信息，建议用户提供订单号。语气友好。"
)
),
HumanMessage(content=last_msg),
]
response = llm.invoke(prompt).content
return {"response": response, "messages": [AIMessage(content=response)]}
def human_node(state: ServiceState) -> dict:
"""转人工节点"""
last_msg = state["messages"][-1].content
response = (
"我理解您的需求，已为您转接人工客服。\n"
"当前排队人数：3人，预计等待时间：2分钟。\n"
"请稍候，人工客服将很快为您服务。"
)
return {"response": response, "messages": [AIMessage(content=response)]}
# ── 3️⃣ 路由函数 ────────────────────────────────
def route_by_intent(state: ServiceState) -> str:
"""根据意图分类结果路由"""
return state["intent"]
# ── 4️⃣ 构建图 ─────────────────────────────────
graph = StateGraph(ServiceState)
graph.add_node("classify", classify_node)
graph.add_node("refund", refund_node)
graph.add_node("query", query_node)
graph.add_node("human", human_node)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
"classify",
route_by_intent,
{"refund": "refund", "query": "query", "human": "human"},
)
graph.add_edge("refund", END)
graph.add_edge("query", END)
graph.add_edge("human", END)
# ── 5️⃣ 编译运行 ───────────────────────────────
app = graph.compile()
# 测试：退款
result = app.invoke(
{"messages": [HumanMessage(content="我要退款！昨天买的东西还没到，不想要了")]}
)
print("【退款场景】", result["response"])
# 测试：查询
result = app.invoke(
{"messages": [HumanMessage(content="我的订单到哪了？订单号 ORD-20260220-001")]}
)
print("【查询场景】", result["response"])
# 测试：转人工
result = app.invoke(
{"messages": [HumanMessage(content="你们这是什么服务！我要投诉！给我转人工！")]}
)
print("【转人工场景】", result["response"])
评论区