当前位置：首页 > news >正文

LangChain1.0系列：中间件深度解析，让 AI智能体上下文控制不失控

news 2025/11/14 6:57:25

相信很多用 LangChain 构建 AI Agent 的开发者都有过这样的经历：本地测试时 Agent 表现得聪明又可靠，能精准理解需求、调用工具完成任务，但一旦部署到生产环境，各种问题就接踵而至。有时候 Agent 会因为对话历史太长而忽略关键信息，有时候会未经授权调用敏感工具，还有时候会因为输入里的敏感数据导致隐私泄露，最后不得不写一堆杂乱的自定义代码来“打补丁”，但效果往往不尽如人意。

这背后的核心问题，其实是上下文管理的失控。AI Agent 的本质是通过对上下文的理解来决策和行动，而上下文工程的核心就是“信息的精细化管理”，该给 Agent 看什么信息、什么时候给、给多少，这些细节直接决定了 Agent 的行为边界和可靠性。在 LangChain v1.0 之前，框架对上下文的支持缺乏系统化的设计，开发者只能在参数配置和手写逻辑之间挣扎，而 v1.0 引入的中间件机制，正是为了解决这个痛点，给上下文管理提供了一套优雅且可扩展的解决方案。
在这里插入图片描述

一、中间件是什么？从 FastAPI 开发者熟悉的逻辑说起

对于熟悉 FastAPI 的开发者来说，理解 LangChain 中间件几乎没有门槛，因为两者的设计思想和底层逻辑高度一致。在 FastAPI 中，中间件是介于 HTTP 请求和接口处理函数之间的“拦截器”，可以在请求到达接口前做身份认证、日志记录，也可以在响应返回前做数据格式化、跨域处理；而在 LangChain 中，中间件扮演的是“Agent 执行流程的协调者”角色，拦截的是用户输入到 AI 模型的调用链路，在不同阶段对信息进行加工和控制。

具体来说，FastAPI 的执行流程是“Request → 中间件链 → 接口处理器 → Response”，而 LangChain 中间件的执行流程则是“用户输入 → 中间件链 → AI 模型/工具 → 响应”。两者都遵循“注册顺序即执行顺序”的规则，多个中间件可以组合成一条处理链，每个中间件负责一个特定的功能，既相互独立又能协同工作。这种设计模式的优势在于，它把复杂的流程拆解成一个个模块化的功能单元，让扩展和维护变得简单直观。

如果用生活中的场景来类比，中间件就像是 AI Agent 的“专属助理团队”：有的助理负责筛选信息，把无关紧要的内容过滤掉；有的助理负责保护隐私，把敏感数据脱敏；有的助理负责管理工具，只给 Agent 开放必要的权限；还有的助理负责记录日志，方便后续排查问题。这些助理按照既定顺序协同工作，确保 Agent 接收到的是“干净、准确、安全”的上下文，从而做出可靠的决策。

在 LangChain v1.0 中，中间件可以在 Agent 执行的四个核心阶段介入：

before_model：模型调用前执行，主要用于输入预处理，比如清洗用户输入中的特殊字符、标准化格式、注入背景知识等；
wrap_model_call：包裹模型调用过程，能够修改传递给模型的参数，比如动态调整可用工具列表、切换模型版本、设置 Token 上限等；
wrap_tool_call：拦截工具调用请求，负责工具的权限校验、参数验证、调用日志记录等；
after_model：模型返回响应后执行，主要用于输出校验、安全检查、结果格式化等。

这四个阶段基本覆盖了 Agent 执行的全流程，让开发者能够在任意节点对上下文和执行逻辑进行干预，实现对 Agent 行为的精细化控制。

二、v1.0 之前的“痛点”：上下文管理的混乱时代

在中间件机制出现之前，构建生产级 AI Agent 堪称“步步惊心”。很多时候 Agent 失败并不是因为模型能力不足，而是上下文处理环节出了问题，这些问题主要集中在四个方面：

首先是上下文切换的灵活性不足。不同场景需要不同的上下文策略：比如面对新用户时，需要提供详细的引导信息；面对老用户时，需要加载历史对话摘要；处理简单查询时，上下文要简洁高效；处理复杂任务时，需要注入更多背景知识。但在 v1.0 之前，想要实现这些动态切换，只能通过大量的条件判断语句，代码臃肿且难以维护。

其次是长对话的上下文管理难题。大语言模型都有 Token 限制，当对话历史过长时，要么会超出 Token 上限导致调用失败，要么会让模型淹没在海量信息中无法聚焦关键内容。之前的解决方案通常是手动编写摘要逻辑，比如当对话消息数超过一定阈值时，调用模型生成历史摘要，再替换掉原始对话，但这种方式需要手动控制时机和逻辑，容易出现遗漏或过度摘要的问题。

第三是工具调用权限的控制困境。一个 Agent 可能配置了多个工具，比如搜索工具、数据库查询工具、邮件发送工具等，但在实际使用中，并不是所有工具都适合在所有场景下开放。比如普通用户不应该拥有数据库写入权限，临时任务不应该调用邮件发送工具，但之前的实现方式要么是在创建 Agent 时固定工具列表，要么是在工具内部做权限判断，缺乏一种全局、统一的权限控制机制，导致工具调用的安全性难以保障。

最后是配置繁琐和代码复用性差。v1.0 之前创建 Agent 时，需要在 AgentExecutor 中配置大量参数，比如 max_iterations（最大尝试次数）、max_execution_time（最长执行时间）、handle_parsing_errors（解析错误处理）、trim_intermediate_steps（中间步骤裁剪数量）等，这些参数相互关联，调整起来非常麻烦，而且不同项目的配置逻辑无法复用。更极端的情况是，为了实现复杂的上下文控制，开发者不得不完全手写 Agent 循环，从对话历史管理、工具选择到模型调用，每一步都要手动实现，不仅代码可读性差，而且维护成本极高。

我们来看一段典型的旧版实现代码：

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI# 配置工具列表
tools = [search_tool, calculator_tool, database_tool]
# 初始化模型
llm = ChatOpenAI(model="gpt-4")
# 创建 Agent
agent = create_openai_tools_agent(llm, tools, prompt)
# 配置 AgentExecutor，参数繁琐且难以理解
agent_executor = AgentExecutor(agent=agent,tools=tools,max_iterations=15,              # 最大尝试次数max_execution_time=300,         # 最长执行时间（秒）handle_parsing_errors=True,     # 是否处理解析错误return_intermediate_steps=True, # 是否返回中间步骤trim_intermediate_steps=10,     # 保留多少个中间步骤verbose=True,# 还有更多令人困惑的参数...
)

如果这些参数还满足不了需求，就只能手写 Agent 循环：

def custom_agent_loop(user_input, tools, llm):# 初始化对话历史messages = [{"role": "user", "content": user_input}]# 手动控制迭代次数for iteration in range(10):# 手动处理长对话：消息数超过20条则生成摘要if len(messages) > 20:messages = summarize_history(messages)# 手动选择可用工具available_tools = select_tools(messages, tools)# 手动生成系统提示词system_prompt = generate_prompt(iteration, available_tools)messages.insert(0, {"role": "system", "content": system_prompt})# 手动调用模型response = llm.invoke(messages)# 手动处理模型响应：判断是否需要调用工具if need_call_tool(response):tool_name, tool_args = parse_tool_call(response)# 手动调用工具tool_result = call_tool(tool_name, tool_args, available_tools)# 手动将工具结果加入上下文messages.append({"role": "tool", "name": tool_name, "content": tool_result})else:# 不需要调用工具则返回结果return response.content# 迭代次数用尽仍未完成任务return "任务处理超时，请稍后再试"

这种实现方式的问题显而易见：代码逻辑分散，上下文管理、工具选择、错误处理等功能混在一起，一旦需求变化，比如要新增“敏感信息过滤”功能，就需要在循环中插入相关代码，导致代码越来越臃肿。而且不同项目之间的自定义循环无法复用，每个项目都要重复造轮子，效率极低。

三、中间件的“魔法”：用模块化思维解决上下文难题

LangChain v1.0 引入的中间件机制，彻底改变了这种混乱的局面。它借鉴了软件工程中的模块化设计思想，把上下文管理、安全控制、工具调度等功能拆分成独立的中间件，开发者可以像搭积木一样，根据需求组合不同的中间件，快速实现复杂的业务逻辑。

3.1 内置中间件：开箱即用的常见场景解决方案

LangChain v1.0 提供了多个常用的内置中间件，覆盖了生产环境中最常见的场景，开发者不需要编写复杂代码，只需简单配置即可使用。

我们来看一个实际的使用示例，这个示例实现了敏感信息保护、长对话摘要和关键操作人工审核三个核心功能：

from langchain.agents import create_agent
from langchain.agents.middleware import (PIIMiddleware,SummarizationMiddleware,HumanInTheLoopMiddleware
)
from langchain_openai import ChatOpenAI# 定义工具：读取邮件和发送邮件
tools = [read_email_tool, send_email_tool]# 初始化模型
llm = ChatOpenAI(model="gpt-4-turbo")# 使用 create_agent 函数创建 Agent，核心是中间件配置
agent = create_agent(llm=llm,tools=tools,# 中间件列表：按执行顺序配置middleware=[# 第一个中间件：处理敏感信息——脱敏邮箱地址PIIMiddleware(pii_types=["email"],  # 要处理的敏感信息类型：邮箱strategy="redact"     # 处理策略：脱敏（将邮箱替换为[EMAIL]）),# 第二个中间件：处理敏感信息——阻断电话号码PIIMiddleware(pii_types=["phone_number"],  # 要处理的敏感信息类型：电话号码strategy="block"             # 处理策略：阻断（直接移除包含电话号码的内容）),# 第三个中间件：长对话摘要——避免 Token 超限SummarizationMiddleware(summary_llm=ChatOpenAI(model="gpt-4-turbo"),  # 用于生成摘要的模型max_tokens_before_summary=500,                # 触发摘要的 Token 阈值summary_prompt="请简洁总结以下对话的核心内容，保留关键信息和用户需求：\n{content}"  # 摘要提示词),# 第四个中间件：关键操作人工审核——发送邮件前需要人工批准HumanInTheLoopMiddleware(interrupt_on={"send_email_tool": {  # 对 send_email_tool 工具进行拦截"allowed_decisions": ["approve", "edit", "reject"],  # 允许的人工决策：批准、编辑、拒绝"prompt": "即将发送以下邮件，请审核：\n收件人：{to}\n主题：{subject}\n内容：{content}\n请选择操作："  # 人工审核提示}})]
)

这段代码看似简单，却解决了生产环境中三个关键问题：

第一个是隐私保护问题。PIIMiddleware 支持多种敏感信息类型，比如邮箱、电话号码、身份证号、银行卡号等，处理策略包括脱敏（redact）、阻断（block）、加密（encrypt）三种。在上面的配置中，邮箱地址会被替换为[EMAIL]，而包含电话号码的内容会被直接移除，确保敏感信息不会泄露给模型或存储在对话历史中。

第二个是长对话管理问题。SummarizationMiddleware 会实时监控对话的 Token 数量，当累计 Token 数超过 500 时，会自动调用指定的模型生成对话摘要，并用摘要替换原始的历史对话，既避免了 Token 超限，又保留了关键信息。这里需要注意的是，摘要模型可以和主模型不同，比如用轻量版模型生成摘要，既能节省成本，又能提高处理速度。

第三个是关键操作安全问题。HumanInTheLoopMiddleware 可以指定需要拦截的工具，当 Agent 要调用这些工具时，会暂停执行并向人工发起审核请求，只有获得人工批准后才会继续执行。这种机制在处理高风险操作时非常重要，比如发送邮件、修改数据库、转账等，能够有效避免 Agent 因逻辑错误或恶意输入导致的损失。

除了这三个中间件，LangChain v1.0 还提供了其他常用的内置中间件：

TokenBudgetMiddleware：Token 预算控制，限制单次对话的 Token 消耗，避免超出成本预算；
CacheMiddleware：响应缓存，缓存常见查询的结果，提高响应速度并降低调用成本；
ErrorHandlingMiddleware：错误处理，统一捕获模型调用、工具调用过程中的异常，并返回友好提示；
LoggingMiddleware：日志记录，记录 Agent 执行过程中的关键信息，比如输入输出、工具调用详情、中间件执行结果等，方便后续排查问题。

这些内置中间件覆盖了隐私保护、Token 管理、安全控制、错误处理、性能优化等核心场景，开发者可以根据项目需求灵活组合使用，无需从零开始编写代码。

3.2 自定义中间件：满足个性化业务需求

内置中间件虽然强大，但无法覆盖所有个性化场景。LangChain v1.0 提供了灵活的自定义中间件接口，让开发者能够根据自己的业务需求，实现专属的上下文控制逻辑。

自定义中间件的核心是继承 AgentMiddleware 类，并实现对应的介入方法（比如 wrap_model_call、wrap_tool_call 等）。下面我们通过一个实际案例，详细讲解自定义中间件的实现过程。

案例需求

假设我们正在开发一个面向不同技术水平用户的 AI 助手：

对于初学者用户，只开放基础工具（简单搜索、基础计算器），并使用轻量模型（gpt-5-nano），同时在系统提示词中加入详细的操作引导；
对于专家用户，开放高级工具（高级搜索、数据分析工具、数据库查询），并使用高性能模型（gpt-5），系统提示词简洁高效，不添加多余引导。

我们需要实现一个中间件，根据用户的技术水平标识，动态调整 Agent 的模型、工具列表和系统提示词。

实现步骤

定义上下文数据类：存储用户的技术水平标识；
实现自定义中间件：继承 AgentMiddleware，重写 wrap_model_call 方法，根据用户技术水平动态调整配置；
使用自定义中间件：在 create_agent 中配置该中间件，验证效果。

完整代码实现

from dataclasses import dataclass
from typing import Callable, Optional
from langchain.agents.middleware import AgentMiddleware
from langchain.agents.middleware.types import ModelRequest, ModelResponse
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent# 1. 定义上下文数据类：存储用户的技术水平标识
@dataclass
class UserContext:user_expertise: str = "beginner"  # 技术水平：beginner（初学者）、expert（专家）user_id: Optional[str] = None     # 可选：用户ID，用于日志记录# 2. 实现自定义中间件
class ExpertiseBasedConfigMiddleware(AgentMiddleware):def __init__(self):# 初始化不同用户类型的配置# 初学者配置：轻量模型 + 基础工具 + 详细引导提示词self.beginner_config = {"model": ChatOpenAI(model="gpt-5-nano", temperature=0.3),"tools": ["simple_search_tool", "basic_calculator_tool"],"system_prompt": "你是一个面向初学者的AI助手，需要用简单易懂的语言解释问题，提供详细的操作步骤，避免使用专业术语。如果用户的问题需要调用工具，要明确告诉用户工具的使用逻辑。"}# 专家配置：高性能模型 + 高级工具 + 简洁提示词self.expert_config = {"model": ChatOpenAI(model="gpt-5", temperature=0.7),"tools": ["advanced_search_tool", "data_analysis_tool", "database_query_tool"],"system_prompt": "你是一个面向专家的AI助手，直接提供精准、高效的答案和工具调用结果，无需多余引导。"}def wrap_model_call(self,request: ModelRequest,handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse:# 从运行时上下文中获取用户技术水平user_context: UserContext = request.runtime.context# 根据用户技术水平选择对应的配置if user_context.user_expertise == "expert":config = self.expert_configelse:config = self.beginner_config# 动态调整模型request.model = config["model"]# 动态调整可用工具（过滤掉不在配置中的工具）request.tools = [tool for tool in request.tools if tool.name in config["tools"]]# 动态调整系统提示词（插入到对话历史的最前面）system_message = {"role": "system", "content": config["system_prompt"]}# 移除已有的系统提示词（避免重复）request.messages = [msg for msg in request.messages if msg.get("role") != "system"]# 添加新的系统提示词request.messages.insert(0, system_message)# 打印日志（方便调试和监控）print(f"用户 {user_context.user_id} - 技术水平：{user_context.user_expertise}，使用模型：{config['model'].model_name}，可用工具：{config['tools']}")# 继续执行后续流程return handler(request)# 3. 定义工具（此处仅为示例，实际需实现工具的具体逻辑）
class SimpleSearchTool:name = "simple_search_tool"description = "简单搜索工具，用于查询基础信息"def invoke(self, query):return f"简单搜索结果：{query} 的基础信息..."class AdvancedSearchTool:name = "advanced_search_tool"description = "高级搜索工具，用于查询专业、深度信息"def invoke(self, query):return f"高级搜索结果：{query} 的专业分析报告..."class BasicCalculatorTool:name = "basic_calculator_tool"description = "基础计算器工具，用于简单的加减乘除运算"def invoke(self, expr):return f"基础计算结果：{expr} = {eval(expr)}"class DataAnalysisTool:name = "data_analysis_tool"description = "数据分析工具，用于处理和分析结构化数据"def invoke(self, data):return f"数据分析结果：{data} 的统计分析报告..."class DatabaseQueryTool:name = "database_query_tool"description = "数据库查询工具，用于查询和操作数据库"def invoke(self, sql):return f"数据库查询结果：执行 SQL '{sql}' 得到的数据..."# 初始化所有工具
tools = [SimpleSearchTool(),AdvancedSearchTool(),BasicCalculatorTool(),DataAnalysisTool(),DatabaseQueryTool()
]# 4. 使用自定义中间件创建 Agent
agent = create_agent(llm=ChatOpenAI(model="gpt-4-turbo"),  # 默认模型（会被中间件覆盖）tools=tools,middleware=[ExpertiseBasedConfigMiddleware()  # 配置自定义中间件],context_schema=UserContext  # 指定上下文数据类
)# 5. 测试不同技术水平用户的请求
def test_agent(user_input, user_expertise, user_id):# 构建运行时上下文runtime_context = UserContext(user_expertise=user_expertise,user_id=user_id)# 调用 Agentresponse = agent.invoke(input={"content": user_input},runtime_context=runtime_context)print(f"\n=== 用户 {user_id}（{user_expertise}）的请求 ===")print(f"输入：{user_input}")print(f"输出：{response['output']}")# 测试初学者用户
test_agent(user_input="如何用 Python 计算 1+2+3+4+5？",user_expertise="beginner",user_id="user_001"
)# 测试专家用户
test_agent(user_input="用 Python 分析销售数据，计算月度增长率并生成可视化图表",user_expertise="expert",user_id="user_002"
)

代码解析

这个自定义中间件的核心逻辑在 wrap_model_call 方法中，它通过 request.runtime.context 获取用户的技术水平标识，然后动态调整模型、工具列表和系统提示词：

模型调整：初学者使用 gpt-5-nano（轻量、低成本），专家使用 gpt-5（高性能、高准确率）；
工具过滤：根据配置的工具名称列表，过滤掉不可用的工具，确保用户只能调用权限范围内的工具；
提示词调整：为不同技术水平的用户提供个性化的系统提示词，提升用户体验。

运行测试代码后，我们会看到：

初学者用户的请求会使用基础工具和详细引导，比如计算 1+2+3+4+5 时，会用基础计算器工具，并给出详细的操作步骤；
专家用户的请求会使用高级工具和简洁提示词，比如分析销售数据时，会调用数据分析工具，直接返回专业的分析结果。

这种实现方式的优势在于，中间件是独立的模块，不需要修改 Agent 的核心逻辑，后续如果要新增“中级用户”类型，只需在中间件中添加对应的配置即可，扩展性极强。

四、中间件架构的核心优势：为什么它能彻底改变 Agent 开发

LangChain v1.0 的中间件机制不仅仅是新增了几个功能，更是对 Agent 开发范式的重新定义。它通过模块化、可组合的设计，解决了传统开发模式中的诸多痛点，其核心优势体现在以下五个方面：

4.1 代码组织更规范，逻辑边界更清晰

中间件机制强制开发者将不同功能拆分成独立的模块，每个中间件只负责一个特定的职责，比如 PIIMiddleware 只处理敏感信息，SummarizationMiddleware 只处理长对话摘要，避免了功能逻辑的耦合。这种“单一职责原则”让代码结构更清晰，可读性更强，后续维护时能够快速定位到相关逻辑。

比如在之前的示例中，隐私保护、长对话管理、人工审核、用户分级这四个功能分别由四个独立的中间件实现，每个中间件的代码都可以单独维护和测试，不会因为修改一个功能而影响其他功能。

4.2 复用性极强，开发效率大幅提升

中间件是独立的可复用组件，一旦开发完成，就可以在不同项目中直接使用。比如我们实现的 ExpertiseBasedConfigMiddleware，不仅可以用在 AI 助手项目中，还可以用在其他需要根据用户属性动态调整 Agent 配置的场景中，无需重复编写代码。

LangChain 官方也在不断丰富中间件生态，开发者可以通过社区共享中间件，进一步降低开发成本。比如需要实现“多语言支持”功能时，不需要自己编写翻译逻辑，直接使用社区提供的 TranslationMiddleware 即可。

4.3 组合灵活，快速适配复杂场景

中间件支持按顺序组合，开发者可以根据业务需求灵活搭配不同的中间件，快速实现复杂的功能。比如：

简单场景：只需要隐私保护 + 错误处理 → 组合 PIIMiddleware + ErrorHandlingMiddleware；
复杂场景：隐私保护 + 长对话摘要 + 人工审核 + Token 预算控制 → 组合 PIIMiddleware + SummarizationMiddleware + HumanInTheLoopMiddleware + TokenBudgetMiddleware。

这种组合式设计让 Agent 能够快速适配不同的业务场景，不需要重构核心代码，只需调整中间件列表即可。

4.4 测试和调试更简单

由于每个中间件都是独立的模块，测试时可以单独验证每个中间件的功能是否正常，再进行整体联调。比如要测试 PIIMiddleware 的脱敏效果，只需单独运行该中间件，输入包含敏感信息的文本，检查输出是否符合预期，无需启动整个 Agent 流程。

同时，中间件可以添加详细的日志记录，方便调试时追踪信息流向。比如在 ExpertiseBasedConfigMiddleware 中打印用户类型、使用的模型和工具，能够快速定位到配置是否生效。

4.5 生产环境适配性更强

中间件机制天生为生产环境设计，覆盖了生产级 Agent 所需的核心能力：

安全性：通过 PIIMiddleware 保护隐私，HumanInTheLoopMiddleware 控制高风险操作，TokenBudgetMiddleware 控制成本；
可靠性：通过 ErrorHandlingMiddleware 处理异常，CacheMiddleware 提高响应稳定性；
可监控性：通过 LoggingMiddleware 记录执行日志，方便后续分析和优化；
可扩展性：支持自定义中间件，能够快速适配业务变化和新需求。

这些特性让开发者能够更自信地将 Agent 部署到生产环境，不用担心上下文管理失控、权限泄露、成本超支等问题。