从零开始构建 LangGraph 智能体：全流程实战指南

作者
  • avatar
    姓名
    Nino
    职业
    Senior Tech Editor

大语言模型（LLM）应用的发展已经迅速从简单的提示词响应模式演变为复杂的多步链式调用。然而，传统的有向无环图（DAG）在构建需要迭代、错误修正和状态持久化的真正自主智能体时往往显得力不从心。这正是 LangGraph 的用武之地。作为 LangChain 生态系统的扩展，LangGraph 允许开发者创建具有循环计算步骤的有状态、多角色应用程序。在本指南中，我们将详细介绍如何从零开始构建一个 LangGraph 智能体，确保你拥有部署生产级 AI 服务所需的基础设施。

为了给这些智能体提供高速推理支持，开发者通常会选择 n1n.ai。通过使用 n1n.ai 作为你的 API 网关，你可以通过统一的计费和优化的延迟访问 GPT-4o、Claude 3.5 Sonnet 和 DeepSeek-V3 等顶尖模型，这对于我们即将构建的迭代循环至关重要。

理解 LangGraph 范式

与遵循线性路径的标准 LangChain 链不同，LangGraph 建立在状态机（State Machine）的概念之上。其核心组件包括：

  1. 状态 (State)：一个共享的数据结构，代表应用程序的当前状态。随着智能体在图中的移动，状态会不断更新。
  2. 节点 (Nodes)：函数或工作单元。在智能体工作流中，一个节点可能是 LLM 在决定做什么，另一个节点可能是工具执行器。
  3. 边 (Edges)：节点之间的连接。它们可以是固定的（始终从 A 到 B），也可以是条件性的（如果 LLM 说“完成”，则转到结束；否则，转到工具节点）。

第一步：环境配置

首先，我们需要安装必要的库。我们将使用 LangGraph 进行编排，使用 LangChain 进行模型抽象。

pip install langgraph langchain-openai langchain-community

为了确保你的智能体能够访问最佳模型，请使用来自 n1n.ai 的 API 密钥配置你的环境变量。使用 n1n.ai 的优势在于其极高的可用性和对多种主流模型的原生支持。

第二步：定义智能体状态

State 是 LangGraph 智能体的骨架。在 Python 中，我们通常使用 TypedDict 来定义。对于基础智能体，我们需要跟踪迄今为止交换的消息列表。

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # add_messages 函数告诉 LangGraph 将新消息追加到现有列表，而不是覆盖它们
    messages: Annotated[list, add_messages]

第三步：创建节点与集成工具

智能体需要“思考”（LLM）和“行动”（工具）。让我们使用 @tool 装饰器定义一个简单的工具。在这个例子中，我们将构建一个天气查询工具。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def get_weather(city: str):
    """调用此工具以获取特定城市的当前天气。"""
    if "london" in city.lower():
        return "20 摄氏度，多云。"
    return "25 摄氏度，晴朗。"

tools = [get_weather]
# 通过 n1n.ai 端点初始化模型
model = ChatOpenAI(model="gpt-4o", api_key="YOUR_N1N_API_KEY", base_url="https://api.n1n.ai/v1").bind_tools(tools)

接下来，我们定义节点。call_model 节点将调用 LLM，而 tool_node 将执行请求的工具。

from langgraph.prebuilt import ToolNode

def call_model(state: AgentState):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]}

tool_node = ToolNode(tools)

第四步：构建图结构

现在我们将这些组件组合成一个 StateGraph。这是我们定义智能体逻辑的地方：从模型开始，如果模型想要使用工具，则转到工具节点，然后再回到模型。

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)

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

# 定义条件边
def should_continue(state: AgentState):
    last_message = state['messages'][-1]
    # 如果没有工具调用，则结束会话
    if not last_message.tool_calls:
        return "end"
    return "continue"

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)

# 添加从 action 回到 agent 的边
workflow.add_edge("action", "agent")

# 编译图
app = workflow.compile()

第五步：高级功能与持久化

LangGraph 最强大的功能之一是 持久化 (Persistence)。通过添加 checkpointer，智能体可以跨不同会话记住对话状态。这对于长时间运行的 RAG（检索增强生成）任务或复杂的多轮推理至关重要。

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用内存数据库进行演示，生产环境建议使用 PostgreSQL
memory = SqliteSaver.from_conn_string(":memory:")
app_with_memory = workflow.compile(checkpointer=memory)

# 使用 thread_id 运行智能体
config = {"configurable": {"thread_id": "user_123"}}
input_message = {"messages": [("user", "伦敦的天气怎么样？")]}

for event in app_with_memory.stream(input_message, config):
    for value in event.values():
        print("接收到的消息:", value["messages"][-1].content)

性能基准与优化策略

在构建基于 LangGraph 的智能体时，延迟是主要的瓶颈。图中的每一个“跳跃”（hop）都需要一次 LLM 调用。为了优化性能，可以参考以下建议：

  • 流式输出 (Streaming)：降低用户感知的延迟。使用 app.stream() 实时展示中间结果。
  • 并行执行 (Parallelism)：利用 LangGraph 的扇出/扇入（fan-out/fan-in）能力加速工具执行。
  • 模型分层 (Model Tiering)：通过 n1n.ai 灵活切换模型。使用较小的模型（如 GPT-4o-mini）进行简单路由，使用大型模型（如 o3 或 Claude 3.5）进行复杂推理，从而平衡成本与性能。

专家建议：人工介入 (Human-in-the-Loop)

对于企业级应用，你可能希望在智能体执行某些工具（特别是涉及财务或数据库写入操作）之前进行审核。LangGraph 原生支持“中断”（interrupts）。你可以配置图在进入 action 节点前暂停，允许人工批准或修改工具调用参数。

总结

从零开始构建 LangGraph 智能体为你的 AI 应用逻辑和状态提供了无与伦比的控制力。通过超越线性链式结构，你可以创建具有自我修正能力和复杂推理能力的鲁棒智能体。为了确保你的智能体能够以极低的延迟（首字延迟 < 100ms）运行，请务必选择像 n1n.ai 这样可靠的 API 聚合平台。

立即在 n1n.ai 获取免费 API 密钥。

参考来源：https://towardsdatascience.com/building-a-langgraph-agent-from-scratch/