LangGraph 入门指南：构建有状态的 AI Agent

LangGraph 是由 LangChain 团队开发的低级编排框架，专门用于构建、管理和部署长时间运行的有状态 Agent。它被 Klarna、Replit、Elastic 等公司广泛使用，为复杂的 Agent 系统提供了强大的基础设施。

目录

什么是 LangGraph？

LangGraph 是一个专注于 Agent 编排的低级框架，提供了构建复杂 AI 系统所需的核心能力。与高级抽象不同，LangGraph 不会隐藏提示词或架构细节，而是为开发者提供完全的控制权。

核心优势

LangGraph 为长时间运行的有状态工作流提供了以下关键能力：

• 持久化执行：构建能够从故障中恢复的 Agent，可以长时间运行并从中断处继续执行
• 人机协作：在任何时间点检查和修改 Agent 状态，实现人工监督
• 全面的记忆系统：创建具有短期工作记忆和跨会话长期记忆的有状态 Agent
• 调试能力：通过 LangSmith 可视化工具深入了解复杂的 Agent 行为
• 生产就绪：可扩展的基础设施，专为处理有状态、长时间运行的工作流而设计

快速开始

安装

首先安装 LangGraph：

pip install langgraph

Hello World 示例

让我们从一个简单的示例开始：

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

def mock_llm(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()

graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})

构建计算器 Agent

让我们通过一个实际的例子来学习 LangGraph 的核心概念。我们将构建一个能够执行数学运算的 Agent。

第一步：定义工具和模型

from langchain.tools import tool
from langchain.chat_models import init_chat_model

model = init_chat_model(
    "claude-sonnet-4-5-20250929",
    temperature=0
)

# 定义工具
@tool
def multiply(a: int, b: int) -> int:
    """将 a 和 b 相乘
    
    Args:
        a: 第一个整数
        b: 第二个整数
    """
    return a * b

@tool
def add(a: int, b: int) -> int:
    """将 a 和 b 相加
    
    Args:
        a: 第一个整数
        b: 第二个整数
    """
    return a + b

@tool
def divide(a: int, b: int) -> float:
    """将 a 除以 b
    
    Args:
        a: 第一个整数
        b: 第二个整数
    """
    return a / b

# 为 LLM 绑定工具
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

第二步：定义状态

图的状态用于存储消息和 LLM 调用次数：

from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

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

第三步：定义模型节点

模型节点用于调用 LLM 并决定是否调用工具：

from langchain.messages import SystemMessage

def llm_call(state: dict):
    """LLM 决定是否调用工具"""
    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="你是一个有用的助手，负责对一组输入执行算术运算。"
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

第四步：定义工具节点

工具节点用于执行工具调用并返回结果：

from langchain.messages import ToolMessage

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}

第五步：定义结束逻辑

条件边函数用于根据 LLM 是否进行了工具调用来路由到工具节点或结束：

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

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """根据 LLM 是否进行了工具调用来决定是否继续循环"""
    messages = state["messages"]
    last_message = messages[-1]
    
    # 如果 LLM 进行了工具调用，则执行操作
    if last_message.tool_calls:
        return "tool_node"
    
    # 否则，停止（回复用户）
    return END

第六步：构建和编译 Agent

# 构建工作流
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 = agent_builder.compile()

运行 Agent

from langchain.messages import HumanMessage

messages = [HumanMessage(content="计算 3 加 4")]
result = agent.invoke({"messages": messages})

for m in result["messages"]:
    m.pretty_print()

LangGraph 的思维方式

在使用 LangGraph 构建 Agent 时，需要遵循一个清晰的思维过程。让我们通过一个客户支持邮件 Agent 的例子来理解这个过程。

第一步：将流程映射为离散步骤

首先识别流程中的不同步骤，每个步骤将成为一个节点（执行特定任务的函数）。然后，勾画这些步骤如何相互连接。

对于客户支持邮件 Agent，我们需要：

1. 读取邮件：提取和解析邮件内容
2. 分类意图：使用 LLM 对紧急程度和主题进行分类
3. 文档搜索：查询知识库获取相关信息
4. Bug 跟踪：在跟踪系统中创建或更新问题
5. 起草回复：生成适当的响应
6. 人工审核：将复杂问题升级给人工处理
7. 发送回复：发送邮件响应

第二步：确定每个步骤需要做什么

对于图中的每个节点，确定它代表什么类型的操作：

• LLM 步骤：需要理解、分析、生成文本或做出推理决策
• 数据步骤：需要从外部源检索信息
• 操作步骤：需要执行外部操作
• 用户输入步骤：需要人工干预

第三步：设计状态

状态是所有节点都可以访问的共享内存。可以把它看作是 Agent 用来跟踪工作过程中学到和决定的一切的笔记本。

什么应该放在状态中？

• 原始邮件和发件人信息（以后无法重建）
• 分类结果（多个后续节点需要）
• 搜索结果和客户数据（重新获取成本高）
• 草稿响应（需要在审核过程中保持）
• 执行元数据（用于调试和恢复）

保持状态原始，按需格式化提示词

这种分离意味着：

• 不同的节点可以根据需要以不同方式格式化相同的数据
• 可以更改提示词模板而不修改状态模式
• 调试更清晰 - 可以准确看到每个节点接收到的数据

让我们定义状态：

from typing import TypedDict, Literal

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
    
    # 原始搜索/API 结果
    search_results: list[str] | None
    customer_history: dict | None
    
    # 生成的内容
    draft_response: str | None
    messages: list[str] | None

第四步：构建节点

现在将每个步骤实现为函数。LangGraph 中的节点就是一个接受当前状态并返回更新的 Python 函数。

适当处理错误

不同的错误需要不同的处理策略：

添加重试策略以自动重试网络问题和速率限制：

from langgraph.types import RetryPolicy

workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
)

第五步：连接起来

现在将节点连接成一个工作图。由于节点处理自己的路由决策，我们只需要几个基本的边。

要启用带有 interrupt() 的人机协作，需要使用 checkpointer 编译以在运行之间保存状态：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import RetryPolicy

# 创建图
workflow = StateGraph(EmailAgentState)

# 添加节点
workflow.add_node("read_email", read_email)
workflow.add_node("classify_intent", classify_intent)
workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3)
)
workflow.add_node("bug_tracking", bug_tracking)
workflow.add_node("draft_response", draft_response)
workflow.add_node("human_review", human_review)
workflow.add_node("send_reply", send_reply)

# 添加基本边
workflow.add_edge(START, "read_email")
workflow.add_edge("read_email", "classify_intent")
workflow.add_edge("send_reply", END)

# 使用 checkpointer 编译以实现持久化
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

工作流与 Agent 的区别

LangGraph 支持构建工作流和 Agent，理解它们的区别很重要：

工作流

工作流具有预定的代码路径，设计为按特定顺序运行。常见的工作流模式包括：

提示词链：每个 LLM 调用处理前一个调用的输出。常用于：

• 将文档翻译成不同语言
• 验证生成内容的一致性

并行化：LLM 同时处理任务。常用于：

• 拆分子任务并并行运行以提高速度
• 多次运行任务以检查不同输出

路由：处理输入然后将其定向到特定上下文的任务。例如，处理产品相关问题的工作流可能首先处理问题类型，然后将请求路由到定价、退款、退货等特定流程。

编排器-工作者：编排器分解任务、委派给工作者并综合输出。LangGraph 通过 Send API 内置支持此模式。

Agent

Agent 是动态的，定义自己的流程和工具使用。它们在持续的反馈循环中运行，用于问题和解决方案不可预测的情况。Agent 比工作流具有更多的自主权，可以决定使用哪些工具以及如何解决问题。

from langchain.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """将 a 和 b 相乘"""
    return a * b

# 为 LLM 绑定工具
tools = [add, multiply, divide]
llm_with_tools = llm.bind_tools(tools)

关键要点

构建这个计算器 Agent 展示了 LangGraph 的思维方式：

1. 分解为离散步骤：每个节点做好一件事。这种分解支持流式进度更新、可以暂停和恢复的持久化执行，以及清晰的调试。

2. 状态是共享内存：存储原始数据，而不是格式化文本。这让不同的节点可以以不同的方式使用相同的信息。

3. 节点是函数：它们接受状态、执行工作并返回更新。当需要做出路由决策时，它们同时指定状态更新和下一个目的地。

4. 错误是流程的一部分：瞬态故障获得重试，LLM 可恢复的错误循环回去并带有上下文，用户可修复的问题暂停等待输入，意外错误冒泡以进行调试。

5. 图结构自然呈现：定义基本连接，节点处理自己的路由逻辑。这使控制流保持明确和可追踪。

下一步

在本指南中，我们介绍了 LangGraph 的基础知识和核心概念。在下一篇文章中，我们将深入探讨：

• 持久化：如何保存和恢复 Agent 状态
• 流式输出：实时获取 Agent 执行的更新
• 人机协作：在执行过程中暂停并获取人工输入
• 记忆系统：跨会话保持上下文

参考资源

• LangGraph 官方文档
• LangGraph GitHub 仓库
• LangChain 文档

内容基于 LangGraph 官方文档改编，遵循内容许可限制