摘要：本文围绕 LangChain Agent 中间件机制展开，通过年会筹备的比喻说明中间件如何监控、管理、限制和兜底 Agent 的执行流程，并介绍 LangChain 提供的默认中间件清单。文章重点讲解摘要中间件（SummarizationMiddleware）的工作原理、触发条件（tokens、messages、fraction）与保留策略（keep），最后给出完整的 Python 代码示例，演示如何配置摘要中间件并验证其效果。
内容参考于：图灵AI大模型全栈
中间件
BOSS就是用户，它向Agent提出问题发布任务，任务是使用30万元完成200人年会，Agent收到任务时就要思考怎么30万举办年会，然后BOSS不放心Agent干活，BOSS就安排秘书去监控Agent，全程记录Agent做的事情，也就是一个日志中间件，这个日志中间件并不会响应Agent做的事情，它只会记录，让后续支持回溯，如果出现问题要一眼可以看到，然后财务中间件，当Agent花费超过1万元就需要审批，也就是一个权限的控制，就是当Agent花费超过1万元是财务中间件就会得到消息，然后就限制Agent等审批，等审批通过了才可以进一步进行，然后备选方案中间件，比如Agent思考了一会选定了场地，准备预定的时候发现场地没有档期，这时候就需要备选方案中间件来重新选择场地，或者说当模型欠费了没法用了就需要一个新的模型来顶替上这都是备选方案中间件做的事情用来兜底
中间件它并不会直接给Agent做事情，它们管理整个Agent的执行流程，Agent负责把事情对正确，中间件负责Agent做事过程中不出现问题
在Agent做任务时，我们可以在任务过程中设置很多个中间件来监控、管理、限制、兜底Agent，从而帮助Agent完成任务
LangChain给我提供了一些默认的中间件
| 摘要生成 (Summarization) | 当接近 Token 限制时，自动总结对话历史。 |
| 人工介入 (Human-in-the-loop) | 暂停执行，等待人工批准工具调用。 |
| 模型调用限制 (Model call limit) | 限制模型调用次数，以防止Agent陷入循环成本过高。 |
| 工具调用限制 (Tool call limit) | 通过限制调用次数来控制工具的执行。以防止Agent陷入循环成本过高。 |
| 模型回退 (Model fallback) | 当主模型失败时，自动回退到备用模型。 |
| 个人身份信息检测 (PII detection) | 检测并处理个人身份信息 (PII)。 |
| 待办事项列表 (To-do list) | 为 Agent（智能体）配备任务规划与跟踪能力。 |
| LLM工具选择器 (LLM tool selector) | 在调用主模型之前，使用 LLM 选择相关的工具。 |
| 工具重试 (Tool retry) | 使用指数退避自动重试失败的调用。 |
| 模型重试 (Model retry) | 使用指数退避自动重试失败的模型调用。 |
| LLM工具模拟器 (LLM tool emulator) | 使用 LLM 模拟工具执行，以进行测试。 |
| 上下文编辑 (Context editing) | 通过修剪或清除工具使用记录来管理对话上下文。 |
| Shell工具 (Shell tool) | 向 Agent 暴露持久化的 Shell 会话以执行命令。 |
| 文件搜索 (File search) | 提供对文件系统文件的 Glob 和 Grep 搜索工具。 |
| 文件系统 (Filesystem) | 为 Agent 提供用于存储上下文和长期记忆的文件系统。 |
| 子Agent (Subagent) | 增加派生（生成）子 Agent 的能力。 |
摘要中间件
它做了一个消息总结的功能，当对话长度超过模型上下文时LangChain会自动给我们压缩上下文，现在模型的上下文长度一般是100万token
触发条件和说明，它有 tokens、messages、fraction 这三个条件，如下图红框的中文说明
它的触发条件可以传递多个，所以它存在or和and的，如下图红框的示例使用中括号它们的关系是or，使用大括号它们的关系是and
下图红框会设置一个提示词，如果这个提示词写的不好，那么Agent会胡说八道
上图红框的提示词可以参考LangChain给我们的默认提示词来写，LangChain的提示词就写的很详细
效果图
下图红框是摘要中间件，下图蓝框是给摘要中间件输入的内容，下图绿框是摘要中间件输出的内容，可以看出它把下图蓝框的多条内容总结成了下图绿框的三条，下图绿框里的第一条是总结之后的内容，剩下的两个是保留的内容
如下图红框，代码trigger中超过4000token后超过10条上下文就进行总结，然后keep保留两条摘要之后最新的两条数据
如下图红框保留的内容，绿框是输入的内容，是要被摘要的内容，下图红框是摘要之后保留最新的两条数据
代码
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
import os
加载环境变量
load_dotenv()
model = "qwen3.7-flash"
api_key = os.getenv("DASHSCOPE_API_KEY")
api_base_url = os.getenv("DASHSCOPE_BASE_URL")
llm = ChatOpenAI(
model=model,
api_key=api_key,
base_url=api_base_url,
temperature=0.1
)
2. 配置工具，TavilySearch是联网搜索的工具LangChain给我写好了工具代码，所以可以直接使用
tools = [TavilySearch(max_results=1)]
"""
SummarizationMiddleware(
model=llm, # 进行摘要的模型
# 当上下文达到4000token后，它就会给我们进行摘要
trigger=("tokens", 4000),
# 条件控制要保留多少上下文信息 只能选择一个
1.fraction- 要保留的模型上下文大小的比例
2.tokens- 要保留的绝对令牌数量
3.messages- 要保留的最近消息数量
keep=("messages", 20),
),
"""
它默认的提示词会返回英文的内容
SHORT_SUMMARY_PROMPT = """你是一个记忆压缩专家。
请将下方的对话历史压缩成一段简洁的背景摘要，保留以下核心：
用户最终想要解决的问题是什么？
已经执行了哪些关键步骤或得到了哪些结论？
还有哪些待办事项？
请直接输出摘要内容，不要包含任何开场白。
待压缩的对话：
{messages}
"""
agent = create_agent(
model=llm,
tools=tools,
middleware=[
SummarizationMiddleware(
model=llm,
# 它默认的提示词会返回英文的内容
SHORT_SUMMARY_PROMPT=SHORT_SUMMARY_PROMPT,
# trigger：是摘要触发的条件
# tokens：当上下文达到4000token后，它就会给我们进行摘要
# messages：消息条数，这里消息条数达到10条就进行总结
# fraction：它是当达到模型最大输入 token 数量的 80%，默认百分之80，这里用的默认值，百分之80就是0.8
trigger=[("tokens", 4000), ("messages", 10)],
# 摘要后要保留多少上下文信息，它只能从tokens、messages、fraction中选择一个作为保留
# 下方 "messages", 2 表示摘要之后保留两条消息
# tokens 的意思摘要之后保留多少token
# fraction的意思是摘要之后保留多少比例
keep=("messages", 2),
),
],
checkpointer=InMemorySaver()
)
def run_test():
print("=== 开始 Agent 自动化测试 ===")
config = {"configurable": {"thread_id": "1"}}
# 场景 1：基础问答 + 工具调用（验证 Tavily 搜索是否正常）
print("\\n[测试点 1: 工具调用]")
query_1 = "2026年3月最新的AI大模型技术趋势是什么？请列出3-4点 简单总结内容"
print(f"用户: {query_1}")
response_1 = agent.invoke({"messages": [{"role": "user", "content": query_1}]}, config)
print(f"Agent 响应: {response_1}")
# 场景 2：连续对话（验证上下文保留与中间件触发）
# 我们故意发送一些长文本，模拟达到 4000 tokens 或 3 条消息的触发条件
print("\\n[测试点 2: 多轮对话与摘要中间件验证]")
test_conversations = [
"请记住我的名字叫‘浩英’，我是一名AI架构师。",
"刚才我问的技术趋势中，哪个对医疗行业影响最大？",
"请基于我们刚才聊到的所有内容，给我写一份200字的行业简报。"
]
for i, user_input in enumerate(test_conversations):
print(f"\\n第 {i + 2} 轮对话输入: {user_input}")
# 执行对话
res = agent.invoke({"messages": [{"role": "user", "content": user_input}]}, config)
print(f"Agent 响应: {res}")
print("\\n=== 测试完成 ===")
if name == "main":
# 可以开启 LangChain 的调试模式查看中间件运行细节
# import langchain
# langchain.debug = True
try:
run_test()
except Exception as e:
print(f"测试过程中出现错误: {e}")
