LangGraph 是一个用于构建有状态 Agent 的底层编排框架。它把 Agent 的执行过程描述成一张图：状态在节点之间流动，边决定下一步执行哪个节点，直到任务完成。

普通的聊天程序通常只有一次模型调用，而 Agent 会不断重复“思考、调用工具、读取结果、继续思考”的循环。随着流程变长，我们还会遇到分支、重试、持久化、人工审批和故障恢复等问题。LangGraph 的价值就是让这些控制逻辑变得显式、可追踪和可恢复。

LangGraph 与 LangChain

LangChain 和 LangGraph 解决的问题并不完全相同：

• LangChain 提供模型、Prompt、Tool、Retriever 和预构建 Agent 等高层抽象；
• LangGraph 提供状态图、持久化、流式执行和 Human-in-the-loop 等编排能力；
• LangSmith 用于追踪、评估和观察 Agent 的执行过程。

如果只是开发标准的 ReAct Agent，LangChain 的 create_agent 通常更简单。如果需要自定义执行路径、审批节点、并行任务或长时间运行的工作流，再直接使用 LangGraph。

核心概念

一张 LangGraph 主要由三个部分组成。

State

State 是整个工作流共享的数据。它可以保存消息、用户信息、中间结果、重试次数或审批状态。

节点不应该随意修改 State，而是返回需要更新的部分。LangGraph 会根据 State Schema 和 Reducer 合并更新。

Node

Node 是执行实际工作的函数，例如：

• 调用 LLM；
• 执行 Tool；
• 查询数据库；
• 检查权限；
• 等待人工审批。

节点读取当前 State，并返回一个局部 State 更新。

Edge

Edge 决定节点之间的执行顺序：

• 普通 Edge 固定跳转到下一个节点；
• Conditional Edge 根据 State 动态选择路径；
• START 和 END 分别表示图的入口和结束。

一个 Tool-calling Agent 的图通常如下：

START
  │
  ▼
Agent Node ─── 无工具调用 ───▶ END
  │
  │ 有工具调用
  ▼
Tool Node
  │
  └──────────────────────────▶ Agent Node

模型可以连续调用多个工具。只有当模型返回最终答案、不再产生 Tool Call 时，图才会进入 END。

开发一个 Agent

下面开发一个简单的天气 Agent。模型负责理解用户意图，Tool 负责返回天气信息，LangGraph 负责控制模型和 Tool 之间的循环。

安装依赖

python -m venv .venv
source .venv/bin/activate
pip install -U langgraph langchain langchain-openai

通过环境变量提供模型凭据：

export OPENAI_API_KEY="..."

不要把 API Key 写入代码或提交到版本控制。

定义 Tool

Tool 是 Agent 与外部世界交互的接口。真实项目中的 Tool 可以调用 API、数据库或内部服务；这里使用静态数据，使示例可以专注于 Agent 的执行逻辑。

from langchain.tools import tool


@tool
def get_weather(city: str) -> str:
    """查询城市当前的天气。"""
    weather = {
        "北京": "晴，26°C",
        "上海": "多云，29°C",
        "深圳": "阵雨，31°C",
    }
    return weather.get(city, f"暂时没有 {city} 的天气数据")

函数的类型标注和 Docstring 会成为 Tool Schema 的一部分。描述应该清晰说明工具能做什么，但不要让模型看到内部凭据或不需要的参数。

定义模型节点

将 Tool 绑定到模型后，模型可以选择返回普通消息，也可以返回结构化 Tool Call。

from langchain.chat_models import init_chat_model
from langgraph.graph import MessagesState


tools = [get_weather]
model = init_chat_model("openai:gpt-4.1-mini").bind_tools(tools)


def call_model(state: MessagesState):
    system_message = {
        "role": "system",
        "content": "你是一个天气助手。需要实时数据时调用工具，不要编造天气。",
    }
    response = model.invoke([system_message, *state["messages"]])
    return {"messages": [response]}

MessagesState 已经定义了消息列表及对应的 Reducer。节点只返回新消息，Reducer 会把它追加到历史记录，而不是覆盖整个列表。

构建执行图

ToolNode 负责执行模型产生的 Tool Call，tools_condition 则根据模型输出决定进入 Tool Node 还是结束工作流。

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition


builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")

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

StateGraph 只是构建器，调用 compile() 后才会得到可以执行的 Graph。这里加入了内存 Checkpointer，用于在同一个 Thread 中保存每一步的 State。

运行 Agent

config = {"configurable": {"thread_id": "demo-thread"}}

inputs = {
    "messages": [
        {"role": "user", "content": "上海和深圳哪个城市更热？"}
    ]
}

for state in graph.stream(inputs, config, stream_mode="values"):
    state["messages"][-1].pretty_print()

执行过程如下：

1. 用户消息进入 agent 节点；
2. 模型生成两个天气 Tool Call；
3. tools_condition 把执行路径路由到 tools；
4. ToolNode 执行工具并把结果写入消息列表；
5. 图重新进入 agent；
6. 模型比较工具结果并生成最终回答；
7. 没有新的 Tool Call，图进入 END。

多轮对话与持久化

Checkpointer 会按照 thread_id 保存 Graph State。下一次调用使用相同的 Thread ID，Agent 就能继续之前的对话：

follow_up = {
    "messages": [
        {"role": "user", "content": "那北京呢？"}
    ]
}

result = graph.invoke(follow_up, config)
print(result["messages"][-1].content)

InMemorySaver 适合本地开发和测试，进程退出后数据就会消失。生产环境应该使用持久化 Checkpointer，例如 PostgreSQL，或者使用自动管理持久化的 Agent Server。

Checkpoint 不只是聊天记忆，它还支持：

• 从失败前的成功步骤继续执行；
• 查看每个节点执行后的 State；
• 回到历史 Checkpoint 进行调试；
• 暂停工作流并等待人工输入。

Human-in-the-loop

当 Agent 准备执行发送消息、修改数据或产生费用等高风险操作时，可以使用 interrupt() 暂停 Graph：

from langgraph.types import Command, interrupt


def review_action(state):
    decision = interrupt(
        {
            "question": "是否允许执行该操作？",
            "action": state["pending_action"],
        }
    )
    return {"decision": decision}

Graph 会保存当前 State，并把审批请求返回给调用方。审批完成后，使用相同的 thread_id 恢复执行：

graph.invoke(Command(resume="approve"), config)

恢复时，包含 interrupt() 的节点会从头重新执行。因此，应把不可重复的副作用放在审批之后，并保证可能重试的节点具有幂等性。

生产实践

State 保持精简

State 应保存控制流程需要的数据，而不是无限增长的原始内容。长对话需要裁剪、摘要或外部存储，否则模型上下文和 Checkpoint 都会不断膨胀。

Tool 保持单一职责

每个 Tool 应完成一个清晰动作，并使用严格的参数类型。读操作和写操作最好分开，写操作可以单独接入审批节点。

处理错误与重试

外部 API 可能超时或限流。根据异常类型配置重试策略，不要让模型在未知错误上无限循环。工具错误应该返回足够的信息供模型调整，但不能泄露凭据和内部实现。

控制循环

Agent 可能因为错误的 Tool Call 不断循环。生产环境需要限制执行步数、调用次数和总超时时间，并为无法完成的任务提供明确的退出路径。

持久化前考虑幂等性

恢复和重试可能再次执行节点。支付、发送消息或写数据库等操作应使用幂等键，避免产生重复副作用。

增加可观测性

Agent 的问题通常不是单个函数报错，而是选择了错误路径或错误工具。应记录节点输入输出、Tool Call、延迟、Token 使用量和最终执行路径，并使用 LangSmith 等工具进行追踪和评估。

什么时候使用 LangGraph

适合直接使用 LangGraph 的场景包括：

• Agent 有明确的多步骤流程和条件分支；
• 需要暂停、恢复或长时间运行；
• 需要保存状态或跨会话记忆；
• 高风险 Tool 必须经过人工审批；
• 需要多个 Agent 或子图协作；
• 需要精确控制失败重试和执行路径。

如果需求只是“模型调用几个工具并回答问题”，优先从高层 Agent API 开始。只有当默认循环无法表达业务流程时，再下沉到 LangGraph。这样既能保留开发效率，也不会过早引入复杂度。

参考资料

• LangGraph overview
• Graph API overview
• Tools
• Persistence
• Human-in-the-loop