Agent 写着写着就成了意大利面条——LangGraph 焊死有状态图编排（上篇：入门概念篇）

本系列共三篇：上篇（入门概念篇）/ 中篇（高级特性篇）/ 下篇（生产实战篇）。完整代码案例可在系列下篇末尾找到 GitHub 仓库链接。

本文示例基于 LangGraph 稳定版，所有代码均基于最新 API 验证。实际使用时请以 PyPI 和 官方文档 为准。文中 API Key / 密钥均为占位符示例，请勿硬编码到代码或提交到代码仓库。

你的 Agent 代码为什么变成了意大利面条

你用 LangChain 写了一个 Agent。刚开始很顺利——Chain 串 Chain，Prompt 套 Prompt，跑起来效果还行。然后你加了工具调用，加了条件分支，加了多轮对话记忆，加了人工审批节点。

突然间你发现：

• 流程控制全靠 if-else 堆叠。每个条件分支都是一层嵌套，三层以后你自己都看不懂代码在干什么。
• 状态在函数间靠参数传递。五个函数共享一个 dict，谁改了什么、什么时候改的，全靠注释和记忆。
• 没有循环能力。Agent 需要反复调用工具直到任务完成，但 LangChain 的 Chain 是线性的，你只能在外面套 while True。
• 无法持久化中间状态。服务重启后，Agent 的执行上下文全丢了，用户得从头来。
• 调试是黑箱。Agent 调了三次工具、走了两次分支，但你不知道它为什么这么走。print 大法散落几百个文件。

这不是你代码写得差，是工具的抽象层级不对。LangChain 的 Chain 是"管道"模型——数据从一头流到另一头。但真实 Agent 不是管道，它是有状态的图：有分支、有循环、有条件跳转、有中断恢复、有人工介入。

LangGraph 就是来解决这个问题的。它把 Agent 建模为有向图：节点执行计算，边定义流程，State 全局共享，支持循环、分支、持久化、人工介入。受 Google Pregel 和 Apache Beam 启发，API 设计参考 NetworkX。

一、LangGraph 概述

1.1 什么是 LangGraph

LangGraph 是 LangChain 团队推出的低级别编排框架，用于构建、管理和部署长期运行的、有状态的 AI Agent。核心理念一句话：把 Agent 建模为有向图。

图中有四个核心元素：

元素	作用	类比
State（状态）	全局数据容器，所有节点共享	共享黑板
Node（节点）	执行计算的函数	函数 / 任务
Edge（边）	定义节点间的流转方向	if-else / goto
Checkpointer（检查点）	持久化状态快照	存档点

LangGraph 不抽象 Prompt、不隐藏架构、不限制认知架构。它是"低级别"的——给你图的基本原语（节点、边、状态），你自己搭。这意味着灵活性和可控性极高，但也意味着你需要理解图的执行模型。

1.2 使用 LangGraph 的理由

能力	LangChain Chain	手写 if-else	LangGraph
线性流程	原生支持	能写但乱	支持
条件分支	不原生	能写但更乱	条件边原生支持
循环（ReAct 轮次）	不支持	while True	图的环原生支持
全局状态管理	参数传递	全局变量	State + Reducer
持久化 / 断点恢复	不支持	自己实现	Checkpointer 原生
人工介入	不支持	自己实现	interrupt 原生
流式输出	部分	自己实现	三种流式模式
可视化调试	不支持	不可能	LangGraph Studio
多 Agent 编排	不支持	极难	Supervisor / 并行 / 反馈循环

一句话：如果你的 Agent 超过 3 个步骤、需要条件分支或循环、需要持久化或人工介入，就该用 LangGraph。

选型提醒：简单线性流程（单次 LLM 调用、简单 RAG 链）不要过度上 LangGraph，普通 LangChain Chain 足够。过度工程会增加不必要的复杂度和维护成本。只有当流程确实需要循环、分支、持久化或人工介入时，才值得引入 LangGraph。

1.3 适用场景

• ReAct Agent：推理-行动-观察循环
• 多步骤 RAG：检索-评估-再检索-生成
• 多 Agent 系统：Supervisor 协调多个专业 Agent
• 人工审批流：AI 生成内容，人工审核后发布
• 长任务编排：研究、写作、代码生成等需要多轮迭代的任务
• 对话系统：多轮对话带状态管理和记忆

1.4 LangGraph vs LangChain vs AutoGen

维度	LangChain	LangGraph	AutoGen
定位	LLM 应用组件库	有状态图编排框架	多智能体对话框架
核心抽象	Chain（管道）	StateGraph（有向图）	Agent（对话角色）
状态管理	参数传递	全局 State + Reducer	对话历史
流程控制	线性 + 简单分支	图（循环、分支、并行）	对话轮次
循环支持	不原生	原生（图的环）	原生（对话循环）
持久化	不原生	Checkpointer	不原生
人工介入	不原生	interrupt	不原生
多 Agent	不原生	Supervisor / 网络	核心能力
适用规模	小型应用	中大型应用	多 Agent 系统
GitHub Stars	活跃开源项目	活跃开源项目	活跃开源项目
开源协议	MIT	MIT	MIT

选型建议：简单 RAG / 单次调用用 LangChain；复杂 Agent 工作流用 LangGraph；纯多 Agent 对话场景用 AutoGen。三者不互斥——LangGraph 可以用 LangChain 的组件，AutoGen 也可以用 LangChain 的 LLM 封装。

二、核心概念体系

2.1 四大核心概念

State：图的"共享内存"。所有节点读写同一个 State 对象。State 用 TypedDict 或 Pydantic 定义，支持自定义合并策略（Reducer）。

Node：一个普通 Python 函数，接收 State，返回 State 的更新部分。每个节点做一件事——调用 LLM、执行工具、检索文档、做业务逻辑。

Edge：定义节点间的执行顺序。普通边是固定的（A 执行完一定到 B），条件边是动态的（A 执行完根据 State 决定去 B 还是 C）。

Checkpointer：在每个节点执行后自动保存 State 快照。支持中断恢复、时间旅行（回滚到任意检查点）、人工介入。

2.2 执行模型（Pregel-like）

LangGraph 的执行模型受 Google Pregel 启发，是一种**超步（superstep）**模型：

每个超步：

1. 读取当前 State
2. 执行节点函数
3. 节点返回 State 更新
4. 用 Reducer 合并更新到全局 State
5. Checkpointer 保存快照
6. 根据边决定下一个节点

这个模型的关键特性：节点间通过 State 通信，不直接调用彼此。这让节点解耦，可独立测试、可并行执行。

大白话总结：每一轮超步 = 执行一批节点 → 合并状态 → 保存检查点 → 挑选下一批节点。循环直到没有节点可执行或到达 END。

2.3 核心执行流程

三、安装与环境配置

3.1 安装

# 核心包
pip install -U langgraph

# 带检查点依赖
pip install -U langgraph langgraph-checkpoint-sqlite langgraph-checkpoint-postgres

# LangChain 集成（可选，但推荐）
pip install -U langgraph langchain langchain-openai

# LangGraph CLI（开发 / 部署工具）
pip install -U langgraph-cli

# Supervisor 多 Agent 库（可选，注意：langgraph-supervisor 目前为实验性库，API 可能变动，生产环境慎用）
pip install -U langgraph-supervisor

版本锁定建议：生产环境务必锁定 langgraph 版本号（如 pip install langgraph==0.2.x），该库 API 迭代比较活跃，避免 pip install -U langgraph 自动升级导致代码失效。建议在 requirements.txt 或 pyproject.toml 中固定版本。

3.2 环境变量

import os

# LLM API Key（按需配置）
# ⚠️ 以下为占位符，实际使用时请通过环境变量或 .env 文件注入，切勿硬编码到代码或提交到 git！
os.environ["OPENAI_API_KEY"] = "sk-xxx"
os.environ["ANTHROPIC_API_KEY"] = "sk-xxx"

# LangSmith 追踪（可选，强烈推荐）
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls__xxx"
os.environ["LANGSMITH_PROJECT"] = "langgraph-demo"

# LangGraph API（如果用 LangGraph Cloud）
# os.environ["LANGGRAPH_API_URL"] = "http://localhost:2024"

四、快速入门

4.1 Hello World（最简示例）

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. 定义状态
class State(TypedDict):
    messages: list[str]

# 2. 定义节点函数
def greet(state: State) -> dict:
    return {"messages": state["messages"] + ["Hello from greet node!"]}

def farewell(state: State) -> dict:
    return {"messages": state["messages"] + ["Goodbye from farewell node!"]}

# 3. 构建图
graph = StateGraph(State)
graph.add_node("greet", greet)
graph.add_node("farewell", farewell)

# 4. 连接边
graph.add_edge(START, "greet")
graph.add_edge("greet", "farewell")
graph.add_edge("farewell", END)

# 5. 编译并执行
app = graph.compile()
result = app.invoke({"messages": ["user: Hi"]})
print(result["messages"])
# ['user: Hi', 'Hello from greet node!', 'Goodbye from farewell node!']

五个步骤：定义 State -> 定义节点 -> 构建图 -> 连接边 -> 编译执行。这就是 LangGraph 的全部基础。

4.2 带条件分支的 Agent

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    intent: str
    response: str

def classify(state: State) -> dict:
    query = state["query"].lower()
    if "code" in query or "program" in query:
        intent = "coding"
    elif "search" in query or "find" in query:
        intent = "search"
    else:
        intent = "chat"
    return {"intent": intent}

def coding_agent(state: State) -> dict:
    return {"response": f"[Coder] 处理编程任务: {state['query']}"}

def search_agent(state: State) -> dict:
    return {"response": f"[Searcher] 搜索相关信息: {state['query']}"}

def chat_agent(state: State) -> dict:
    return {"response": f"[Chat] 回答问题: {state['query']}"}

# 条件路由函数
def route_by_intent(state: State) -> str:
    return state["intent"]  # 返回节点名称

# 构建图
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("coding", coding_agent)
graph.add_node("search", search_agent)
graph.add_node("chat", chat_agent)

graph.add_edge(START, "classify")

# 条件边：根据 intent 路由到不同节点
graph.add_conditional_edges(
    "classify",
    route_by_intent,
    {
        "coding": "coding",
        "search": "search",
        "chat": "chat",
    }
)

# 所有 Agent 执行完到 END
graph.add_edge("coding", END)
graph.add_edge("search", END)
graph.add_edge("chat", END)

app = graph.compile()

# 测试
print(app.invoke({"query": "帮我写个 Python 函数"})["response"])
# [Coder] 处理编程任务: 帮我写个 Python 函数

print(app.invoke({"query": "搜索 LangGraph 教程"})["response"])
# [Searcher] 搜索相关信息: 搜索 LangGraph 教程

条件分支是 LangGraph 最核心的能力之一。add_conditional_edges 接收三个参数：源节点、路由函数、路由映射。路由函数返回字符串，映射决定去哪个节点。

五、StateGraph 核心

5.1 StateGraph 基本用法

StateGraph 是 LangGraph 最核心的图构建器，专门搭建带共享全局状态的 Agent 工作流。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息列表，自动追加
    user_id: str
    step_count: int
    context: dict

graph = StateGraph(AgentState)

# 注册节点
graph.add_node("init", init_node)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)

# 连接边
graph.add_edge(START, "init")
graph.add_edge("init", "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

app = graph.compile()

StateGraph vs 基础 Graph：基础 Graph 没有统一 State，节点间靠 inputs 字典传参。StateGraph 有全局 State，所有节点共享。绝大多数业务开发优先使用 StateGraph。

5.2 MessagesState（预置状态）

LangGraph 提供了 MessagesState，内置了消息列表的状态定义，省去手动写 Annotated[list, add_messages]：

from langgraph.graph import StateGraph, START, END
from langgraph.graph import MessagesState

# MessagesState 等价于：
# class MessagesState(TypedDict):
#     messages: Annotated[list, add_messages]

graph = StateGraph(MessagesState)

def chatbot(state: MessagesState) -> dict:
    return {"messages": [("ai", "我是聊天机器人")]}

graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

app = graph.compile()
result = app.invoke({"messages": [("user", "你好")]})
print(result["messages"][-1].content)  # 我是聊天机器人

六、状态（State）管理

6.1 TypedDict 状态定义

from typing import TypedDict

class SimpleState(TypedDict):
    query: str
    documents: list[str]
    answer: str
    score: float

TypedDict 是 Python 的类型提示工具，定义一个字典的键和值类型。LangGraph 用它来声明 State 的结构。

6.2 Annotated 与状态合并策略

这是 LangGraph 最精妙的设计之一。当多个节点返回同一个字段的更新时，怎么合并？默认是覆盖，但你可以用 Annotated 指定合并策略（Reducer）：

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import add_messages

class State(TypedDict):
    # 覆盖策略（默认）：后写覆盖先写
    query: str
    answer: str

    # 追加策略：用 operator.add，列表拼接
    documents: Annotated[list[str], add]

    # 消息追加策略：LangGraph 内置的 add_messages
    messages: Annotated[list, add_messages]

    # 自定义 Reducer
    score: Annotated[float, "max"]  # 取最大值（需要自定义 reducer）

add_messages 是 LangGraph 内置的消息合并策略：新消息追加到列表末尾，如果消息 ID 已存在则替换。这适合对话场景——消息只增不改，除非显式修正。

自定义 Reducer 示例：

from typing import TypedDict, Annotated
# 注意：以下 keep_max 和 merge_dicts 是自定义 Reducer 函数，可直接定义在模块顶层

def keep_max(left: float, right: float) -> float:
    """取两个值的最大值"""
    return max(left, right) if left is not None else right

def merge_dicts(left: dict, right: dict) -> dict:
    """深度合并两个字典"""
    result = (left or {}).copy()
    result.update(right or {})
    return result

class State(TypedDict):
    confidence: Annotated[float, keep_max]
    metadata: Annotated[dict, merge_dicts]
    steps: Annotated[list[str], add]  # operator.add = 列表拼接

Reducer 的执行时机：每个超步结束时，节点返回的 State 更新按字段逐一用对应 Reducer 合并到全局 State。

6.3 状态读写操作

节点函数读取 State 的方式跟读字典一样：

def my_node(state: State) -> dict:
    # 读取
    query = state["query"]
    docs = state.get("documents", [])  # 安全读取，带默认值

    # 处理逻辑
    result = process(query, docs)

    # 返回更新（只返回要更新的字段，不是全量 State）
    return {
        "answer": result,
        "documents": docs + [result],  # 如果有 Reducer，这里追加
    }

关键规则：节点只返回要更新的字段，不需要返回完整 State。LangGraph 会用 Reducer 自动合并。

七、节点（Nodes）

7.1 节点函数规范

节点就是一个普通 Python 函数：

def node_name(state: State) -> dict:
    """
    节点函数规范：
    - 入参：当前 State（只读语义，不要原地修改）
    - 返回：State 的更新部分（dict）
    - 不要返回完整 State，只返回变化的字段
    """
    # 业务逻辑
    result = do_something(state["query"])

    # 返回更新
    return {"answer": result}

也可以用异步函数：

async def async_node(state: State) -> dict:
    docs = await async_retrieve(state["query"])
    return {"documents": docs}

7.2 节点类型

节点类型	说明	示例
LLM 节点	调用大语言模型	调 GPT-4o 生成回答
工具节点	执行外部工具	搜索、计算、API 调用
检索节点	从向量库检索文档	RAG 的 retrieve 步骤
逻辑节点	纯业务逻辑	分类、过滤、格式化
Agent 节点	一个完整的子 Agent	多 Agent 系统中的专业 Agent
人工节点	等待人工输入	Human-in-the-Loop 审批

7.3 带 LLM 的节点

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def llm_node(state: State) -> dict:
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}  # add_messages Reducer 自动追加

def summarize_node(state: State) -> dict:
    docs = state["documents"]
    context = "\n".join(docs)
    response = llm.invoke([
        HumanMessage(content=f"基于以下内容总结：\n{context}")
    ])
    return {"answer": response.content, "messages": [response]}

7.4 带工具调用的节点

from langchain_core.tools import tool
from langchain_core.messages import ToolMessage  # 方式 2 手动实现需要
from langgraph.prebuilt import ToolNode

@tool
def search_web(query: str) -> str:
    """搜索网页"""
    # 实际调用搜索 API
    return f"搜索结果: {query}"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    # ⚠️ 警告：eval() 是高危函数，允许执行任意 Python 代码，严禁在生产环境中使用！
    # 生产替代方案：使用 ast.literal_eval（仅限字面量）或 numexpr 库（数学表达式）
    # pip install numexpr; import numexpr; return str(numexpr.evaluate(expression))
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {e}"

tools = [search_web, calculator]

# 方式 1：用预置 ToolNode
tool_node = ToolNode(tools)

# 方式 2：手动实现
def call_tools(state: State) -> dict:
    last_message = state["messages"][-1]
    tool_calls = last_message.tool_calls

    results = []
    for tc in tool_calls:
        for t in tools:
            if t.name == tc["name"]:
                result = t.invoke(tc["args"])
                results.append(
                    ToolMessage(content=str(result), tool_call_id=tc["id"])
                )
    return {"messages": results}

ToolNode 是预置节点，自动处理 tool_calls 的解包、执行、返回 ToolMessage。生产环境推荐用 ToolNode。

八、边（Edges）

8.1 边的类型

边类型	方法	说明
普通边	add_edge(A, B)	A 执行完一定到 B
条件边	add_conditional_edges(A, router, mapping)	A 执行完根据 router 返回值决定去哪
入口边	add_edge(START, A)	图从 A 开始
出口边	add_edge(A, END)	A 执行完图结束

8.2 普通边（Fixed Edges）

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

普通边定义固定的执行顺序。A 执行完，下一个一定是 B。

8.3 条件边（Conditional Edges）

def should_continue(state: State) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"  # 有工具调用，去执行工具
    return END  # 没有工具调用，结束

graph.add_conditional_edges(
    "agent",        # 源节点
    should_continue, # 路由函数
    {
        "tools": "tools",  # 路由返回 "tools" -> 去节点 "tools"
        END: END,          # 路由返回 END -> 结束
    }
)

条件边的路由函数返回一个字符串，映射字典决定这个字符串对应哪个节点。如果不传映射字典，路由函数返回值直接当作节点名。

九、条件边与分支

9.1 条件分支示例

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    difficulty: str
    answer: str

def assess(state: State) -> dict:
    query = state["query"]
    if len(query) > 100 or any(w in query for w in ["分析", "对比", "设计"]):
        return {"difficulty": "hard"}
    return {"difficulty": "easy"}

def easy_handler(state: State) -> dict:
    return {"answer": f"快速回答: {state['query']}"}

def hard_handler(state: State) -> dict:
    return {"answer": f"深度分析: {state['query']}"}

def route_by_difficulty(state: State) -> str:
    return "hard" if state["difficulty"] == "hard" else "easy"

graph = StateGraph(State)
graph.add_node("assess", assess)
graph.add_node("easy", easy_handler)
graph.add_node("hard", hard_handler)

graph.add_edge(START, "assess")
graph.add_conditional_edges(
    "assess",
    route_by_difficulty,
    {"hard": "hard", "easy": "easy"}
)
graph.add_edge("easy", END)
graph.add_edge("hard", END)

app = graph.compile()

9.2 多层条件嵌套

def route_first(state: State) -> str:
    if state["difficulty"] == "hard":
        return "research"
    return "direct"

def route_after_research(state: State) -> str:
    if state.get("need_review"):
        return "review"
    return "generate"

graph = StateGraph(State)
graph.add_node("assess", assess)
graph.add_node("research", research_node)
graph.add_node("review", review_node)
graph.add_node("direct", direct_node)
graph.add_node("generate", generate_node)

graph.add_edge(START, "assess")
graph.add_conditional_edges("assess", route_first, {
    "research": "research",
    "direct": "direct"
})
# research 之后还可以条件分支
graph.add_conditional_edges("research", route_after_research, {
    "review": "review",
    "generate": "generate"
})
graph.add_edge("review", "generate")
graph.add_edge("direct", END)
graph.add_edge("generate", END)

app = graph.compile()

条件边可以任意嵌套——任何节点之后都可以接条件边，形成复杂的多层决策树。

十、START 与 END

10.1 START（入口点）

from langgraph.graph import START

# 图的入口
graph.add_edge(START, "first_node")

START 是图的虚拟入口节点。每个图必须有一个从 START 出发的边，定义图的执行起点。一个图可以有多个从 START 出发的边（并行起点）：

graph.add_edge(START, "fetch_data")
graph.add_edge(START, "fetch_context")
# fetch_data 和 fetch_context 并行执行

10.2 END（终止点）

from langgraph.graph import END

# 图的出口
graph.add_edge("last_node", END)

END 是虚拟出口节点。到达 END 意味着图执行结束，返回最终 State。条件边也可以返回 END 来提前终止。

---

下一篇：《中篇——高级特性篇》将讲解图编译执行、流式输出、持久化检查点、Human-in-the-Loop、多Agent工作流、ReAct Agent、错误处理等生产级能力。