写在前面

你有没有过这样的体验：让 AI 帮你查个天气，如果下雨就取消明天的户外预约。结果它只会一句一句地回答你，你得自己查天气、自己判断、自己取消预约，然后把结果告诉它。

这让我感到一种深深的割裂——明明大模型已经如此聪明，为什么它不能"一气呵成"地把事情做完？

直到我遇见了 Agent（智能代理）。

这份笔记记录了我从零开始理解 Agent 的完整旅程：它是什么、怎么造、怎么用好。如果你也在学习 AI 应用开发，希望这些文字能成为你路上的一盏灯。

---

一、觉醒：从 Chain 到 Agent

1.1 一条走不通的"流水线"

在学 Agent 之前，我已经会用 Chain（链） 来组合 LLM 调用了。Chain 的工作方式很直觉——预先定义好一条固定的处理流程，输入数据像流水线上的零件一样，按顺序流经每一步，最终得到输出：

用户输入 → 提示词模板 → LLM → 输出解析 → 结果
└─────────────────────────────┘
        固定流程，预先编排

说实话，Chain 在很多场景下用起来挺舒服的。翻译一段文字、总结一篇文章、提取关键信息……这些流程明确的任务，Chain 简单又可靠。

但现实世界不是流水线。

用户问：“帮我查查北京明天会不会下雨，如果下雨，帮我取消明天的户外预约。”

这个任务里藏着一个致命的问题：步骤二要不要执行，取决于步骤一的结果。 你在写代码的时候，根本不知道明天会不会下雨，也就无法预先确定流程该走哪条路。

这就是 Chain 的天花板——它只能走事先铺好的轨道，不能在岔路口自己做选择。

1.2 Agent：让 LLM 当自己的"司机"

Agent 的核心思想，一句话就能说清：

把 LLM 当作"大脑"，让它自己决定下一步做什么。

用公式来表达：

Agent = LLM + 自主决策 + 工具调用

在 Chain 中，是你（开发者）决定执行流程；在 Agent 中，是 LLM 自己决定执行流程。它会根据用户的输入和当前上下文，自主选择要调用哪个工具、执行什么操作，然后观察结果，再决定是否继续行动。

这个区别，像极了"你告诉司机怎么走"和"你告诉司机去哪儿，他自己找路"的差异。

对比维度	Chain（链）	Agent（智能代理）
谁决定流程	开发者在代码中预定义	LLM 在运行时自主决定
执行路径	线性的，固定的	循环的，动态的
工具调用	在固定位置调用固定工具	LLM 按需选择，调用次数不确定
处理意外情况	无法应对预期之外的情况	可根据工具返回结果调整策略
适合的任务	结构明确、步骤固定的任务	开放式、需要判断和决策的任务
可预测性	高——每次路径相同	低——不同输入可能走不同路径
开发复杂度	低——流程清晰可控	较高——需设计好工具和提示词

实践经验： 不要所有场景都用 Agent。如果任务流程是确定的（比如"翻译一段文字"），用 Chain 更简单可靠。只有当任务需要动态判断和多步决策时，才需要 Agent。杀鸡焉用牛刀。

1.3 Agent 的四个核心组件

Agent 不是一块铁板，它由四个协作运转的组件构成：

┌──────────────────────────────────────────────┐
│                  Agent 系统                    │
│                                              │
│   ┌──────────┐    指导     ┌──────────┐      │
│   │ Planning │ ─────────→ │   LLM    │      │
│   │  (规划)   │            │  (大脑)   │      │
│   └──────────┘            └────┬─────┘      │
│                               │ 决策          │
│   ┌──────────┐    反馈     ┌───▼──────┐     │
│   │  Memory  │ ←──────── │  Tools   │      │
│   │  (记忆)   │            │  (工具)   │      │
│   └──────────┘            └──────────┘      │
└──────────────────────────────────────────────┘

• LLM（大脑）：理解用户意图，做出每一步的决策，是整个 Agent 的中枢。
• Tools（工具）：执行 LLM 的决策，与外部世界交互。LLM “想”，工具 “做”。
• Memory（记忆）：存储上下文，让 Agent "记住"对话历史。
• Planning（规划）：把复杂任务分解成可执行的步骤，指导 LLM 按合理顺序调用工具。

但注意： 最简单的 Agent 只需要 LLM + Tools 就能工作。Memory 和 Planning 是增强能力。

Planning 去哪了？

你可能在论文里见过 Agent 的四组件架构（比如 Lilian Weng 那篇著名的 Agent 论文）。但在 LangChain 的工程实现中，Planning 并不是一个独立的物理组件，而是一种隐含在 Agent 运行机制和提示词中的"能力"。你不需要手写 planning = ... 这样的代码。

Planning 有两种主要体现方式：

方式一：边走边看（ReAct 模式）——LangChain 中最基础、最常用的模式。

用户："帮我查北京天气，如果下雨就取消明天的会议"

第1步局部规划：我需要先查天气 → 调用天气工具
第2步局部规划：天气是下雨，所以下一步取消会议 → 调用日程工具
第3步局部规划：两件事都做完了 → 回复用户

没有全局规划，而是步步为营。每走一步都想一想"接下来该干什么"。

方式二：谋定而后动（Plan-and-Execute 模式）——适合特别复杂的任务。

用户："调研中美欧三地的AI市场，写一份对比报告"

阶段一（Planner）：先纯思考，输出步骤清单
  → 步骤1：搜索中国AI市场数据
  → 步骤2：搜索美国AI市场数据
  → 步骤3：搜索欧洲AI市场数据
  → 步骤4：对比分析三地差异
  → 步骤5：撰写报告

阶段二（Executor）：按清单逐一执行

这种模式需要 LangGraph 来构建更复杂的工作流。

总结： 在 LangChain 开发中，你实际需要动手构建的核心组件是三个：LLM、Tools、Memory。Planning 的灵魂隐藏在你选择的 Agent 类型和系统提示词中。

1.4 工作机制：感知-推理-行动的循环

Agent 不是"一次调用就出结果"，而是通过一个**循环（Loop）**来逐步完成任务：
保存下来直接上传](https://img-home.csdnimg.cn/images/20230724024159.png?origin_url=img_agent_loop.jpg&pos_id=img-UPYSqrjv-1783904487023)

用前面那个"查天气取消预约"的例子，看看 Agent 内部是怎么一步步处理的：

第 1 轮循环：

• 感知：收到用户消息"帮我查查北京明天会不会下雨，如果下雨，帮我取消明天的户外预约"
• 推理：用户想知道天气，我需要先查天气 → 决定调用"天气查询"工具
• 行动：调用 get_weather("北京", "明天") → 返回"明天北京：小雨"

第 2 轮循环：

• 感知：收到工具返回结果"小雨"
• 推理：明天下雨，用户要求下雨时取消户外预约 → 决定调用"取消预约"工具
• 行动：调用 cancel_appointment("户外预约", "明天") → 返回"已取消"

第 3 轮循环：

• 感知：收到工具返回结果"已取消"
• 推理：天气查了，预约也取消了，任务完成 → 决定直接回复用户
• 输出：“北京明天预报有小雨，我已经帮您取消了明天的户外预约。”

仔细体会这个过程的精妙之处： 整个过程中，没有任何一行代码预先规定了"先查天气再取消预约"这个流程。是 LLM 根据用户意图和中间结果，自主决定了每一步该做什么。如果明天不下雨，LLM 在第 2 轮就会直接回复用户，根本不会调用取消预约的工具。

这就是 Agent 和 Chain 的根本区别：执行路径不是写死的，而是由 LLM 动态决定的。

---

二、造手：工具定义与使用

2.1 工具的本质——Function Calling

没有工具的 Agent，就像一个只会说话但没有手脚的人，什么实际操作也完成不了。

在深入 LangChain 之前，我们先理解一个底层机制：工具调用的本质是大模型的 Function Calling 能力。

当你给 Agent 配置工具时，实际上发生的事情是这样的：

① 你定义工具（函数名 + 参数说明 + 功能描述）
                    ↓
② LangChain 把工具信息转换成 JSON Schema，随提示词一起发给 LLM
                    ↓
③ LLM 阅读工具描述，根据用户问题决定：需要调用哪个工具、传入什么参数
                    ↓
④ LLM 返回一个"工具调用指令"（不是直接执行，而是告诉框架"我要调用XX工具"）
                    ↓
⑤ LangChain 框架接收指令，在本地执行对应的函数
                    ↓
⑥ 执行结果返回给 LLM，LLM 继续推理

关键理解： LLM 自身并不能执行任何工具。它只是根据工具描述"选择"要调用什么、传什么参数。真正执行工具的是你的代码。LLM 的角色更像是一个"调度员"。

这意味着一件至关重要的事：工具的描述写得好不好，直接决定了 LLM 能不能正确地选择和调用它。

2.2 原生 Function Calling——繁重的手工活

为了理解 LangChain 在背后帮我们做了什么，先看看不用 LangChain 时直接用 OpenAI SDK 实现的 Function Calling：

from openai import OpenAI
import json

client = OpenAI()

# ===== 第1步：用 JSON Schema 手动描述工具 =====
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市在指定日期的天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称",
                    },
                    "date": {
                        "type": "string",
                        "description": "日期，格式为YYYY-MM-DD",
                    }
                },
                "required": ["city", "date"],
                "additionalProperties": False,
            },
            "strict": True,
        },
    },
]

# ===== 第2步：定义工具的实际执行逻辑 =====
def get_weather(city, date):
    # 实际项目中这里会调用真实的天气API
    return f"{city} 在 {date} 天气多云，有下雨的可能性。"

# ===== 第3步：把用户消息和工具描述一起发给LLM =====
messages = [{"role": "user", "content": "北京2025-12-25的天气怎么样？"}]
response = client.chat.completions.create(
    model="gpt-4.1",
    messages=messages,
    tools=tools,
)

# ===== 第4步：LLM返回的不是文字，而是"工具调用指令" =====
# ===== 第5步：我们在本地执行工具，把结果喂回给LLM =====
messages.append(response.choices[0].message)
for tool_call in response.choices[0].message.tool_calls or []:
    if tool_call.function.name == "get_weather":
        args = json.loads(tool_call.function.arguments)
        result = get_weather(args["city"], args["date"])
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps({"weather": result}),
        })

# ===== 第6步：LLM拿到工具结果后，生成最终的自然语言回复 =====
final_response = client.chat.completions.create(
    model="gpt-4.1", messages=messages, tools=tools,
)
print(final_response.choices[0].message.content)

观察这段代码，你会发现手动实现 Function Calling 非常繁琐：要手写 JSON Schema、要手动解析工具调用指令、要手动把结果喂回去、要手动管理消息列表……如果有多个工具、多轮调用，代码量会爆炸式增长。

2.3 LangChain 的优雅解法——@tool 装饰器

LangChain 提供了 @tool 装饰器，只需要写一个普通的 Python 函数，加上装饰器和类型注解，LangChain 就会自动帮你完成所有脏活：

from langchain.tools import tool

@tool
def get_weather(city: str, date: str) -> str:
    """获取指定城市在指定日期的天气。
    
    Args:
        city: 城市名称，如"北京"、"上海"
        date: 日期，格式为YYYY-MM-DD
    """
    # 实际项目中调用天气API，这里用模拟数据
    return f"{city} 在 {date} 天气多云，有下雨的可能性。"

对比一下：原生方式需要手写十几行 JSON Schema 来描述一个工具，LangChain 只需要一个 @tool 装饰器加上规范的 docstring。效果完全一样——LangChain 会在背后自动生成 JSON Schema 发给 LLM。

三个影响 LLM 调用准确性的关键点

要素	作用	写法建议
函数名	LLM 根据函数名初步判断工具用途	用清晰的动词+名词，如 get_weather、search_documents
docstring	LLM 根据描述理解工具具体功能	写清楚"这个工具做什么"，越具体越好
参数类型注解	LLM 根据类型和描述决定传入什么值	每个参数都要有类型注解和说明

常见错误： docstring 写得太简略（如"查天气"），导致 LLM 不确定什么时候该用这个工具、该传什么参数。docstring 是你和 LLM 之间的"说明书"，写得越清楚，LLM 用得越准。

用 Pydantic 定义复杂参数

当工具的参数比较复杂时，可以用 Pydantic 模型来定义参数结构，提供更精确的约束：

from langchain.tools import tool
from pydantic import BaseModel, Field

class GetWeatherArgs(BaseModel):
    """天气查询参数"""
    city: str = Field(description="城市名称，如'北京'、'上海'")
    date: str = Field(description="查询日期，格式为YYYY-MM-DD")

@tool(args_schema=GetWeatherArgs)
def get_weather(city: str, date: str) -> str:
    """获取指定城市在指定日期的天气预报"""
    return f"{city} 在 {date} 天气多云，有下雨的可能性。"

2.4 用 create_agent 组装 Agent

工具定义好之后，就可以用 LangChain 的 create_agent 函数来构建一个完整的 Agent 了。它会帮你处理所有底层细节——消息管理、工具调用解析、结果回传、循环控制。

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch

# 第1步：初始化 LLM
llm = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 第2步：准备工具列表（自定义工具 + 第三方工具）
search = TavilySearch(max_results=5)  # 第三方搜索工具
tools = [get_weather, search]

# 第3步：创建 Agent
agent = create_agent(
    model=llm,           # 指定 LLM 作为大脑
    tools=tools,         # 传入工具列表
    system_prompt="你是一个智能助手，请根据用户的需求调用合适的工具来帮助他们。",
)

create_agent 的核心参数：

参数	作用	是否必填
model	指定 LLM（Agent 的大脑）	必填
tools	工具列表（Agent 的手脚）	必填
system_prompt	系统提示词，指导 Agent 的行为风格	可选
checkpointer	记忆存储（后面详细讲）	可选
middleware	中间件列表（后面详细讲）	可选

2.5 两种调用方式

invoke（一次性调用）——等待 Agent 完成所有推理和工具调用后，一次性返回最终结果：

result = agent.invoke(
    {"messages": [{"role": "user", "content": "今天北京的天气怎么样？"}]}
)
print(result["messages"][-1].content)

stream（流式调用）——实时输出 Agent 每一步的中间过程，适合需要展示"Agent 正在思考/执行"的场景：

for step in agent.stream(
    {"messages": [{"role": "user", "content": "今天北京的天气怎么样？"}]}
):
    print(step, end="\n\n")

流式调用的输出会依次展示：LLM 推理过程 → 工具调用 → 工具返回结果 → LLM 最终回复。

开发建议： 调试阶段用 stream 可以观察 Agent 每一步在干什么，方便排查问题；生产环境中根据产品形态选择——聊天界面适合流式，后台任务适合一次性调用。

2.6 一个完整的可运行例子

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langchain_tavily import TavilySearch

# ===== 1. 定义自定义工具 =====
@tool
def calculate(expression: str) -> str:
    """计算数学表达式的结果。
    
    Args:
        expression: 数学表达式，如 "2 + 3 * 4"、"100 / 7"
    """
    try:
        result = eval(expression)
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算出错：{e}"

@tool
def get_current_date() -> str:
    """获取当前日期和时间，不需要任何参数。"""
    from datetime import datetime
    return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

# ===== 2. 使用第三方工具 =====
search = TavilySearch(max_results=3)

# ===== 3. 初始化 LLM =====
llm = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# ===== 4. 创建 Agent =====
agent = create_agent(
    model=llm,
    tools=[calculate, get_current_date, search],
    system_prompt="""你是一个智能助手，拥有以下能力：
- 计算数学表达式
- 查询当前日期时间
- 搜索网络信息
请根据用户的问题，选择合适的工具来回答。如果不需要工具就能回答，直接回答即可。""",
)

# ===== 5. 运行 Agent =====
for i, step in enumerate(agent.stream(
    {"messages": [{"role": "user", "content": "今天是几号？帮我算一下距离2026年五一还有多少天"}]}
), start=1):
    print(f"=== 第 {i} 步 ===")
    print(step, end="\n\n")

在这个例子中，Agent 会自主完成以下步骤（无需我们编码控制流程）：

1. 调用 get_current_date 获取今天日期
2. 自己计算天数差，或调用 calculate 来辅助计算
3. 组织语言回复用户

2.7 LangSmith——Agent 的"X光机"

Agent 的执行过程是动态的，有时候出了问题很难排查——比如 LLM 选错了工具、传错了参数、或者陷入了无限循环。LangSmith 是 LangChain 官方提供的追踪调试工具，可以可视化 Agent 每一步的执行细节。

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "你的API Key"
os.environ["LANGSMITH_PROJECT"] = "my-agent-project"

启用后，每次 Agent 运行的完整轨迹（每轮推理、每次工具调用、每个参数和返回值）都会记录到 LangSmith 平台上，方便回溯和分析。

---

三、插上翅膀：MCP 工具接入

3.1 一个现实的痛点

上一章我们学会了用 @tool 定义本地工具。但本地工具有一个现实问题：

场景：你想让 Agent 具备"查火车票"的能力

方式一：自己写本地工具

• 需要研究 12306 的 API 文档
• 需要处理认证、签名、加密
• 需要处理各种异常和边界情况
• 需要持续维护（API 一更新就得改）

方式二：如果有人已经把"查火车票"封装成了一个标准化的服务，你只需要"接上去"就能用呢？

问题在于，不同的人封装工具服务的方式各不相同——有的用 REST API，有的用 WebSocket，有的用 gRPC……如果每接入一个外部工具都要写一套不同的适配代码，那就太痛苦了。

3.2 MCP——AI 领域的"USB-C 标准"

MCP（Model Context Protocol，模型上下文协议） 就是来解决这个问题的。它定义了一套统一的标准，让所有工具服务都以相同的方式暴露能力，AI 应用以相同的方式接入——无论底层工具是什么、在哪里运行。

一句话总结：MCP 是 AI 领域的"USB-C 标准"，它统一了 LLM 与外部工具之间的通信方式。

3.3 MCP 架构——三个角色

MCP 采用客户端-服务器架构，涉及三个角色：

┌─────────────────┐         ┌─────────────────┐         ┌─────────────────┐
│    MCP Host     │         │   MCP Client    │         │   MCP Server    │
│  (你的 Agent)    │ ───────→│  (适配器)        │ ───────→│  (工具服务)      │
│                 │         │                 │         │                 │
│ 运行 AI 应用    │         │ 与 Server 通信   │         │ 提供工具能力     │
│ LangChain 程序  │         │ LangChain 适配器 │         │ 别人封装好的工具 │
└─────────────────┘         └─────────────────┘         └─────────────────┘

3.4 MCP 工作流程

当 Agent 通过 MCP 调用一个外部工具时，完整的流程是这样的：

第①步 - 握手
  Host："你有哪些工具？"
  Server："我有 query_train（查火车票）、book_ticket（订票）……"
                    ↓
第②步 - 注入
  Host 将工具描述和用户问题一起发给 LLM
  （和本地工具完全一样——LLM 不知道也不关心工具是本地的还是远程的）
                    ↓
第③步 - 决策
  LLM："我要调用 query_train，参数是 {from: '北京', to: '上海', date: '2026-05-01'}"
                    ↓
第④步 - 路由执行
  Host → Server："执行 query_train({from: '北京', to: '上海', date: '2026-05-01'})"
  Server → Host："找到3趟列车：G1 07:00, G3 08:00, G5 09:00"
                    ↓
第⑤步 - 继续推理
  Host 将结果返回给 LLM，LLM 继续推理或输出最终回复

关键理解： 对 LLM 来说，MCP 工具和本地工具没有任何区别——它看到的都是"工具名 + 描述 + 参数"。MCP 只是改变了工具在你代码端的接入方式，对 LLM 完全透明。

3.5 两种传输协议

传输协议	原理	适用场景	示例
Stdio	通过标准输入/输出通信	本地开发、调试	Server 和你的程序在同一台机器上
Streamable HTTP	通过 HTTP 流式传输通信	生产环境、云服务	Server 部署在远程服务器上
SSE	服务器发送事件	需要服务端主动推送的场景	实时通知（较少使用）

实际开发中用得最多的是前两种：本地开发调试用 Stdio（简单、无需网络），生产部署用 Streamable HTTP（支持远程调用）。

3.6 编写 MCP Server（Stdio 方式）

# 文件名：mcp_server_stdio.py
# 安装依赖：uv add mcp
from mcp.server.fastmcp import FastMCP

# 创建 MCP Server 实例
mcp = FastMCP("MyTools")

# 用 @mcp.tool() 定义工具（写法和 LangChain 的 @tool 非常相似）
@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数的和"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """计算两个整数的乘积"""
    return a * b

# 启动 Server
if __name__ == "__main__":
    mcp.run(transport="stdio")  # 以 Stdio 方式运行

MCP Server 还支持暴露资源（Resource）和提示词模板（Prompt），不仅仅是工具：

# 资源：提供可读取的数据（类似 GET 接口）
@mcp.resource("greeting://default")
def get_greeting() -> str:
    """返回一条默认问候语"""
    return "Hello from MCP Server!"

# 提示词模板：提供预定义的提示词
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    """生成问候语的提示词"""
    styles = {
        "friendly": "写一句友善的问候",
        "formal": "写一句正式的问候",
        "casual": "写一句轻松的问候",
    }
    return f"为{name}{styles.get(style, styles['friendly'])}"

• 资源（@mcp.resource）——“静态的文件柜”：定义了一个唯一的 URI，Agent 可以去"读取"里面的背景资料，不需要"执行"什么动作。
• 提示词（@mcp.prompt）——“标准化的模板库”：存在 Server 端的一套"话术模板"。让 U 盘（Server）自带说明书，告诉 Host（大模型）：“如果你想用我，你应该这样问问题”。

为什么要把提示词放在 Server 里？ 如果这个 Server 是别人写的第三方服务（比如 GitHub 官方提供的 MCP Server），GitHub 最知道怎么引导大模型写出高质量的 PR Review。所以 GitHub 直接在 Server 里把提示词写好，你只管调用，生成出来的提示词直接喂给 LLM。

测试 Server

import asyncio
import sys
import os
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters

async def test():
    current_dir = os.path.dirname(os.path.abspath(__file__))
    server_script_path = os.path.join(current_dir, "mcp_server.py")
    
    # 配置 Server 的启动命令
    server_params = StdioServerParameters(
        command=sys.executable,      # 使用当前 Python 解释器
        args=[server_script_path],   # 使用绝对路径
    )
    
    # 连接 Server
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()  # 第①步：握手
            
            # 查看可用工具
            tools = await session.list_tools()
            print("可用工具:", tools)
            
            # 调用工具
            result = await session.call_tool("add", {"a": 10, "b": 20})
            print("调用结果:", result)  # 输出: 30

asyncio.run(test())

代码中最核心的一行是：

async with stdio_client(server_params) as (read, write):

用 USB 类比来说，这行代码就是"把准备好的 U 盘插到主机的 USB 接口上，并接通数据线"。拆成三个部分：

• stdio_client(server_params) —— 启动 Server 进程。Client 往 Server 的 stdin 写数据，从 Server 的 stdout 读结果。不走网络端口，极其轻量。
• as (read, write) —— 获取两根"数据线"。write 负责 Host → Server 发送指令，read 负责 Server → Host 返回结果。像一部对讲机：一根线负责说，一根线负责听。
• async with ... —— 自动管理生命周期。进入时建立连接，退出时自动关闭 Server 进程。不需要手写 session.close()，不会有僵尸进程残留。

3.7 编写 MCP Server（HTTP 方式）

当 Server 需要部署在远程服务器上时，使用 Streamable HTTP：

# 文件名：mcp_server_http.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MyTools")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数的和"""
    return a + b

if __name__ == "__main__":
    mcp.run(transport="streamable-http")  # 默认启动在 127.0.0.1:8000

对应的 Client 代码：

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

async def test():
    url = "http://127.0.0.1:8000/mcp"
    async with streamable_http_client(url=url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("可用工具:", tools)
            result = await session.call_tool("add", {"a": 10, "b": 20})
            print("调用结果:", result)

asyncio.run(test())

对比两种方式： Server 端的工具定义代码完全一样，只是启动时的 transport 参数不同。Client 端的连接方式不同（一个启动本地进程，一个连 HTTP 地址），但调用工具的 API 完全一致。这就是 MCP 协议标准化带来的好处。

3.8 LangChain 接入 MCP——一行代码的事

实际开发中不需要手写 Client——LangChain 提供了 langchain-mcp-adapters 包，可以直接把 MCP Server 的工具转换成 Agent 可用的工具：

# 安装依赖：uv add langchain-mcp-adapters
import os
import sys
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

# 精准定位 Server 文件路径
current_dir = os.path.dirname(os.path.abspath(__file__))
server_script_path = os.path.normpath(
    os.path.join(current_dir, "..", "stdio", "mcp_server.py")
)

# 第1步：配置 MCP Server 连接（可以同时连多个！）
client = MultiServerMCPClient({
    "my-local-tools": {
        "transport": "stdio",
        "command": sys.executable,
        "args": [server_script_path],
    },
    "12306-mcp": {
        "transport": "streamable_http",
        "url": "https://mcp.api-inference.modelscope.net/8e63d4dbeef046/mcp"
    }
})

# 第2步：获取所有 Server 的工具，创建 Agent
async def main():
    # 自动连接所有 Server，获取全部工具列表
    tools = await client.get_tools()
    print(f"成功获取到 {len(tools)} 个工具")
    
    llm = ChatOpenAI(model="gpt-4o-mini")
    agent = create_agent(llm, tools)
    
    # 第3步：像平常一样使用 Agent
    result = await agent.ainvoke({
        "messages": [("user", "信阳有多少个火车站")]
    })
    print("Agent回复:", result["messages"][-1].content)

asyncio.run(main())

核心价值： 通过 MultiServerMCPClient，你可以同时接入任意数量的 MCP Server，它们的工具会被统一转换成 LangChain 工具格式。对 Agent 来说，MCP 工具和本地 @tool 工具用起来没有任何区别。

3.9 本地工具 vs MCP 工具

对比维度	本地工具（@tool）	MCP 工具
定义位置	写在你的项目代码中	运行在独立的 MCP Server 上
适合场景	业务逻辑简单、不需要复用	通用能力、需要跨项目/跨团队复用
维护方式	和主项目一起维护	独立部署、独立维护
使用门槛	低——写个函数加个装饰器	中——需要启动 Server
生态复用	无——只有你自己能用	强——任何支持 MCP 的应用都能接入

实际建议： 项目早期或工具逻辑简单时，直接用 @tool 定义本地工具最快。当工具需要被多个项目复用、或者你想接入社区已有的工具服务时，用 MCP。两种方式可以混合使用——在同一个 Agent 中同时挂载本地工具和 MCP 工具。

---

四、记忆：让 Agent 不再"失忆"

4.1 失忆的 Agent

试想这样一个场景：

第1次调用：
  用户："我叫张三"
  Agent："你好张三！有什么可以帮你？"

第2次调用：
  用户："我叫什么名字？"
  Agent："抱歉，我不知道你叫什么名字。"  ← 失忆了！

这不是 Bug，而是 Agent 的默认行为。回忆第一章的工作循环——Agent 每次 invoke 都是一次独立的"感知→推理→行动"过程。上一次调用的对话内容，不会自动带入下一次调用中。

这就好比你每天找同一个客服咨询问题，但对方每天换一个新人，昨天说过的话今天得从头说一遍。

4.2 Checkpointer——给 Agent 装上"海马体"

LangChain 通过 checkpointer 机制实现 Agent 的记忆。工作原理非常简单：

每次调用结束后：
  Checkpointer 自动保存本次对话的所有消息
                    ↓
下次调用开始时：
  Checkpointer 自动加载之前保存的消息，拼接到新的输入前面
                    ↓
效果：
  LLM 看到的消息列表 = 历史消息 + 本次新消息
  → Agent 就"记住"了之前的对话

使用方式只需两步：

from langgraph.checkpoint.memory import InMemorySaver

# 创建 checkpointer（内存存储，程序重启后数据会丢失）
checkpointer = InMemorySaver()

# 创建 Agent 时传入 checkpointer
agent = create_agent(
    model=llm,
    tools=tools,
    checkpointer=checkpointer,  # ← 就这一行
)

4.3 Thread ID——多会话隔离

一个 Agent 通常会同时服务多个用户。不同用户的对话历史不应该互相干扰——用户 A 的聊天记录不应该出现在用户 B 的对话中。

LangChain 通过 thread_id 来隔离不同的会话。每个不同的 thread_id 维护一份独立的消息列表：

thread_id: "user_张三" → [消息1, 消息2, 消息3, ...]
thread_id: "user_李四" → [消息A, 消息B, ...]
thread_id: "user_王五" → [消息X, 消息Y, ...]

调用 Agent 时，通过 config 参数指定 thread_id：

# 张三的对话
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫张三"}]},
    config={"configurable": {"thread_id": "user_张三"}},
)

# 李四的对话（完全独立，互不干扰）
agent.invoke(
    {"messages": [{"role": "user", "content": "我叫李四"}]},
    config={"configurable": {"thread_id": "user_李四"}},
)

4.4 完整示例——记忆的前后对比

import datetime
from langchain_tavily import TavilySearch
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver

# 准备工具和模型
search = TavilySearch(max_results=5)
llm = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 创建有记忆的 Agent
checkpointer = InMemorySaver()
agent = create_agent(
    model=llm,
    tools=[search],
    checkpointer=checkpointer,
)

# ===== 第1次调用 =====
print("=== 第1次调用 ===")
for chunk in agent.stream(
    input={"messages": [
        {"role": "system", "content": f"当前时间：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"},
        {"role": "user", "content": "今天北京天气怎么样？"},
    ]},
    config={"configurable": {"thread_id": "abc123"}},
):
    print(chunk, end="\n\n")

# ===== 第2次调用（相同 thread_id → 有记忆） =====
print("=== 第2次调用 ===")
for chunk in agent.stream(
    input={"messages": [{"role": "user", "content": "我刚才问你什么了？"}]},
    config={"configurable": {"thread_id": "abc123"}},  # ← 相同 thread_id
):
    print(chunk, end="\n\n")
# Agent 会回答："你刚才问了北京今天的天气"

# ===== 第3次调用（不同 thread_id → 无记忆） =====
print("=== 第3次调用（新会话） ===")
for chunk in agent.stream(
    input={"messages": [{"role": "user", "content": "我刚才问你什么了？"}]},
    config={"configurable": {"thread_id": "xyz789"}},  # ← 不同 thread_id
):
    print(chunk, end="\n\n")
# Agent 会回答："这是我们第一次对话，你还没有问过任何问题"

4.5 记忆的代价

Checkpointer 解决了"失忆"问题，但引入了一个新问题：随着对话轮次增加，保存的消息列表会越来越长。

第1轮：  [用户消息1, AI回复1]                          → 2条消息
第10轮： [用户消息1, AI回复1, ..., 用户消息10, AI回复10] → 20条消息
第100轮：[用户消息1, AI回复1, ..., 用户消息100, AI回复100] → 200+条消息

每次调用时，所有历史消息都会发送给 LLM。消息太多会导致两个问题：Token 消耗剧增（费钱），甚至超出模型的上下文窗口限制（报错）。

这个问题的解决方案就是下一章要讲的中间件——可以在消息发给 LLM 之前自动进行压缩和总结。

---

五、守卫者：Agent 中间件

5.1 什么是中间件

中间件是一种插入 Agent 执行流程中的"拦截器"。它可以在 Agent 执行的各个关键节点上介入，对数据进行加工处理。

用户消息
    ↓
┌───────────────────────────────────┐
│  before_model 中间件               │ ← 消息发给 LLM 之前
│  （压缩历史消息、注入额外上下文）      │
├───────────────────────────────────┤
│           LLM 推理                 │
├───────────────────────────────────┤
│  after_model 中间件                │ ← LLM 返回结果之后
│  （记录日志、过滤敏感内容）           │
├───────────────────────────────────┤
│  wrap_tool 中间件                  │ ← 工具执行前后
│  （人工审核、权限控制）              │
└───────────────────────────────────┘
    ↓
最终回复

使用方式很简单——创建中间件实例，通过 middleware 参数传入 create_agent：

agent = create_agent(
    model=llm,
    tools=tools,
    checkpointer=checkpointer,
    middleware=[middleware_a, middleware_b],  # ← 中间件列表
)

5.2 消息压缩中间件——解决记忆膨胀

SummarizationMiddleware 会在消息发给 LLM 之前，自动检查消息列表的长度。当消息量超过设定的阈值时，它会用一个独立的 LLM 调用，把旧消息压缩成一段摘要，从而大幅减少 Token 消耗。

from langchain.agents.middleware import SummarizationMiddleware
from langchain_openai import ChatOpenAI

summary_middleware = SummarizationMiddleware(
    model=ChatOpenAI(model="gpt-4o-mini"),  # 用于生成摘要的 LLM
    trigger=("messages", 100),  # 触发条件：消息数量达到100条时压缩
)

压缩的效果：

压缩前（100条消息）：
[用户:你好, AI:你好, 用户:天气?, AI:晴天, ..., 用户:最新问题]
                    ↓ SummarizationMiddleware 介入
压缩后（2条消息）：
[系统:以下是之前对话的摘要：用户询问了天气、订单状态等问题..., 用户:最新问题]

trigger 参数支持三种触发策略：

触发策略	写法	含义
按消息数量	("messages", 100)	消息数量达到 100 条时触发压缩
按 Token 比例	("fraction", 0.5)	Token 数达到模型上下文窗口的 50% 时触发
按 Token 绝对值	("tokens", 3000)	Token 数达到 3000 时触发

5.3 人工审核中间件——给危险操作加把锁

有些操作是高风险的——比如转账、删除数据、发送邮件。即使 LLM 决定要执行这些操作，我们也希望先让人类确认一下再真正执行。

HumanInTheLoopMiddleware 就是做这件事的。它会在指定的工具执行前暂停 Agent，等待人类审核：

from langchain.agents.middleware import HumanInTheLoopMiddleware

hitl_middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "transfer_money": True,   # 转账 → 需要审核
        "delete_record": True,    # 删除 → 需要审核
        "get_weather": False,     # 查天气 → 不需要审核
    }
)

完整示例

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

# ===== 1. 定义工具 =====
@tool
def get_weather(city: str) -> str:
    """查询天气"""
    return f"{city}的天气晴朗，气温25度。"

@tool
def transfer_money(amount: int, to_account: str) -> str:
    """转账操作（敏感操作，需要审核）
    
    Args:
        amount: 转账金额（元）
        to_account: 收款账户名称
    """
    print(f">>> 正在执行转账: {amount}元 → {to_account}")
    return f"成功转账 {amount} 元给 {to_account}。"

# ===== 2. 配置中间件 =====
hitl_middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "transfer_money": True,   # 转账需要审核
        "get_weather": False,     # 查天气不需要
    }
)

# ===== 3. 创建 Agent =====
llm = init_chat_model("gpt-4o-mini", model_provider="openai")
checkpointer = InMemorySaver()
agent = create_agent(
    model=llm,
    tools=[get_weather, transfer_money],
    middleware=[hitl_middleware],
    checkpointer=checkpointer,  # 人工审核必须配合 checkpointer 使用
)

def run_demo():
    config = {"configurable": {"thread_id": "thread-1"}}
    
    # 用户请求转账
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "请帮我转账 100 元给 Alice"}]},
        config=config,
    )
    
    # 检查是否被拦截
    if "__interrupt__" in result:
        interrupt_value = result["__interrupt__"][0].value
        print("⚠ 操作被拦截，等待审核")
        
        # 人工审核 → 批准
        decision = "approve"
        action_requests = interrupt_value.get("action_requests", [])
        decisions = [{"type": decision} for _ in action_requests]
        
        result = agent.invoke(
            Command(resume={"decisions": decisions}),
            config=config,
        )
        
        for msg in result.get("messages", []):
            if hasattr(msg, "type") and msg.type == "ai" and not getattr(msg, "tool_calls", None):
                print(f"[Agent]: {msg.content}")

run_demo()

执行流程：

用户："请帮我转账 100 元给 Alice"
            ↓
LLM 推理：需要调用 transfer_money(100, "Alice")
            ↓
HumanInTheLoopMiddleware 拦截！⏸
            ↓
人工审核：approve ✅ / reject ❌
            ↓
批准 → 继续执行 transfer_money → 返回结果
拒绝 → Agent 收到拒绝信息，重新组织回复

---

六、匠心：Agent 最佳实践

前面五章我们学会了 Agent 的概念和各项能力。但在实际项目中，光会用 API 还不够——工具怎么设计、提示词怎么写、出了问题怎么排查，这些"怎么用好"的经验往往决定了 Agent 的实际表现。

6.1 工具设计五原则

工具是 Agent 与外部世界的接口。工具设计得好，LLM 就能准确调用；设计得不好，LLM 就会选错工具、传错参数，甚至根本不知道什么时候该用它。

原则	为什么重要	正面示例	反面示例
单一职责	LLM 更容易理解功能明确的工具	get_weather、get_forecast 各做一件事	handle_weather_and_forecast_and_alert 一个工具做太多事
描述清晰	LLM 完全依赖描述来决定何时用、怎么用	“获取指定城市今天的实时天气，返回温度和天气状况”	“查天气”
参数具体	减少 LLM 猜测参数格式的可能	date: str = Field(description="日期，格式YYYY-MM-DD")	date: str（没有格式说明）
错误友好	LLM 可以根据错误信息调整策略	返回 “城市名’北精’无法识别，你是否指’北京’？”	抛出 KeyError: '北精'
幂等安全	避免 Agent 重试时产生副作用	get_user(id=123) 调用多次结果相同	create_order() 调用多次会创建多个订单

经验法则： 如果一个工具的 docstring 超过 3 句话才能描述清楚它做什么，说明这个工具承担了太多职责，应该拆分。

6.2 提示词优化——给 Agent 一本"工作手册"

系统提示词是你给 Agent 的"工作手册"。一个好的系统提示词应该告诉 Agent 它的角色、可用的工具、工作流程和注意事项。

# ✅ 好的系统提示词：结构清晰，指导明确
GOOD_PROMPT = """你是一个专业的数据分析助手。

你的工作流程：
1. 理解用户的分析需求
2. 使用 search 工具获取相关数据
3. 使用 calculate 工具进行计算
4. 用简洁的语言呈现分析结果

注意事项：
- 计算结果保留2位小数
- 如果数据不足以得出结论，主动告知用户
- 不要编造数据，所有数据必须来自工具查询
"""

# ❌ 差的系统提示词：太模糊，没有指导价值
BAD_PROMPT = """你是一个AI助手，帮助用户解决问题。"""

提示词优化的关键点：

要素	作用	示例
角色定位	约束 Agent 的回答风格和专业度	“你是一个专业的数据分析助手”
工作流程	引导 Agent 按合理顺序操作	“先搜索数据，再计算，最后呈现”
约束条件	避免 Agent 犯常见错误	“不要编造数据”
输出格式	统一回复的格式和质量	“计算结果保留2位小数”

6.3 三层调试法

Agent 的执行过程是动态的、不确定的，出了问题比固定流程的 Chain 更难排查。以下是三个层次的调试手段：

第一层：开启日志（快速定位）

import logging
logging.basicConfig(level=logging.DEBUG)

开启后可以在控制台看到每次 LLM 调用的输入输出、每次工具调用的参数和结果。

第二层：使用 LangSmith（可视化追踪）

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-key"
os.environ["LANGSMITH_PROJECT"] = "my-agent-debug"

LangSmith 会把 Agent 的每一步执行记录下来，在 Web 界面上以时间线的形式展示，可以清晰地看到：LLM 每次推理的输入输出、选择了哪个工具、传了什么参数、工具返回了什么、共执行了多少轮循环。

第三层：流式输出排查（观察中间过程）

# 用 stream 替代 invoke，实时观察每一步
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "你的测试问题"}]}
):
    print(chunk, end="\n\n")

调试经验： 大多数 Agent 问题的根源可以归为三类——工具描述不够清晰（LLM 选错工具）、系统提示词引导不足（LLM 执行顺序混乱）、工具返回值格式不规范（LLM 无法解读结果）。遇到问题时优先从这三个方向排查。

6.4 性能优化

问题	优化方法	效果
响应慢	只加载当前任务必需的工具，避免给 LLM 太多选择	减少 LLM 决策时间
Token 消耗高	使用 SummarizationMiddleware 压缩历史消息	降低每次调用的 Token 用量
工具调用次数多	优化工具设计，一次返回尽量完整的信息	减少循环轮次
吞吐量不足	使用 ainvoke 异步调用，支持并发处理多个请求	提高吞吐量

---

七、回望与远眺：课程总结

7.1 知识全景图

回顾整个学习旅程，我们按照"是什么 → 怎么造 → 怎么用好"的逻辑，逐步构建了一个具备完整能力的 Agent：

Model I/O → Chain → RAG → Agent
                              │
                   ┌──────────┼──────────┐
                   │          │          │
                工具调用     记忆       中间件
              (@tool/MCP)  (checkpointer) (middleware)
                   │          │          │
                   └──────────┼──────────┘
                              │
                          最佳实践
                       (工具设计/提示词/调试)

章节	核心问题	学到了什么
一、Agent 介绍	Agent 是什么？和 Chain 有什么区别？	Agent = LLM + 自主决策 + 工具调用，通过循环完成动态任务
二、工具定义与使用	怎么让 Agent 能"做事"？	@tool 定义工具，create_agent 组装 Agent
三、MCP 工具接入	怎么接入别人已有的工具？	MCP 统一工具接入标准，MultiServerMCPClient 接入远程工具
四、记忆管理	怎么让 Agent"记住"对话？	checkpointer + thread_id 实现多会话记忆
五、中间件	怎么控制 Agent 的执行过程？	消息压缩（SummarizationMiddleware）、人工审核（HumanInTheLoopMiddleware）
六、最佳实践	怎么把 Agent 做好？	工具设计原则、提示词优化、调试技巧

7.2 技术栈定位

每一层都是在上一层的基础上增加新的能力：

• Model I/O 让你能调用模型
• Chain 让你能编排流程
• RAG 让模型拥有知识
• Agent 让模型拥有"判断力"和"行动力"

Agent 处于技术栈的最上层，是 AI 应用能力的集大成者。

7.3 常见问题

Q: Chain 和 Agent 如何选择？

任务流程确定时用 Chain（简单可靠），任务需要动态判断时用 Agent（灵活但复杂度更高）。两者可以结合——Agent 内部的某些子任务可以用 Chain 实现。

Q: RAG 和 Agent 可以结合吗？

可以，而且非常常见。典型做法是把 RAG 检索封装成一个工具，Agent 在需要时自主调用它。

Q: 如何优化 Agent 的响应速度？

四个方向：只加载必需的工具（减少 LLM 决策开销）、用 SummarizationMiddleware 压缩长对话（减少 Token）、使用异步调用 ainvoke（支持并发）、优化工具的返回值格式（减少循环轮次）。

Q: 什么时候需要 LangGraph？

当你需要多个 Agent 协作、需要复杂的条件分支和循环逻辑、或需要比 create_agent 更精细的流程控制时，就需要 LangGraph。它是 Agent 的进阶编排工具。

7.4 下一步

• 动手实践： 选择一个实际场景（如个人知识库问答、自动化数据分析），用学到的知识构建一个 Agent
• 深入 LangGraph： 学习状态图、条件边、多 Agent 协作，构建更复杂的工作流
• 关注 AI 生态： 关注社区动态可以快速扩展 Agent 的能力

推荐资源：

• LangChain 官方文档：https://docs.langchain.com/
• LangSmith 调试追踪：https://www.langchain.com/langsmith
• LangGraph 文档：https://docs.langchain.com/oss/python/langgraph
• GitHub 示例库：https://github.com/langchain-ai/langgraph/tree/main/examples

---

写在最后

学完这整个旅程，我最深的感受是：Agent 的本质不是技术的堆砌，而是思维的转变。

过去我们写代码，是在"告诉机器怎么做"——每一步流程、每一个判断、每一条路径，都是我们预先设计好的。而 Agent 让我们开始"告诉机器要做什么"，然后信任它自己去想怎么做。

这种转变令人兴奋，也令人敬畏。

当 LLM 从一个"只会回答问题的文本生成器"，变成了一个"能感知、能推理、能行动的智能体"，我们面对的就不再是一个工具，而是一个协作者。

而我们要做的，是为这个协作者设计好工具（让它有能力）、设计好记忆（让它有 continuity）、设计好中间件（让它有底线），然后用清晰的提示词告诉它：你是谁，你要做什么，你的边界在哪里。

这，就是 Agent 的全部哲学。

“The best way to predict the future is to invent it.” —— Alan Kay

最好的预测未来的方式，就是亲手创造它。