从零搭建一个 Agent：LangChain + LangGraph 实战

最终要做出的东西

这篇文章不从一个封装好的 create_agent() 开始，而是用 LangGraph 的 Graph API 手工搭出 Agent 的运行回路。这样做多写了几行代码，却能真正看清模型、工具、状态、路由和记忆分别发生在哪里。

完成后，这个 Agent 能够：

• 理解用户的自然语言请求；
• 自主判断是否需要调用工具；
• 查询一份本地天气数据并完成加法、乘法计算；
• 把工具结果交回模型，再组织成最终回答；
• 用同一个 thread_id 保留多轮对话上下文。

示例只依赖模型 API，不依赖外部天气服务，因此很适合先把 Agent 的骨架跑通，再逐步替换成真实业务工具。

LangChain 和 LangGraph 各自负责什么

在这个项目中，两者不是替代关系，而是上下分层：

组件	本文中的职责
LangChain	初始化聊天模型、定义工具、描述消息，让不同模型供应商尽量使用统一接口
LangGraph	保存状态、编排节点、设置条件边、执行循环并接入检查点记忆

LangChain 官方把工具定义为带有明确输入与输出的可调用函数，模型会根据上下文决定何时调用、传什么参数。LangGraph 则把工作流表达为状态、节点和边，适合需要循环、分支、持久化和人工介入的有状态 Agent。可分别参考 LangChain Tools 文档 与 LangGraph Graph API 概览。

Agent 的核心是一个可终止循环

先把“大模型会思考”这种模糊表述放在一边。从程序视角看，本文的 Agent 就是下面这个循环：

START
  ↓
调用模型 ──没有工具请求──→ END
  │
  └──产生工具请求──→ 执行工具 ──→ 把结果追加到消息 ──→ 再次调用模型

模型负责决策，工具负责执行，LangGraph 负责保证状态沿图流动。任何循环都必须有退出条件：当最后一条模型消息不再包含工具调用时，图就结束并把回答交给用户。

第一步：创建项目与安装依赖

创建一个干净目录和虚拟环境：

mkdir langgraph-agent-demo
cd langgraph-agent-demo

python -m venv .venv
source .venv/bin/activate

pip install -U langgraph "langchain[anthropic]"

本文采用 Anthropic 作为可运行示例。配置 API Key 和模型名：

export ANTHROPIC_API_KEY="替换为你的 API Key"
export MODEL_NAME="claude-sonnet-4-6"

如果你使用 OpenAI、Google Gemini 或其他供应商，只需安装对应的 LangChain 集成包，设置该供应商要求的环境变量，并把 MODEL_NAME 换成支持工具调用的模型。init_chat_model() 提供了统一入口，官方支持的供应商和初始化方式可查看 LangChain Models 文档。

第二步：定义 Agent 可以使用的工具

新建 agent.py，先定义三个工具：

from langchain.tools import tool


@tool
def get_weather(city: str) -> str:
    """查询指定城市的演示天气。

    Args:
        city: 城市中文名，例如上海、北京或深圳。
    """
    weather_data = {
        "上海": "晴，31°C，东南风 2 级",
        "北京": "多云，28°C，北风 3 级",
        "深圳": "阵雨，30°C，湿度 82%",
    }
    return weather_data.get(city, f"暂时没有 {city} 的天气数据")


@tool
def add(a: int, b: int) -> int:
    """计算两个整数之和。

    Args:
        a: 第一个整数。
        b: 第二个整数。
    """
    return a + b


@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数之积。

    Args:
        a: 第一个整数。
        b: 第二个整数。
    """
    return a * b

@tool 会根据函数签名生成输入 Schema，并把 docstring 作为工具说明。类型标注决定参数结构，说明文字则帮助模型判断工具何时适用，所以二者都不是装饰品。

真实项目中的工具可以查询数据库、调用内部 HTTP API、读写工单或发送消息，但有三个原则不要丢：

1. 一个工具只承担一个明确动作；
2. 输入、输出尽量结构化且可校验；
3. 不要把密钥、数据库连接等敏感信息暴露给模型参数。

第三步：初始化模型并绑定工具

继续在 agent.py 中加入模型配置：

import os

from langchain.chat_models import init_chat_model


MODEL_NAME = os.getenv("MODEL_NAME", "claude-sonnet-4-6")

model = init_chat_model(
    MODEL_NAME,
    temperature=0,
    timeout=30,
    max_retries=2,
)

tools = [get_weather, add, multiply]
model_with_tools = model.bind_tools(tools)

bind_tools() 并不会立即执行工具。它只是把工具名称、说明和参数 Schema 提供给模型。模型返回工具调用请求后，真正的函数执行仍由后面的 ToolNode 完成。

这里把 temperature 设为 0，是为了让演示结果更稳定；timeout 和 max_retries 则避免一次网络抖动让进程无限等待。生产系统还应在工具层分别配置超时、重试与熔断，而不是只依赖模型客户端。

第四步：定义模型节点

LangGraph 自带的 MessagesState 适合以消息为核心状态的 Agent。每个节点读取当前状态，返回需要合并到状态中的增量：

from langchain.messages import SystemMessage
from langgraph.graph import MessagesState


SYSTEM_PROMPT = """
你是一个简洁、可靠的中文助手。
需要事实或计算结果时必须使用可用工具，不要编造工具返回值。
拿到工具结果后，给出结论，并简要说明使用了哪些数据。
""".strip()


def call_model(state: MessagesState) -> dict:
    response = model_with_tools.invoke(
        [
            SystemMessage(content=SYSTEM_PROMPT),
            *state["messages"],
        ]
    )
    return {"messages": [response]}

这个节点只做一件事：把系统提示词和当前消息历史交给模型，再把模型响应追加回 messages。它既可能返回普通回答，也可能返回一个或多个工具调用请求。

第五步：连接节点与条件边

接下来把模型节点、工具节点和路由组合成图：

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, handle_tool_errors=True))

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

这里最关键的是 tools_condition：

• 最后一条模型消息包含工具调用时，路由到名为 tools 的节点；
• 不包含工具调用时，路由到 END；
• 工具执行完后，经固定边回到 agent，让模型读取工具结果并继续决策。

ToolNode 是 LangGraph 提供的预构建节点，能够把工具返回值转换为工具消息，并处理一次响应中的多个工具调用。相关用法见 LangChain ToolNode 说明。

第六步：加入线程级短期记忆

如果直接 builder.compile()，一次调用结束后，下一次调用不会自动继承上一次对话。开发阶段可以用内存检查点保存同一线程的状态：

from langgraph.checkpoint.memory import InMemorySaver


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

检查点与 thread_id 配合使用。相同 thread_id 表示同一段对话，不同 ID 则彼此隔离。官方记忆文档也明确区分了线程级短期记忆与跨会话长期记忆，详见 LangGraph Memory 文档。

InMemorySaver 只适合本地开发：进程退出后数据就会消失，多进程也不能共享。生产环境应换成数据库支持的 checkpointer，例如 PostgreSQL 实现。

第七步：运行一次完整对话

在文件末尾加入调用代码：

def ask(question: str, thread_id: str = "demo-user-1") -> str:
    result = agent.invoke(
        {
            "messages": [
                {"role": "user", "content": question},
            ]
        },
        {
            "configurable": {"thread_id": thread_id},
            "recursion_limit": 12,
        },
    )
    return str(result["messages"][-1].content)


if __name__ == "__main__":
    print(ask("上海天气怎么样？另外帮我计算 23 乘以 17。"))
    print(ask("把刚才的乘积再加 10。"))

运行：

python agent.py

第一轮中，模型通常会调用 get_weather 与 multiply，得到天气和 391；第二轮复用了同一个 thread_id，因此模型能从历史消息中拿到 391，再调用 add 得到 401。

recursion_limit 是很重要的保险丝。如果模型与工具因为错误路由反复循环，达到上限后图会停止，而不是无休止消耗请求额度。

完整代码

下面是可以直接保存为 agent.py 的完整版本：

import os

from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import MessagesState, START, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition


@tool
def get_weather(city: str) -> str:
    """查询指定城市的演示天气。

    Args:
        city: 城市中文名，例如上海、北京或深圳。
    """
    weather_data = {
        "上海": "晴，31°C，东南风 2 级",
        "北京": "多云，28°C，北风 3 级",
        "深圳": "阵雨，30°C，湿度 82%",
    }
    return weather_data.get(city, f"暂时没有 {city} 的天气数据")


@tool
def add(a: int, b: int) -> int:
    """计算两个整数之和。

    Args:
        a: 第一个整数。
        b: 第二个整数。
    """
    return a + b


@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数之积。

    Args:
        a: 第一个整数。
        b: 第二个整数。
    """
    return a * b


MODEL_NAME = os.getenv("MODEL_NAME", "claude-sonnet-4-6")

model = init_chat_model(
    MODEL_NAME,
    temperature=0,
    timeout=30,
    max_retries=2,
)

tools = [get_weather, add, multiply]
model_with_tools = model.bind_tools(tools)

SYSTEM_PROMPT = """
你是一个简洁、可靠的中文助手。
需要事实或计算结果时必须使用可用工具，不要编造工具返回值。
拿到工具结果后，给出结论，并简要说明使用了哪些数据。
""".strip()


def call_model(state: MessagesState) -> dict:
    response = model_with_tools.invoke(
        [SystemMessage(content=SYSTEM_PROMPT), *state["messages"]]
    )
    return {"messages": [response]}


builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools, handle_tool_errors=True))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")

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


def ask(question: str, thread_id: str = "demo-user-1") -> str:
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        {
            "configurable": {"thread_id": thread_id},
            "recursion_limit": 12,
        },
    )
    return str(result["messages"][-1].content)


if __name__ == "__main__":
    print(ask("上海天气怎么样？另外帮我计算 23 乘以 17。"))
    print(ask("把刚才的乘积再加 10。"))

如何验证它不是“碰巧能跑”

Agent 含有模型决策，端到端输出不应只用字符串完全匹配来测试。建议把测试拆成三层：

1. 工具单元测试：输入固定参数，断言确定结果；
2. 图路由测试：用可控的假模型分别返回“调用工具”和“直接结束”；
3. 端到端评测：检查是否选对工具、参数是否正确、最终答案是否包含关键事实。

最简单的工具测试如下：

def test_math_tools() -> None:
    assert multiply.invoke({"a": 23, "b": 17}) == 391
    assert add.invoke({"a": 391, "b": 10}) == 401

不要只测试最后一句自然语言。即使最终回答看起来正确，也可能是模型绕过工具直接猜出来的；调试时需要同时观察模型消息中的 tool_calls 与工具返回的 ToolMessage。

从演示升级到生产系统

这个 Agent 已具备最小闭环，但距离生产可用还差几项工程能力：

• 真实数据源：把演示天气字典替换为带鉴权、超时和响应校验的 API 客户端。
• 持久化检查点：使用数据库 checkpointer，保证重启恢复和多实例共享状态。
• 权限边界：查询类工具可以自动执行，付款、删除、发信等高风险工具应加入人工确认节点。
• 幂等设计：会产生副作用的工具必须携带业务幂等键，防止重试造成重复写入。
• 上下文治理：长对话不能无限追加消息，需要裁剪、摘要或把稳定信息写入长期存储。
• 可观测性：记录节点耗时、模型 token、工具参数、异常与重试，但要过滤密钥和个人信息。
• 失败策略：区分模型超时、工具失败、参数错误和权限不足，给每类错误设计明确的恢复路径。
• 效果评测：为真实任务建立数据集，持续评估工具选择、参数正确率、任务完成率和成本。

LangGraph 的价值并不是把简单调用画成图，而是把“下一步做什么”变成显式、可测试、可恢复的控制流。当业务需要审批、重试、并行工具、子图或长时间运行任务时，这种显式结构会比隐藏在一个大函数里的循环更容易维护。

常见问题与排查顺序

模型从不调用工具：先检查模型是否支持工具调用，再检查 docstring 是否清楚描述使用场景，最后确认确实调用了 bind_tools()。

工具执行后没有最终回答：检查是否存在从 tools 返回 agent 的边，以及工具结果是否作为消息写回状态。

不同用户看到了彼此上下文：不要使用固定 thread_id 服务所有请求。它必须与经过认证的会话或用户对话 ID 绑定。

Agent 一直循环：设置 recursion_limit，检查系统提示词是否要求不可能完成的动作，并确认工具错误不会诱导模型无休止重试。

本地有记忆，重启后丢失：这是 InMemorySaver 的预期行为；部署时换成数据库 checkpointer。

下一步可以怎么扩展

掌握这个最小图之后，可以按同一思路逐步增加能力：

1. 给工具接入真实 API 或数据库；
2. 用自定义状态保存用户身份、权限和任务阶段；
3. 在危险操作前插入人工审批与恢复节点；
4. 把长任务拆成规划、执行、校验三个子图；
5. 接入流式事件，让前端展示“正在调用哪个工具”；
6. 用持久化 Store 保存跨线程的用户偏好与业务知识。

如果只记住一句话：LangChain 负责把模型和工具接进来，LangGraph 负责让状态沿着可控路径运行。一个可靠的 Agent，不是提示词越长越好，而是状态明确、工具受控、循环可终止、失败可恢复。

参考资料

• LangGraph Quickstart
• LangGraph Overview
• LangGraph Memory
• LangChain Models
• LangChain Tools