## 最终要做出的东西

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

完成后，这个 Agent 能够：

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

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

## LangChain 和 LangGraph 各自负责什么

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

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

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

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

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

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

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

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

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

```bash
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 和模型名：

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

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

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

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

```python
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` 中加入模型配置：

```python
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。每个节点读取当前状态，返回需要合并到状态中的增量：

```python
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`。它既可能返回普通回答，也可能返回一个或多个工具调用请求。

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

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

```python
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 说明](https://docs.langchain.com/oss/python/langchain/tools#toolnode)。

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

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

```python
from langgraph.checkpoint.memory import InMemorySaver


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

检查点与 `thread_id` 配合使用。相同 `thread_id` 表示同一段对话，不同 ID 则彼此隔离。官方记忆文档也明确区分了线程级短期记忆与跨会话长期记忆，详见 [LangGraph Memory 文档](https://docs.langchain.com/oss/python/langgraph/add-memory)。

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

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

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

```python
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。"))
```

运行：

```bash
python agent.py
```

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

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

## 完整代码

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

```python
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. 端到端评测：检查是否选对工具、参数是否正确、最终答案是否包含关键事实。

最简单的工具测试如下：

```python
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](https://docs.langchain.com/oss/python/langgraph/quickstart)
- [LangGraph Overview](https://docs.langchain.com/oss/python/langgraph/overview)
- [LangGraph Memory](https://docs.langchain.com/oss/python/langgraph/add-memory)
- [LangChain Models](https://docs.langchain.com/oss/python/langchain/models)
- [LangChain Tools](https://docs.langchain.com/oss/python/langchain/tools)