当 AI 学会自己思考：我的 Agent 智能体学习笔记
写在前面
你有没有过这样的体验：让 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 会自主完成以下步骤（无需我们编码控制流程）：
- 调用
get_current_date获取今天日期 - 自己计算天数差，或调用
calculate来辅助计算 - 组织语言回复用户
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
最好的预测未来的方式，就是亲手创造它。
更多推荐
18
0- 0
已为社区贡献7条内容
所有评论(0)