LangGraph 深度技术解析：

本文基于 LangGraph 官方文档（对 LangGraph 的架构设计、核心概念、高级特性、工作流模式和生态体系进行全面解析。

---

图表总览

1. LangChain 生态分层架构

2. 核心架构三要素

3. Agent 工具调用循环流程

4. Pregel 超级步执行模型

5. 持久化双系统架构

6. 人在回路中断与恢复流程

7. 时间旅行: 重放与分叉

8. 多 Agent 架构模式对比

9. 五种工作流模式速查

10. 框架能力对比矩阵

能力维度	LangGraph	OpenAI Agents SDK	Google ADK	AutoGen	CrewAI
图结构编排	■■■■■	■■□□□	■■■□□	■□□□□	■■□□□
状态管理	■■■■■	■■□□□	■■□□□	■□□□□	■■□□□
人在回路	■■■■■	■■□□□	■■□□□	■■■□□	■■□□□
持久化执行	■■■■■	□□□□□	□□□□□	□□□□□	□□□□□
时间旅行	■■■■■	□□□□□	□□□□□	□□□□□	□□□□□
流式处理	■■■■□	■■■□□	■■□□□	■■□□□	■■□□□
多 Agent	■■■■□	■■■□□	■■■□□	■■■■□	■■■■■
生产就绪	■■■■■	■■■□□	■■□□□	■■■□□	■■■□□
学习曲线	■■■■■ （较陡）	■■□□□ （简单）	■■■□□ （中等）	■■□□□ （简单）	■■□□□ （简单）

■ 代表能力强度，满分为 5 格。学习曲线中 ■ 越多表示学习成本越高。

11. 错误处理决策树

12. 持久化执行三种模式

性能： exit > async > sync　|　持久性： sync > async > exit

---

概述

LangGraph 是由 LangChain Inc. 开发的低层级编排框架与运行时，专门用于构建、管理和部署长时间运行的、有状态的 AI Agent。与 LangChain 不同，LangGraph 不抽象提示词或架构，而是专注于 Agent 编排的底层能力：持久化执行、流式处理、人在回路和内存管理。^[1][2]

截至 2026 年，LangGraph 已被 Klarna、Uber、J.P. Morgan、Replit、Elastic 等企业用于生产环境，GitHub 上获得 41,100+ 个项目依赖，累计 6,992 次提交，最新版本为 1.2.7。^[1]

LangGraph 的设计灵感来源于 Google 的 Pregel 大规模图处理系统和 Apache Beam 数据处理框架，公共接口设计借鉴了 NetworkX 图计算库。它由 LangChain 团队构建，但可以独立于 LangChain 使用。^[2]

在 LangChain 生态中的定位

LangChain 产品体系采用分层架构，LangGraph 处于编排运行时层：

层级	产品	职责
Agent 框架	Deep Agents	规划、子代理、文件系统工具、上下文管理
组件框架	LangChain	模型集成、工具、Agent 循环的抽象与组件
编排运行时	LangGraph	持久化执行、流式处理、人在回路、内存
平台	LangSmith	追踪、评估、提示管理、部署

^[2]

核心架构

LangGraph 的核心思想是将 Agent 工作流建模为有向图。图由三个关键组件构成：State（状态）、Node（节点）和 Edge（边）。通过组合节点和边，可以创建具有循环和条件分支的复杂工作流，状态在执行过程中不断演进。^[9]

State：共享状态

State 是 LangGraph 中所有节点共享的数据结构，代表应用程序在某一时刻的快照。State 可以是 TypedDict、dataclass 或 Pydantic BaseModel，但通常使用 TypedDict。^[9]

State 的设计原则是存储原始数据，按需格式化提示词。这意味着不同节点可以用不同方式格式化同一数据，提示词模板的修改不影响 State 结构，调试时能清楚看到每个节点接收到的原始数据。^[4]

from typing import TypedDict, Literal
from typing_extensions import Annotated
import operator

class EmailClassification(TypedDict):
    intent: Literal["question", "bug", "billing", "feature", "complex"]
    urgency: Literal["low", "medium", "high", "critical"]
    topic: str
    summary: str

class EmailAgentState(TypedDict):
    # 原始邮件数据
    email_content: str
    sender_email: str
    email_id: str
    # 分类结果
    classification: EmailClassification | None
    # 原始搜索结果
    search_results: list[str] | None
    customer_history: dict | None
    # 生成的内容
    draft_response: str | None
    messages: list[str] | None

^[4]

Reducer：状态更新策略

Reducer 是理解 LangGraph 状态管理的关键。State 中的每个键都有独立的 reducer 函数，决定节点返回的更新如何应用到 State 上。如果没有显式指定 reducer，默认行为是覆盖——新值直接替换旧值。^[9]

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    foo: int                    # 默认 reducer：覆盖
    bar: Annotated[list[str], add]  # 自定义 reducer：追加

在上面的例子中，如果初始状态为 {"foo": 1, "bar": ["hi"]}，第一个节点返回 {"foo": 2}，状态变为 {"foo": 2, "bar": ["hi"]}。第二个节点返回 {"bar": ["bye"]}，状态变为 {"foo": 2, "bar": ["hi", "bye"]}——bar 使用了 operator.add 将两个列表合并。^[9]

对于消息列表，LangGraph 提供了内置的 add_messages reducer，它不仅能追加新消息，还能根据消息 ID 覆盖已有消息，并支持将字典自动反序列化为 LangChain Message 对象。^[9]

多 Schema 支持

LangGraph 支持为图定义独立的输入 Schema、输出 Schema 和私有 State Schema，提供更精细的数据控制：^[9]

• InputState：限制图的输入字段
• OutputState：限制图的输出字段
• OverallState：包含所有内部状态字段的完整 Schema
• PrivateState：节点间内部通信使用的私有通道

builder = StateGraph(
    OverallState,
    input_schema=InputState,
    output_schema=OutputState
)

Node：执行节点

Node 是编码 Agent 逻辑的 Python 函数。它接收当前 State 作为输入，执行计算或副作用操作，返回 State 更新。节点可以包含 LLM 调用，也可以只是普通代码——“节点做工作，边决定下一步做什么”。^[9]

def llm_call(state: dict):
    """LLM 决定是否调用工具"""
    return {
        "messages": [
            model_with_tools.invoke(
                [SystemMessage(content="你是一个有用的助手。")]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

def tool_node(state: dict):
    """执行工具调用"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

^[3]

每个节点应遵循幂等性原则——当工作流恢复时，已完成的节点不会重复执行，但如果某个任务启动后未完成，恢复时会重新运行该任务。建议使用幂等操作或幂等键来避免重复写入。^[7]

Edge：边与路由

Edge 决定下一个执行哪个节点，分为两种类型：^[9]

• 普通边：固定路由，节点 A 完成后总是执行节点 B
• 条件边：根据当前 State 动态选择下一个节点

from typing import Literal

def should_continue(state) -> Literal["tool_node", "__end__"]:
    """根据 LLM 是否发起工具调用决定下一步"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return "__end__"

# 构建图
agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_edge("__start__", "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", "__end__"])
agent_builder.add_edge("tool_node", "llm_call")
agent = agent_builder.compile()

^[3]

Command：节点内部控制

LangGraph 1.x 引入了 Command 原语，允许节点在返回时同时更新状态和指定下一步路由，而不仅仅返回 State 更新：^[9]

from langgraph.types import Command

def classify_intent(state) -> Command[Literal["search", "human_review", "draft"]]:
    classification = structured_llm.invoke(prompt)
    
    if classification['urgency'] == 'critical':
        goto = "human_review"
    elif classification['intent'] in ['question', 'feature']:
        goto = "search"
    else:
        goto = "draft"
    
    return Command(
        update={"classification": classification},
        goto=goto
    )

^[4]

Pregel 执行模型

LangGraph 底层使用基于消息传递的图算法，受 Google Pregel 启发。程序以离散的"超级步"（super-step）执行：同一超级步中的节点并行运行，不同超级步的节点顺序执行。每个超级步开始时，所有节点处于非活跃状态；当节点通过入边接收到新消息时变为活跃状态，执行函数并产生更新。超级步结束时，没有入边消息的节点投票暂停。当所有节点都非活跃且没有消息在途时，图执行终止。^[9][11]

两种 API 风格

LangGraph 提供两种构建 Agent 的 API 风格，适用于不同的开发偏好和场景复杂度。

Graph API

Graph API 是 LangGraph 的传统接口，通过显式定义节点和边来构建图结构。适合需要清晰可视化执行路径、复杂分支和并行处理的场景。^[3]

from langgraph.graph import StateGraph, START, END

# 定义 State
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

# 构建工作流
agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")

# 编译并调用
agent = agent_builder.compile()
result = agent.invoke({"messages": [HumanMessage(content="Add 3 and 4.")]})

^[3]

Functional API

Functional API 是 LangGraph 1.0 引入的接口，使用 @entrypoint 和 @task 装饰器，允许在单个函数内使用标准控制流（循环、条件）来定义 Agent。适合对现有代码改动最小、快速迭代的场景。^[3]

from langgraph.func import entrypoint, task

@task
def call_llm(messages: list[BaseMessage]):
    """LLM 决定是否调用工具"""
    return model_with_tools.invoke(
        [SystemMessage(content="你是一个有用的助手。")] + messages
    )

@task
def call_tool(tool_call: ToolCall):
    """执行工具调用"""
    tool = tools_by_name[tool_call["name"]]
    return tool.invoke(tool_call)

@entrypoint()
def agent(messages: list[BaseMessage]):
    model_response = call_llm(messages).result()

    while True:
        if not model_response.tool_calls:
            break

        # 并行执行工具调用
        tool_result_futures = [
            call_tool(tool_call) for tool_call in model_response.tool_calls
        ]
        tool_results = [fut.result() for fut in tool_result_futures]
        messages = add_messages(messages, [model_response, *tool_results])
        model_response = call_llm(messages).result()

    messages = add_messages(messages, model_response)
    return messages

^[3]

API 选择指南

维度	Graph API	Functional API
编程模型	显式节点和边	标准控制流（循环、条件）
可视化	支持图可视化	无内置可视化
适合场景	复杂分支、并行处理	快速原型、现有代码迁移
持久化	通过 checkpointer	通过 @task 装饰器
人在回路	通过 interrupt	通过 interrupt

高级特性

持久化与内存

LangGraph 提供两套互补的持久化系统：^[5]

Checkpointer（检查点器） 负责持久化线程的图状态快照，用于短期、线程范围内的内存管理，支持会话连续性、人在回路工作流、时间旅行和容错恢复。

Store（存储器） 持久化图状态之外的应用数据，用于长期、跨线程的内存管理，如用户偏好、事实和共享知识。

维度	Checkpointer	Store
持久化内容	图状态快照	应用定义的键值数据
作用范围	单个线程	跨线程
内存类型	短期、线程内	长期、跨线程
适用场景	会话连续性、人在回路、时间旅行	用户偏好、事实、共享知识
访问方式	在配置中传 thread_id	从节点或应用代码读写

^[5]

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()
store = InMemoryStore()

graph = builder.compile(checkpointer=checkpointer, store=store)

result = graph.invoke(
    {"messages": [{"role": "user", "content": "你好，我叫张三。"}]},
    {"configurable": {"thread_id": "thread-1"}},
)

^[5]

生产环境应使用持久化存储后端，如 PostgresSaver（PostgreSQL，支持异步）或 SqliteSaver（本地文件，适合开发环境）。InMemorySaver 仅存储在内存中，进程重启后数据丢失。^[5]

人在回路

LangGraph 通过 interrupt() 函数实现动态人在回路，允许在图执行的任意位置暂停并等待外部输入。与静态断点不同，interrupt 是动态的——可以放置在代码中的任何位置，并且可以基于应用逻辑进行条件触发。^[6]

interrupt 的工作流程：^[6]

1. 调用 interrupt() 时，图执行在精确位置暂停
2. Checkpointer 保存当前图状态
3. 值通过 stream.interrupts 返回给调用方
4. 图无限期等待，直到使用 Command(resume=...) 恢复执行
5. 恢复值成为 interrupt() 调用的返回值

from langgraph.types import interrupt, Command

def approval_node(state) -> Command[Literal["proceed", "cancel"]]:
    # 暂停执行，请求审批
    is_approved = interrupt({
        "question": "是否批准此操作？",
        "details": state["action_details"]
    })

    if is_approved:
        return Command(goto="proceed")
    else:
        return Command(goto="cancel")

恢复执行时，传递 True 批准或 False 拒绝：^[6]

# 批准
graph.stream_events(Command(resume=True), config=config, version="v3").output

# 拒绝
graph.stream_events(Command(resume=False), config=config, version="v3").output

常见人在回路模式：

• 审批工作流：在执行关键操作（API 调用、数据库变更、金融交易）前暂停
• 审查和编辑：让人类审查和修改 LLM 输出或工具调用
• 工具调用中断：在执行工具调用前暂停以审查和编辑
• 输入验证：在进入下一步前验证人类输入
• 多中断处理：并行分支同时触发中断时，通过中断 ID 映射恢复值

^[6]

interrupt 的重要规则：不要将 interrupt() 调用包裹在 try/except 中；不要在节点内重排 interrupt 调用顺序；不要返回复杂值；interrupt 之前的副作用必须是幂等的。^[6]

持久化执行

持久化执行是一种在关键节点保存进程进度的技术，使工作流能够在暂停后从精确位置恢复，适用于需要人在回路的场景和可能遇到中断的长时间运行任务。^[7]

LangGraph 提供三种持久化模式，允许在性能和数据一致性之间取得平衡：^[7]

模式	描述	性能	持久性
"exit"	仅在图执行退出时持久化	最高	最低，无法从中间故障恢复
"async"	异步持久化，下一步执行时并行保存	较高	较好，进程崩溃时有小概率丢失
"sync"	同步持久化，下一步前先写入检查点	较低	最高，每个检查点都写入后才继续

graph.stream(
    {"input": "test"},
    durability="sync"  # 最高持久性模式
)

^[7]

确定性重放 是持久化执行的核心要求。当恢复工作流时，代码不会从停止的同一行恢复，而是从合适的起始点重新执行到停止点。因此，非确定性操作和有副作用的操作必须包裹在 @task 或节点中，以确保恢复时不会重复执行，而是从持久化层获取结果。^[7]

优雅关闭（langgraph >= 1.2）允许在当前超级步完成后停止运行中的图并保存可恢复的检查点，适用于处理 SIGTERM 信号或外部资源回收：^[7]

from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained

control = RunControl()
signal.signal(signal.SIGTERM, lambda *_: control.request_drain("sigterm"))

try:
    result = graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    # 图已提前停止并保存了检查点
    # 下次启动时用相同 config 恢复
    pass

时间旅行

LangGraph 通过检查点机制支持时间旅行，包括重放（Replay）和分叉（Fork）两种模式：^[8]

重放：从先前的检查点重新执行。节点会重新执行——不是从缓存读取。LLM 调用、API 请求和 interrupt 都会重新触发，可能产生不同结果。

# 获取状态历史
history = list(graph.get_state_history(config))

# 找到 write_joke 之前的检查点
before_joke = next(s for s in history if s.next == ("write_joke",))

# 从该检查点重放
replay_result = graph.invoke(None, before_joke.config)

^[8]

分叉：从先前的检查点创建新分支并修改状态。update_state 不会回滚线程，而是创建一个从指定点分支的新检查点，原始执行历史保持不变。

# 从 write_joke 之前分叉，修改 topic
fork_config = graph.update_state(
    before_joke.config,
    values={"topic": "chickens"},
)

# 从分叉点继续执行
fork_result = graph.invoke(None, fork_config)
print(fork_result["joke"])  # 关于鸡的笑话，而非原始袜子笑话

^[8]

工作流与 Agent 模式

LangGraph 将 AI 应用分为两类：^[10]

• 工作流：具有预定代码路径，按特定顺序运行
• Agent：动态定义自己的流程和工具使用

提示词链

每个 LLM 调用处理前一个调用的输出，常用于执行可分解为更小、可验证步骤的任务。^[10]

# Graph API 实现
workflow = StateGraph(State)
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
    "generate_joke", check_punchline,
    {"Fail": "improve_joke", "Pass": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)
chain = workflow.compile()

^[10]

并行化

多个 LLM 同时处理任务，可以是运行多个独立子任务（提高速度），也可以是多次运行同一任务检查不同输出（提高置信度）。^[10]

# Graph API：通过从 START 到多个节点的边实现并行
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
# 三个节点并行执行，结果在 aggregator 汇聚
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")

^[10]

路由

处理输入并将其导向特定任务。例如，产品问题工作流先处理问题类型，然后将请求路由到定价、退款、退货等特定流程。^[10]

def route_decision(state: State):
    if state["decision"] == "story":
        return "llm_call_1"
    elif state["decision"] == "joke":
        return "llm_call_2"
    elif state["decision"] == "poem":
        return "llm_call_3"

router_builder.add_conditional_edges(
    "llm_call_router",
    route_decision,
    {"llm_call_1": "llm_call_1", "llm_call_2": "llm_call_2", "llm_call_3": "llm_call_3"},
)

^[10]

编排器-工作者

编排器将任务分解为子任务，委托给工作者执行，然后综合工作者输出为最终结果。当子任务无法像并行化那样预定义时，此模式提供了更大灵活性。^[10]

评估器-优化器

一个 LLM 生成响应，另一个 LLM 评估该响应并提供反馈，循环改进直到满足质量标准。

Agent（工具调用循环）

Agent 是最灵活的模式——LLM 自主决定使用哪些工具、何时使用，以及何时完成任务。核心是 LLM 与工具之间的循环调用：^[10]

# 核心循环：LLM 调用 → 是否调用工具？→ 是则执行工具 → 回到 LLM 调用
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")  # 工具执行后回到 LLM

^[10]

多 Agent 架构

LangGraph 支持多种多 Agent 编排模式，适用于不同复杂度的协作场景。

监督者模式

中央监督者 Agent 协调多个专业 Agent，控制所有通信流和任务分配。监督者根据当前上下文和任务需求决定调用哪个 Agent。^[15]

from langgraph.graph import StateGraph, START, END

# 监督者节点：根据状态决定路由到哪个专业 Agent
def supervisor(state) -> Command[Literal["researcher", "coder", "writer", "__end__"]]:
    # LLM 分析当前状态，决定下一步
    decision = supervisor_llm.invoke(...)
    return Command(goto=decision.next_agent)

# 专业 Agent 节点
def researcher(state):
    """研究 Agent：搜索信息"""
    ...
    return Command(goto="supervisor")  # 返回监督者

def coder(state):
    """编码 Agent：编写代码"""
    ...
    return Command(goto="supervisor")

# 构建图
builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.add_edge(START, "supervisor")
graph = builder.compile(checkpointer=checkpointer)

^[15]

层级模式

多层监督者嵌套，高层监督者管理低层监督者，低层监督者再管理具体 Agent。适用于组织结构复杂的大型系统。

网络模式

Agent 之间直接通信，没有中央协调者。每个 Agent 可以调用其他 Agent，适用于高度协作的场景。

模式选择

模式	适用场景	优势	劣势
监督者	明确分工的协作	控制清晰、易于调试	监督者成为瓶颈
层级	大型复杂系统	可扩展性好	实现复杂
网络	高度协作场景	灵活性高	难以预测和调试

错误处理策略

LangGraph 提供分层错误处理机制，不同类型的错误采用不同策略：^[4]

错误类型	处理者	策略	适用场景
瞬态错误（网络、限流）	系统自动	重试策略	临时故障，重试通常可解决
LLM 可恢复错误	LLM	存储错误到状态并循环	LLM 看到错误后调整策略
用户可修复错误	人类	interrupt() 暂停	需要用户输入才能继续
重试耗尽后的可恢复故障	开发者	error_handler	运行补偿/恢复分支
未知错误	开发者	让其冒泡	需要调试的未知问题

from langgraph.types import RetryPolicy

# 自动重试策略
workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
)

# 错误处理器（langgraph >= 1.2）
from langgraph.errors import NodeError

def payment_error_handler(state, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

workflow.add_node(
    "charge_payment",
    charge_payment,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
    error_handler=payment_error_handler,
)

^[4]

与其他 Agent 框架对比

2025-2026 年间，AI Agent 框架经历了第一轮洗牌，LangGraph、OpenAI Agents SDK 和 Google ADK 成为主流选择。^[12]

维度	LangGraph	OpenAI Agents SDK	Google ADK	AutoGen	CrewAI
架构模型	有向图	线性 Agent	组件化	对话驱动	角色分工
流程控制	图结构，循环+条件分支	线性 Handoff	模块组合	对话轮次	角色任务
状态管理	内置 Checkpointer + Store	有限	有限	无内置	无内置
人在回路	原生 interrupt	有限	有限	对话级	任务级
持久化执行	原生支持	不支持	不支持	不支持	不支持
时间旅行	支持	不支持	不支持	不支持	不支持
生产就绪	高（Klarna、Uber 等）	中	低	中	中
适合场景	复杂决策链、生产系统	OpenAI 生态快速原型	Google 生态	人机协作对话	团队式任务

^[12][16]

LangGraph 的核心优势在于低层级控制力——它不抽象提示词或架构，而是提供底层编排基础设施。这使得开发者可以精确控制 Agent 的每一步执行，但同时也意味着更高的学习成本和开发复杂度。^[2]

Agent 框架的演化路径体现了从简单到复杂的趋势：LangChain（组件组合）→ LangGraph（图编排，2025 年将 Agent 推入生产环境）→ DeepAgents（Agent 自管理，2026 年新兴方向）。^[13]

生态体系

LangSmith

LangSmith 提供 Agent 的可观测性和评估能力。通过追踪执行路径、捕获状态转换和提供运行时指标，LangSmith 为复杂 Agent 行为提供了深度可视化。设置 LANGSMITH_TRACING=true 和 API Key 即可开始追踪。^[1]

LangSmith Deployment

LangSmith Deployment 是专门为长时间运行、有状态工作流设计的部署平台，支持可扩展的基础设施，提供 Studio 进行可视化原型设计。^[1]

Deep Agents

Deep Agents 是建立在 LangGraph 之上的高层 Agent 框架，提供规划、子代理、文件系统工具和上下文管理能力。如果需要快速构建 Agent 而非从头编排，Deep Agents 是推荐选择。^[1]

LangGraph Studio

LangGraph Studio 提供可视化界面来原型、调试和部署 Agent，支持实时查看图执行路径、状态转换和节点输出。

实战示例：客服邮件 Agent

以下是一个完整的客服邮件处理 Agent 示例，综合运用了 State 设计、条件路由、人在回路和错误处理：^[4]

from typing import Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command, RetryPolicy
from langgraph.checkpoint.memory import InMemorySaver

# 1. 定义 State
class EmailClassification(TypedDict):
    intent: Literal["question", "bug", "billing", "feature", "complex"]
    urgency: Literal["low", "medium", "high", "critical"]
    topic: str
    summary: str

class EmailAgentState(TypedDict):
    email_content: str
    sender_email: str
    classification: EmailClassification | None
    search_results: list[str] | None
    draft_response: str | None
    status: str | None

# 2. 定义节点
def read_email(state: EmailAgentState) -> dict:
    """读取并解析邮件"""
    return {"status": "reading"}

def classify_intent(state) -> Command[Literal["search", "human_review", "draft", "bug_tracking"]]:
    """使用 LLM 分类邮件意图和紧急程度"""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"分析这封客户邮件: {state['email_content']}"
    )
    
    if classification['intent'] == 'billing' or classification['urgency'] == 'critical':
        goto = "human_review"
    elif classification['intent'] in ['question', 'feature']:
        goto = "search"
    elif classification['intent'] == 'bug':
        goto = "bug_tracking"
    else:
        goto = "draft"
    
    return Command(update={"classification": classification}, goto=goto)

def search_documentation(state) -> Command[Literal["draft"]]:
    """搜索知识库"""
    try:
        results = knowledge_base.search(state['classification']['topic'])
    except SearchAPIError as e:
        results = [f"搜索暂时不可用: {str(e)}"]
    
    return Command(update={"search_results": results}, goto="draft")

def draft_response(state) -> Command[Literal["human_review", "send_reply"]]:
    """生成回复草稿"""
    classification = state.get('classification', {})
    response = llm.invoke(build_prompt(state))
    
    needs_review = (
        classification.get('urgency') in ['high', 'critical'] or
        classification.get('intent') == 'complex'
    )
    
    goto = "human_review" if needs_review else "send_reply"
    return Command(update={"draft_response": response.content}, goto=goto)

def human_review(state) -> Command[Literal["send_reply", END]]:
    """人在回路：人工审查"""
    decision = interrupt({
        "instruction": "审查并编辑回复内容",
        "content": state["draft_response"],
        "classification": state["classification"],
    })
    
    if decision.get("approved"):
        return Command(
            update={"draft_response": decision.get("edited_content", state["draft_response"])},
            goto="send_reply"
        )
    else:
        return Command(update={"status": "rejected"}, goto=END)

def send_reply(state: EmailAgentState):
    """发送回复"""
    email_service.send(state["draft_response"])
    return {"status": "sent"}

# 3. 构建并编译图
builder = StateGraph(EmailAgentState)
builder.add_node("read_email", read_email)
builder.add_node("classify_intent", classify_intent)
builder.add_node("search", search_documentation)
builder.add_node("bug_tracking", bug_tracking)
builder.add_node("draft", draft_response)
builder.add_node("human_review", human_review)
builder.add_node("send_reply", send_reply)

builder.add_edge(START, "read_email")
builder.add_edge("read_email", "classify_intent")
builder.add_edge("bug_tracking", "draft")
builder.add_edge("send_reply", END)

# 添加重试策略
# builder.add_node("search", search_documentation,
#     retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0))

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

# 4. 调用
config = {"configurable": {"thread_id": "email-001"}}
result = graph.invoke(
    {"email_content": "我被重复收费了！", "sender_email": "user@example.com"},
    config=config
)

# 如果触发了人在回路中断，恢复执行：
# from langgraph.types import Command
# graph.invoke(Command(resume={"approved": True, "edited_content": "修改后的回复"}), config=config)

^[4]

设计最佳实践

State 设计

• 存储原始数据，按需格式化提示词，而非存储格式化文本^[4]
• 只存储不可派生的数据：如果数据可以从其他数据计算得出，按需计算而非存储^[4]
• 使用 Annotated 指定 Reducer：对于列表类状态（如消息），使用 Annotated[list, add] 避免覆盖^[9]
• 分离输入/输出 Schema：当字段较多时，使用 input_schema 和 output_schema 约束图的接口^[9]

节点设计

• 单一职责：每个节点只做一件事^[4]
• 幂等性：节点应设计为幂等的，以支持持久化执行的重放^[7]
• 包裹副作用：非确定性操作和副作用应包裹在 @task 中^[7]
• 错误分层处理：瞬态错误用重试，LLM 可恢复错误存状态并循环，用户错误用 interrupt^[4]

生产部署

• 使用持久化 Checkpointer：生产环境使用 PostgresSaver 而非 InMemorySaver^[5]
• 控制 checkpoint 增长：长时间对话中检查点会累积，定期清理或设置保留策略^[5]
• thread_id 管理：保持 thread_id 在 255 字符以内，使用 UUID 或哈希^[5]
• 启用 LangSmith 追踪：设置 LANGSMITH_TRACING=true 获得执行可视化^[1]
• 合理选择持久化模式：性能敏感场景用 "async"，数据一致性要求高用 "sync"^[7]

总结

LangGraph 作为低层级编排框架，其核心价值在于为有状态、长时间运行的 AI Agent 提供了完整的底层基础设施。通过图结构建模工作流，LangGraph 实现了循环、条件分支、并行执行等复杂控制流，同时通过检查点机制保证了持久化执行、人在回路和时间旅行等高级能力。

与 AutoGen 的对话驱动、CrewAI 的角色分工不同，LangGraph 基于图结构强调流程控制，适合复杂决策链和生产环境部署。^[16]在 2025-2026 年的工程实践中，LangGraph 已成为复杂 Agent 场景的首选框架，它解决了早期 LangChain 的核心痛点：天生支持循环与图结构，对应用流程和状态有细粒度控制，内置持久化功能支持跨交互保持上下文。^[14]

随着 Deep Agents 等高层框架的出现，LangGraph 的定位更加清晰——它不试图简化 Agent 开发，而是提供构建任何 Agent 所需的底层编排能力。对于需要精确控制、生产级可靠性和复杂工作流的团队，LangGraph 是当前最成熟的选择。