LangGraph 教程：使用 Python 构建有状态的循环 AI 智能体

大语言模型 (LLM) 应用的形态正在从简单的“提示-响应”模式演变为具有自主能力的“有状态智能体” (Stateful Agents)。虽然 LangChain 在线性工作流方面表现出色，但在处理复杂的现实场景（如迭代编程、多步研究或自主邮件管理）时，开发者需要更强大的编排层。这就是 LangGraph 发挥作用的地方。

在本篇深度教程中，我们将探讨如何利用 LangGraph 构建有状态的 AI 智能体。我们还将展示集成来自 n1n.ai 的高性能 API 如何显著提升这些智能体的可靠性和响应速度。阅读完本指南后，你将拥有一个能够自主决策的邮件处理智能体。

什么是 LangGraph？为什么它至关重要？

LangGraph 是一个用于构建有状态、多角色 LLM 应用的开源库。它构建在 LangChain 之上，引入了 状态图 (State Graph) 的概念。与传统的有向无环图 (DAG) 不同，LangGraph 允许 循环 (Cycles)，这对于需要重复执行任务直到满足特定条件的智能体行为至关重要。

其核心优势包括：

持久化 (Persistence)：在每一步执行后自动保存智能体的状态。
循环能力 (Cycles)：支持自我纠错和逻辑推理的迭代循环。
人机协同 (Human-in-the-loop)：可以暂停执行以等待人工审批或输入。
多智能体支持：在单个图中协调多个专业智能体协同工作。

为了驱动这些复杂的流程，稳定的后端 API 是必不可少的。通过 n1n.ai，你可以轻松切换 Claude 3.5 Sonnet、OpenAI o3 或 DeepSeek-V3 等顶级模型，确保你的智能体始终拥有最适合任务的“大脑”。

核心概念：节点、边与状态

在编写代码之前，我们需要理解 LangGraph 应用的三大支柱：

状态 (State)：一个共享的数据结构，代表应用的当前状况。每个节点都可以读取并更新此状态。
节点 (Nodes)：工作流中的具体步骤或函数。节点可以是 LLM 调用、工具执行或一段 Python 代码。
边 (Edges)：定义了从一个节点移动到另一个节点的逻辑。条件边 (Conditional Edges) 允许图根据 LLM 的输出进行分支跳转。

环境配置

你需要 Python 3.9 或更高版本。我们强烈建议使用 n1n.ai 来获取 LLM 访问权限，因为它通过单一接口提供了多个模型，极大地简化了开发难度。

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows 用户使用 `venv\Scripts\activate`

# 安装依赖
pip install langgraph langchain-openai pydantic

设置环境变量（以 n1n.ai 为例）：

export OPENAI_API_KEY="你的_n1n_API密钥"
export OPENAI_API_BASE="https://api.n1n.ai/v1"

实战：构建自动邮件处理智能体

我们的目标是构建一个智能体，能够读取邮件、分类（如：技术支持、销售、垃圾邮件），然后根据分类结果起草回复或直接归档。

1. 定义状态 (State)

首先，定义图中需要追踪的信息。

from typing import TypedDict, List

class AgentState(TypedDict):
    email_content: str
    category: str
    draft_response: str
    steps_taken: List[str]

2. 定义节点 (Nodes)

每个节点都是一个接收 State 并返回更新后 State 的 Python 函数。

from langchain_openai import ChatOpenAI

# 通过 n1n.ai 初始化模型
llm = ChatOpenAI(model="gpt-4o", api_key="your_key", base_url="https://api.n1n.ai/v1")

def categorize_email(state: AgentState):
    content = state["email_content"]
    prompt = f"请对这封邮件进行分类：{content}。类别：技术支持、销售、垃圾邮件。"
    response = llm.invoke(prompt)
    return {"category": response.content, "steps_taken": ["已分类"]}

def draft_reply(state: AgentState):
    content = state["email_content"]
    prompt = f"针对以下内容起草一份礼貌的回复：{content}"
    response = llm.invoke(prompt)
    return {"draft_response": response.content, "steps_taken": ["已起草回复"]}

def archive_email(state: AgentState):
    return {"steps_taken": ["已归档"]}

3. 构建图逻辑

使用 StateGraph 将节点连接起来。

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("categorizer", categorize_email)
workflow.add_node("drafter", draft_reply)
workflow.add_node("archiver", archive_email)

# 设置入口点
workflow.set_entry_point("categorizer")

# 添加条件边
def route_email(state: AgentState):
    if "垃圾邮件" in state["category"]:
        return "archive"
    return "draft"

workflow.add_conditional_edges(
    "categorizer",
    route_email,
    {
        "archive": "archiver",
        "draft": "drafter"
    }
)

# 闭环连接
workflow.add_edge("drafter", END)
workflow.add_edge("archiver", END)

# 编译图
app = workflow.compile()

为什么 LangGraph 优于普通 Chain？

在标准的 LangChain Chain 中，流程是固定的。如果 LLM 在分类步骤出错，流程就会中断。而在 LangGraph 中，你可以添加循环。例如，如果 drafter 发现分类结果不合理，它可以将状态传回给 categorizer 重新处理。这种自我修复能力是真正的 AI 智能体与简单脚本的本质区别。

使用 n1n.ai 进行性能优化

在运行有状态智能体时，由于每个节点都需要调用 LLM，延迟会迅速累积。使用 n1n.ai 具有以下优势：

极低延迟：通过优化路由，将请求发送至最近的数据中心。
高可用性：支持多供应商自动故障切换。如果 OpenAI 接口响应慢，n1n.ai 可以自动切换至 Anthropic 节点。
成本透明：在一个看板中监控复杂图结构产生的所有 Token 消耗。

高级功能：状态持久化

LangGraph 支持 checkpointer。这意味着如果你的服务器在执行中途崩溃，智能体可以从停止的那个节点精确恢复。

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用 SQLite 存储状态
memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

# 通过 thread_id 标识不同的会话
config = {"configurable": {"thread_id": "user_888"}}
app.invoke({"email_content": "我的订单还没有收到，请核查"}, config)

总结

对于希望超越基础 LLM 脚本、构建生产级 AI 智能体的开发者来说，LangGraph 是必不可少的工具。通过利用状态图、循环和持久化，你可以构建出不仅更聪明、而且更具鲁棒性的系统。

为了确保你的智能体发挥出最佳性能，请务必选择高质量的 API 聚合服务。立即在 n1n.ai 获取免费 API Key。

参考来源：https://realpython.com/langgraph-python/