LangChain 1.0 入门实战 · Part 6：LangChain Agent 中间件（Middleware）入门介绍

中间件（Middleware）是 LangChain 1.0 的重大更新点之一：允许开发者在 Agent 执行过程中，通过“钩子”介入并改写行为，从而更细粒度控制 Agent 的每个环节。传统 React Agent 核心循环图加入中间件后的流程图直觉理解：原本 Agent 是“固定流程”；加入中间件后，你可以在关键阶段插入逻辑（监控、改消息、换模型、拦截工具、重试/降级等）。Middleware

拉普拉斯之妖44

576人浏览 · 2026-01-02 22:06:25

拉普拉斯之妖44 · 2026-01-02 22:06:25 发布

LangChain 1.0 入门实战 · Part 6：LangChain Agent 中间件（Middleware）入门介绍

0. 文件概览

Notebook 前半部分是课程/宣传说明（包含多张外链图片）。
技术主体从标题 “Part 6.LangChain Agent中间件入门介绍” 开始，核心包括：
1. LangChain 1.0 中间件核心能力与钩子（hooks）
2. 用中间件实现动态模型选择（routing）
3. 消息压缩：Trimming / Deleting / Summarization
4. 用中间件实现 Human-in-the-loop（人在闭环）

1. 环境准备

1.1 加载环境变量

import os
from dotenv import load_dotenv 
load_dotenv(override=True)

用途：从 .env 读取 API Key 等配置（例如 DeepSeek、Tavily 等），override=True 表示覆盖已有同名环境变量。

2. LangChain 1.0 中间件核心功能介绍

2.1 中间件是什么？

中间件（Middleware）是 LangChain 1.0 的重大更新点之一：允许开发者在 Agent 执行过程中，通过“钩子”介入并改写行为，从而更细粒度控制 Agent 的每个环节。

Notebook 里用两张流程图对比：

传统 React Agent 核心循环图（外链图片：core_agent_loop.png）
加入中间件后的流程图（外链图片：middleware_final.png）

直觉理解：

原本 Agent 是“固定流程”；
加入中间件后，你可以在关键阶段插入逻辑（监控、改消息、换模型、拦截工具、重试/降级等）。

2.2 中间件能做什么？（功能类别）

Notebook 用表格列出典型能力（表格里部分描述用 ... 省略，但含义明确）：

功能类别	功能说明	常用钩子
监控 / 日志 / 指标	跟踪行为、模型调用次数、token 用量等	`before_model`、`after_model`、`wrap_model_call`
修改输入 / 上下文工程	调用前改消息历史、加 system prompt、裁剪上下文等	`before_model`、`modify_model_request`
动态模型 / 工具路由	根据状态决定用哪个模型/启用哪些工具	`modify_model_request`、`wrap_model_call`
控制流程 / 限流 / 重试 / 降级	超时、重试、降级、跳转结束等	`wrap_model_call`、`before_model`（跳转）
合规 / 审核（Guardrails / HITL）	人审、敏感内容/PII 检测、安全策略等	`after_model`（或 `wrap_tool_call`）
摘要 / 上下文裁剪	对话长时自动摘要/压缩	`before_model`
工具选择与调用管理	决定工具可用性、限制调用频率等	`modify_model_request`、`wrap_tool_call`
结构化输出控制	输出格式规范、结构校验/重写	`modify_model_request`、`after_model`
状态扩展	自定义 state schema（次数、用户信息等），读写状态	`before_model`、`after_model`

2.3 “模型相关”三类关键钩子（重点）

Notebook 总结：模型相关钩子主要三种（可单用/组合）：

before_model：模型调用前
- 常见用途：总结/裁剪历史、注入系统指令、敏感信息脱敏、状态校验、条件分支跳转
modify_model_request：发送请求前的精准改写
- 常见用途：改模型名、参数、tools 列表、messages 等
after_model：模型返回后
- 常见用途：人审（HITL）、输出校验/重写、打安全标签、可观测数据收集

此外还有两种“包装器”式钩子（把一次调用整体包起来）：

wrap_model_call：包裹一次模型调用
- 适合：动态换模型、改温度、A/B、回退策略、重试/熔断/缓存/降级
- Notebook 特别强调：v1 把“动态模型选择”正式迁入这里
wrap_tool_call：包裹一次工具调用
- 适合：工具超时/重试、白黑名单、错误上报、在人审批准前阻断高风险工具（写文件、SQL、HTTP 等）

3. 模型动态选择（Dynamic Routing）

3.1 场景目标

根据问题复杂度自动选择：

复杂问题 → deepseek-reasoner（推理更强，适合多步骤/长上下文/证明与规划）
简单问题 → deepseek-chat（更快/更省）

并结合工具（例：Tavily 搜索）：

from langchain_community.tools.tavily_search import TavilySearchResults
web_search = TavilySearchResults(max_results=2)
tools = [web_search]

3.2 核心做法：用 `@wrap_model_call` 动态替换 `request.model`

Notebook 的关键代码骨架如下（原文中间出现了 ...，说明 Notebook 省略了部分实现细节；但核心模式很清晰）：

from langchain_deepseek import ChatDeepSeek
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain_core.messages import HumanMessage

basic_model = ChatDeepSeek(model="deepseek-chat")
reasoner_model = ChatDeepSeek(model="deepseek-reasoner")

def _get_last_user_text(messages) -> str:
    for m in reversed(messages):
        if isinstance(m, HumanMessage):
            return m.content if isinstance(m.content, str) else ""
    return ""

...  # 这里 Notebook 省略了 dynamic_deepseek_routing 的完整函数体（见下方解释）

agent = create_agent(
    model=basic_model,
    tools=tools,
    middleware=[dynamic_deepseek_routing]
)

Notebook 对代码的解释（逐条）

两种模型并存

chat：日常问答快、省
reasoner：复杂推理强

wrap_model_call 的意义

能在一次模型调用“外层加壳”
在调用 handler(request) 前，直接改 request.model → 实现运行时路由

复杂度启发式（heuristics）
Notebook 提供了 3 条判断复杂的启发式规则（很实用也很可解释）：

会话长度：msg_count > 10 → 上下文长，综合难度更高
最近输入长度：last_len > 120 → 描述长，往往需要分解规划
关键词命中：如“证明/推导/规划/多步骤/step-by-step/chain of thought/数学/逻辑证明/约束求解”等 → 判定复杂

结论：复杂 → deepseek-reasoner，否则 → deepseek-chat

3.3 触发示例

简单问题（走 chat）

messages = {"messages": [{"role": "user", "content": "你好，请介绍下你自己。"}]}
result = agent.invoke(messages)

复杂问题（走 reasoner）

complex_question = """请帮我详细推理以下数学问题：
假设一个粒子以恒定加速度a沿直线运动，初速度为v0，位移为s。
推导出其速度与时间t的函数关系，并逐步解释每一步的物理意义。
"""

result = agent.invoke({
    "messages": [HumanMessage(content=complex_question)]
})

4. 消息压缩（Message Compression）

目标：对话变长会导致 token 爆炸、成本上升、甚至超上下文限制。
LangChain 1.0 在中间件层面提供多种策略控制“历史消息规模”。

Notebook 给了三种策略：Trimming / Deleting / Summarization。

4.1 修剪消息 Trimming（最轻量）

核心思想：只保留最近 N 条消息或 M 个 token 内上下文，其余裁剪。

常配合 @before_model：每次模型调用前检查 token/消息条数，接近阈值就裁剪
优点：简单快速、成本可控
缺点：可能丢失远期记忆

Notebook 示例代码（原 cell 中间用 ... 省略了裁剪细节）：

from langchain_deepseek import ChatDeepSeek
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from typing import Any

model = ChatDeepSeek(model="deepseek-chat")

@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """在模型调用前修剪消息历史，只保留前1+后3条。"""
    messages = state["messages"]
    ...

配合 checkpointer（用于同一 thread 的会话记忆）：

config = {
    "configurable": {
        "thread_id": "2"
    }
}
agent.invoke({"messages": "你好，我叫陈明"}, config)
agent.invoke({"messages": "帮我写一句每日格言"}, config)
agent.invoke({"messages": "请介绍下你自己。"}, config)
agent.invoke({"messages": "你还记得我叫什么吗？"}, config)

这里的意图是：多轮对话后检查裁剪是否影响“记忆保持”。

4.2 删除消息 Deleting（更主动、更精确）

核心思想：通过 RemoveMessage 精确指定删除哪些消息，例如：

删除最早 2 条
只删除 tool 调用消息
直接重置整个会话
常配合 @after_model：模型回复后清理历史，让下一轮上下文更干净
适合长周期运行 Agent / 工作流：防止上下文膨胀，并实现“阶段化会话管理”

Notebook 示例（中间也用 ... 省略了删除逻辑细节）：

from langchain_deepseek import ChatDeepSeek
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig

model = ChatDeepSeek(model="deepseek-chat")

@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """模型调用后，删除最早的两条消息"""
    messages = state["messages"]
    ...

调用演示：

config: RunnableConfig = {"configurable": {"thread_id": "session-1"}}

agent.invoke({"messages": "你好，我叫小明"}, config)
agent.invoke({"messages": "请记住我的名字"}, config)
agent.invoke({"messages": "写一首关于春天的诗"}, config)
final = agent.invoke({"messages": "你还记得我叫什么吗？"}, config)
print(final["messages"][-1].content)

4.3 汇总消息 Summarization（最智能的压缩）

核心思想：不直接丢弃历史，而是：

调用一个较便宜的模型对历史对话做摘要
用摘要替代长历史，再拼回当前上下文
→ 保留“长期记忆的语义”，压缩 token 成本

LangChain 1.0 提供 SummarizationMiddleware：

max_tokens_before_summary：超过阈值触发摘要
model：摘要模型（通常更便宜）
messages_to_keep：摘要后仍保留最近若干条原始消息

Notebook 代码：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig

main_model = "deepseek-reasoner"   # 主模型（执行任务）
summary_model = "deepseek-chat"   # 摘要模型（更省）

agent = create_agent(
    model=main_model,
    tools=[],
    middleware=[
        SummarizationMiddleware(
            model=summary_model,
            max_tokens_before_summary=3000,
            messages_to_keep=10,
        )
    ],
    checkpointer=InMemorySaver(),
)

触发演示：

config: RunnableConfig = {"configurable": {"thread_id": "summary-demo"}}

agent.invoke({"messages": "你好，我叫小明"}, config)
agent.invoke({"messages": "请写一首关于夏天的诗"}, config)
agent.invoke({"messages": "现在写一篇关于秋天的随笔"}, config)
agent.invoke({"messages": "再帮我写一段冬天的散文"}, config)
final = agent.invoke({"messages": "你能总结一下四季的特点吗？"}, config)

print(final["messages"][-1].content)

5. 借助中间件实现人在闭环（Human in the Loop, HITL）

5.1 为什么需要 HITL？

Notebook 的核心观点（原文中有 ... 省略，但语义明确）：

在高风险任务中（数据库写入、财务操作、邮件发送等）
完全让大模型自主执行可能不安全/不合规
HITL 允许在关键动作前插入人工审批：approve / edit / reject
→ 让 Agent 行为符合企业安全、合规、伦理要求

5.2 实现方式：HumanInTheLoopMiddleware 拦截工具调用

Notebook 示例：拦截 Tavily 搜索工具执行前，要求人工确认：

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

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "tavily_search_results_json": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                    "description": lambda tool_name, tool_input, state: (
                        f"🔍 模型准备执行 Tavily 搜索：'{tool_input.get('query', '')}'"
                    ),
                }
            },
            description_prefix="⚠️ 工具执行需要人工审批"
        )
    ],
)

运行配置：

config = {
    "configurable": {
        "thread_id": "23"
    }
}

调用：

result = agent.invoke(
    {"messages": "你好，请问2024年诺贝尔奖得主有哪些？"},
    config
)

Notebook 还给了两张部署后效果截图（外链图片），用于展示实际 UI 中如何进行 approve/edit/reject 的交互。

6. 本节关键要点总结（浓缩记忆版）

Middleware = Agent 运行时“可插拔控制层”：把“定制点”从业务代码抽离出来，增强可维护性与可观测性。
三大模型钩子：before_model（前置处理）→ modify_model_request（改请求）→ after_model（后处理）
两大包装钩子：
- wrap_model_call：动态换模型、重试/降级/熔断、A/B、回退
- wrap_tool_call：工具权限、超时、重试、审核、风控
动态模型路由：用启发式规则决定走 chat 还是 reasoner，性价比很高
消息压缩三件套：
- Trimming：快但可能丢记忆
- Deleting：精确控制，适合阶段化工作流
- Summarization：最智能，适合长期对话
HITL：对高风险工具/动作加人工审批闭环（approve/edit/reject）