LangChain 1.0 中间件通关指南：从异步基石到生产级三大件(十一)

🚀 开篇：别让你的Agent"排队办事"

很多朋友写智能体，往往卡在两步：

异步解决第一个问题，中间件解决第二个问题。

今天咱们就把同步/异步、MCP、中间件这三个硬骨头，一次啃透。

📦 第一部分：地基——Python异步，不是"多线程"的平替

⚠️ 核心认知： 异步是单线程任务调度，不是并行计算。

🔸 同步 = 排队办事

import time
time.sleep(2) # CPU在这干瞪眼，线程被阻塞

任务按顺序执行，前一个I/O没完，后面的只能等着。

🔸 异步 = 叫号系统

import asyncio
await asyncio.sleep(2) # 释放线程，CPU去处理其他任务

提交任务后不用傻等，I/O完成了会自动通知你。

📊 一图看懂区别

🎯 记住一句话： 异步不是让程序"跑得更快"，而是让程序"忙起来不闲着"。

🔌 第二部分：MCP协议——Agent世界的"USB接口"

有了异步调度，接下来解决工具怎么接的问题。

以前怎么接工具？

# 每个工具都要写定制代码
def call_weather_api(city):
# 定制HTTP请求…

def call_database_query(sql):
# 定制数据库连接…

换一个Agent框架，全部重写。

现在用MCP

MCP = 模型上下文协议，标准化工具的定义和调用。

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
"weather": {
"transport": "streamable_http",
"url": "http://localhost:8000/mcp"
},
"math": {
"transport": "stdio",
"command": "python",
"args": ["math_server.py"]
}
})
tools = await client.get_tools() # 一把梭，全量拉取

MCP两种传输方式

🎯 MCP最大价值： 工具独立部署、独立升级，Agent只管调用，解耦！

🛡️ 第三部分：中间件——Agent的"生产级护城河"（重点）

如果说异步是发动机，MCP是接口，那中间件就是保障系统稳定、合规、可靠的"护城河"。

LangChain 1.0的中间件分三大类：

┌─────────────────────────────────────────────┐
│ 中间件三大件 │
├─────────────────────────────────────────────┤
│ 1. 预置中间件 │ 开箱即用，配置就行 │
│ 2. 装饰器中间件 │ 轻量级切面，一个注解搞定 │
│ 3. 类中间件 │ 复杂逻辑，组合钩子 │
└─────────────────────────────────────────────┘

🔸 1. 预置中间件：开箱即用的"安全阀"

① SummarizationMiddleware（自动摘要）

对话一长就爆上下文窗口？这个中间件自动压缩早期历史。

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
model="gpt-4",
tools=[…],
middleware=[SummarizationMiddleware(
trigger=("tokens", 1000), # 超过1000 token触发
keep=("messages", 5) # 保留最近5条
)]
)

原理： 前15轮对话 → 压缩成"用户询问了退货政策" → 省Token、省成本。

② Human-in-the-loop（人工审核）

delete_file、send_email这类操作，必须人点头才能执行。

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
model="gpt-4",
tools=[dangerous_tool],
middleware=[HumanInTheLoopMiddleware(
tools=["dangerous_write"],
interrupt=True
)],
checkpointer=InMemorySaver() # ⚠️ 必须有，否则状态丢失
)

三种决策：

⚠️ 踩坑提醒： HITL必须配合checkpointer，否则中断后无法恢复。

🔸 2. 装饰器中间件：轻量级"切面"

只想在某个节点插一脚（记日志、校验输入）？一个注解搞定。

钩子分类

实战：拦截敏感词

from langchain.agents.middleware import after_model

@after_model
def block_sensitive_words(state, response):
if "BLOCKED_CN" in response["output"]:
return {"output": "⚠️ 检测到违规内容，已终止输出。"}
return response

🔸 3. 类中间件：复杂逻辑的"核武器"

需要组合多个钩子、维护状态？继承AgentMiddleware。

class LoggingMiddleware(AgentMiddleware):
def before_agent(self, state, runtime):
print(f"[{time.time()}] Agent开始执行")

def after_agent(self, state, runtime):
print(f"[{time.time()}] Agent执行完毕")

关键规则：

• ✅ 必须继承AgentMiddleware

• ✅ 方法名固定：before_agent、after_model等

• ✅ 每次调用要new对象，不要传类

🔄 中间件执行顺序："洋葱模型"

多个中间件共存时，执行顺序是先入后出：

注册：[A, B, C]

执行流程：
before_A → before_B → before_C → 【核心调用】 → after_C → after_B → after_A

实战建议：

⚠️ 第四部分：避坑指南——三个致命误区

🎯 记住： 异步解决I/O阻塞，中间件解决生产治理，各司其职，别混为一谈。

📝 结语

这篇文章是LangChain系列的第一篇，我们厘清了：

• ✅ 异步 → 让Agent不闲着（I/O密集场景）

• ✅ MCP → 让工具标准化接入（解耦部署）

• ✅ 中间件 → 让Agent可治理（安全、成本、可观测）

下次预告： 我们聊LangSmith——Agent的"监控仪表盘"。怎么追踪每一次模型调用、怎么分析Token消耗、怎么定位Agent"抽风"的根本原因。

💬 互动话题： 你的Agent上生产了吗？遇到过哪些"坑"？HITL对接前端页面怎么搞？欢迎评论区交流！

如果觉得有用，点个 👍 和 ⭐ ，催更下一期！