上一篇用 create_agent 搭出了第一个智能体，结尾留了三个现实问题：外部工具接入成本高、Agent 天生失忆、敏感操作无人把关。这一篇分别用 MCP 协议、Checkpointer 记忆和中间件解决它们。这也是本系列的最后一篇，文末对四大模块做一个总收束。

一、MCP：给工具接入定一个标准

1.1 为什么需要 MCP

假设你想让 Agent 具备"查火车票"的能力。自己写本地工具意味着：研究 12306 的接口文档、处理认证签名、处理各种异常边界、接口一变还得跟着维护。而如果有人已经把这个能力封装成了服务，你只要"接上去"就好。

但现实的麻烦在于，每个人封装服务的方式都不一样——REST、WebSocket、gRPC 各行其是，每接一个外部工具就要写一套适配代码。**MCP（Model Context Protocol，模型上下文协议）**就是为此而生的统一标准：所有工具服务以相同的方式暴露能力，AI 应用以相同的方式接入，无论底层工具是什么、跑在哪里。最贴切的类比是 USB-C——MCP 是 AI 领域的"USB-C 标准"，统一了大模型与外部工具之间的接口。

1.2 架构与工作流程

MCP 是客户端-服务器架构，三个角色：

角色职责在我们的场景中对应
MCP Host运行 AI 应用的宿主程序你的 LangChain Agent 程序
MCP Client与 Server 通信的客户端LangChain 的 MCP 适配器
MCP Server提供工具能力的服务端别人封装好的工具服务

工作流程五步：Agent 启动时与 Server 握手，拿到工具列表和描述；Host 把工具描述随用户问题注入给模型；模型决策调用哪个工具、传什么参数；Host 通过 MCP 协议把调用请求路由给 Server 执行并取回结果；结果交还模型继续推理。

发现关键了吗——这和上一篇讲的 Function Calling 流程一模一样。对模型来说，MCP 工具和本地工具没有任何区别，它看到的都是"工具名 + 描述 + 参数"。MCP 改变的只是工具在你代码端的接入方式，对模型完全透明。

Client 和 Server 之间的通信支持多种传输方式：Stdio 通过标准输入/输出通信，适合 Server 和你的程序在同一台机器上（本地开发调试）；Streamable HTTP 走 HTTP 流式传输，适合 Server 部署在远程（生产环境）。实际开发中用得最多的就是这两种。

1.3 写一个 MCP Server

接别人的 Server 之前，先自己写一个，理解服务端是怎么工作的。官方 SDK 的 FastMCP 让这件事和写 @tool 一样简单：

# mcp_server_stdio.py    （pip install mcp）
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MyTools")

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

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

if __name__ == "__main__":
    mcp.run(transport="stdio")   # 换成 "streamable-http" 即变为 HTTP 服务

Stdio 模式下 Client 会在后台把这个脚本作为子进程拉起，双方直接通过进程的标准输入/输出通信，不占网络端口，极其轻量。切到 HTTP 模式时，工具定义的代码一行不用改，只换 transport 参数——这正是协议标准化的好处。

除了工具，MCP Server 还能暴露两类东西：资源（@mcp.resource），类似只读的文件柜，给模型提供可读取的背景资料（系统日志、员工手册、配置参数）；提示词模板（@mcp.prompt），Server 自带的"话术模板"。后者的价值在于：如果 Server 是第三方提供的（比如 GitHub 官方的 MCP Server），服务提供方最清楚怎么引导模型用好自己的工具，直接把提示词内置在 Server 里，你只管取用。

1.4 LangChain 接入：MultiServerMCPClient

实际开发中不需要手写 MCP Client，langchain-mcp-adapters 包直接把 MCP Server 的工具转成 LangChain 工具：

# pip install langchain-mcp-adapters
import sys, asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

client = MultiServerMCPClient({
    # 本地 Stdio Server：作为子进程拉起
    "my-local-tools": {
        "transport": "stdio",
        "command": sys.executable,          # 用当前虚拟环境的 Python
        "args": ["./mcp_server_stdio.py"],  # 建议用绝对路径
    },
    # 远程 HTTP Server：比如社区提供的 12306 查票服务
    "12306-mcp": {
        "transport": "streamable_http",
        "url": "https://example.com/mcp",
    },
})

async def main():
    tools = await client.get_tools()   # 连接所有 Server，汇总全部工具
    agent = create_agent(model="openai:gpt-4o-mini", tools=tools)

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "武汉有多少个火车站？"}]}
    )
    print(result["messages"][-1].content)

asyncio.run(main())

get_tools() 返回的工具和本地 @tool 定义的格式完全相同，可以和本地工具混在同一个列表里传给 Agent。选型上的建议：项目早期或逻辑简单，本地 @tool 最快；工具需要跨项目复用、或想直接接入社区现成的服务时，用 MCP。两者随时混用。

二、记忆：让 Agent 记住上一轮

2.1 失忆是默认行为

先看一个新手必踩的坑：

第1次调用：  用户："我叫张三"     Agent："你好张三！"
第2次调用：  用户："我叫什么？"   Agent："抱歉，我不知道你的名字。"

这不是 Bug。每次 invoke 都是一次独立的"感知→推理→行动"，上一次的对话内容不会自动带入下一次。就像你每天找同一家客服，对方每天换个新人，昨天说的话今天得重讲一遍。

2.2 Checkpointer：两行代码接上记忆

LangChain 通过 checkpointer 机制实现记忆，原理很直白：每次调用结束自动保存本轮全部消息，下次调用开始自动加载历史消息拼在新输入前面——模型看到的是"历史消息 + 本次新消息"，自然就"记住"了：

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()   # 内存存储，程序重启即丢；生产可换持久化实现

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    checkpointer=checkpointer,   # ← 就这一行
)

2.3 thread_id：会话隔离

一个 Agent 通常同时服务多个用户，张三的聊天记录绝不能出现在李四的对话里。LangChain 用 thread_id 隔离会话——每个 thread_id 维护一份独立的消息历史，调用时通过 config 指定：

config_a = {"configurable": {"thread_id": "user_张三"}}
config_b = {"configurable": {"thread_id": "user_李四"}}

# 第1次调用：张三报上名字
agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config=config_a)

# 第2次调用：同一个 thread_id → 记得
r = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config=config_a)
print(r["messages"][-1].content)   # "你叫张三"

# 第3次调用：不同 thread_id → 全新会话，互不干扰
r = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config=config_b)
print(r["messages"][-1].content)   # "这是我们第一次对话……"

2.4 记忆的代价

Checkpointer 解决了失忆，也带来一个新问题：对话轮次越多，消息列表越长——第 100 轮时已经积累两百多条消息，而每次调用都要把全部历史发给模型。后果有两个：Token 消耗剧增（费钱），甚至撑爆模型的上下文窗口（报错）。这个问题的答案在下一节。

三、中间件：在 Agent 的关键节点上做手脚

3.1 中间件是什么

中间件（Middleware）是插入 Agent 执行流程中的"拦截器"，可以在几个关键节点介入，对数据进行加工：

介入位置时机典型用途
before_model消息发给模型之前压缩历史消息、注入额外上下文
after_model模型返回结果之后记录日志、过滤敏感内容
wrap_tool_call工具执行前后人工审核、权限控制

使用方式和 checkpointer 一样简单——通过 middleware 参数传给 create_agent，可以传多个。下面看两个最实用的内置中间件。

3.2 消息压缩：SummarizationMiddleware

这就是上一节遗留问题的解法。它在消息发给模型之前检查消息列表的规模，超过阈值就用一次独立的模型调用把旧消息压缩成一段摘要：

from langchain.agents.middleware import SummarizationMiddleware
from langchain.agents import create_agent

summary_middleware = SummarizationMiddleware(
    model="openai:gpt-4o-mini",     # 用于生成摘要的模型
    trigger=("messages", 100),      # 消息数达到100条时触发压缩
)

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[...],
    checkpointer=checkpointer,
    middleware=[summary_middleware],
)

压缩前后的对比：一百条消息进去，出来的是"一条摘要 + 最近的消息"，Token 占用断崖式下降，而关键信息以摘要形式保留。trigger 支持三种触发策略：

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

3.3 人工审核：HumanInTheLoopMiddleware

有些操作是高风险的——转账、删数据、发邮件。即使模型决定执行，也应该先经人类确认。HumanInTheLoopMiddleware 会在指定工具执行前暂停 Agent，等待审核：

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

@tool
def transfer_money(amount: int, to_account: str) -> str:
    """转账操作。
    :param amount: 转账金额（元）
    :param to_account: 收款账户名称
    """
    return f"成功转账 {amount} 元给 {to_account}。"

hitl_middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "transfer_money": True,   # 转账 → 需要审核
        "get_weather": False,     # 查天气 → 直接放行
    }
)

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_weather, transfer_money],
    middleware=[hitl_middleware],
    checkpointer=InMemorySaver(),   # 人工审核必须配合 checkpointer 使用
)

config = {"configurable": {"thread_id": "thread-1"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "请帮我转账100元给Alice"}]},
    config=config,
)

# Agent 在执行 transfer_money 前被暂停，返回结果里带 __interrupt__
if "__interrupt__" in result:
    print("操作被拦截，等待人工审核……")
    interrupt_value = result[][].value
    
    decisions = [{: }  _  interrupt_value[]]

    
    result = agent.invoke(Command(resume={: decisions}), config=config)

(result[][-].content)

注意两点。其一，必须配合 checkpointer——Agent 被打断后，"执行到哪了"这个状态要靠 checkpointer 保存，审核通过后才能原地恢复。其二，恢复时的决定不止 approve 一种，也可以拒绝执行，Agent 会把"该操作被拒绝"作为观察结果继续推理。

到这里，上一篇结尾的三个问题全部有了答案：接工具难——MCP；失忆——Checkpointer + thread_id；敏感操作裸奔——HumanInTheLoopMiddleware。

四、系列总结

六篇文章，正好走完 LangChain 的四大模块，把整条主线倒着串一遍：

  • Model I/O（第一、二篇）：init_chat_model 统一接入任意厂商模型，四种消息写法，同步/异步/流式/批量四种调用形态，提示词模板，输出解析（JsonOutputParser 与 with_structured_output）；
  • Chains（第二篇）：一切皆 Runnable，prompt | model | parser 的串行链，RunnableParallel 并行链，串并组合的混合链；
  • RAG（第三、四篇）：索引侧的加载/切分/双向量编码/入库，检索侧的稠密+稀疏混合检索、RRF 融合、标量过滤，最后把命中内容拼进提示词生成答案；
  • Agents（第五、六篇）：Function Calling 机制，@tool 定义工具，create_agent 构建智能体，MCP 标准化接入外部工具，Checkpointer 记忆与会话隔离，中间件做消息压缩与人工审核。

这四个模块之间不是并列关系，而是能力的层层叠加：会调模型，才能把调用编排成链；会编排，才能搭出"检索 + 生成"的 RAG 流水线；而 Agent 把前面所有能力全部收编——它的大脑是 Model I/O，它的执行循环是 LangGraph 意义上的链，RAG 检索完全可以包成一个工具挂给它调用。学到这里，你已经具备了搭建绝大多数大模型应用的组件库。

往下走有两个自然的延伸方向：一是 LangGraph——本系列的 Agent 底层就是它，当你需要更精细地控制执行图（多 Agent 协作、复杂分支、循环上限）时，值得专门去学；二是工程化——把 LangSmith 的追踪评估用起来，把 InMemorySaver 换成持久化存储，把示例里的模拟工具换成真实业务接口。框架给的是积木，应用的成色取决于你对场景的理解。

老规矩，以官方文档为准，把每段代码亲手跑一遍。这个系列到此完结，希望它陪你走完了从"会调 API"到"能搭智能体"的这段路。