AI代理中间件：构建安全智能系统的核心技术

1. Middleware中间件：AI代理的智能控制塔

在构建AI代理系统时，我们常常面临一个关键矛盾：既希望AI能够自主决策和执行任务，又需要对它的行为进行必要的约束和监控。Middleware（中间件）正是解决这一矛盾的完美方案。它就像给AI代理装上了"控制塔"，在不影响核心功能的前提下，为系统添加了安全检查、性能监控、错误处理等关键能力。

想象一下，你有一个能干的AI助手，它可以自主思考、决策和执行任务。但如果完全放任它自由行动，可能会遇到以下问题：

• 用户不小心泄露了敏感信息（如手机号、邮箱）
• 对话历史过长导致模型响应变慢
• 需要人工审核某些敏感操作
• 不同用户需要不同的功能权限

Middleware就是为解决这些问题而生的。它通过在AI代理的关键执行节点插入"拦截点"，实现了对代理行为的精细控制。这种设计既保留了AI的自主性，又确保了系统的安全性和可靠性。

2. Middleware的核心工作原理

2.1 中间件的执行流程

要理解Middleware如何工作，我们需要先了解LangChain Agent的基本执行流程：

code复制用户请求 → 模型思考 → 决定是否使用工具 → 使用工具 → 观察结果 → 再思考 → 返回结果

Middleware就是在这个循环的各个关键节点上插入监控和控制逻辑。具体来说，它可以在以下6个关键时机进行干预：

2.2 中间件的执行顺序

当使用多个Middleware时，它们的执行顺序非常重要。Middleware会按照添加的顺序形成一个"层层包裹"的结构：

code复制用户请求
├─ 中间件1.before_agent
│  ├─ 中间件2.before_agent
│  │  ├─ Agent核心处理
│  │  └─ 中间件2.after_agent
└─ 中间件1.after_agent
返回结果

这种设计意味着：

1. before_agent按照从外到内的顺序执行
2. after_agent按照从内到外的顺序执行
3. 最先添加的Middleware位于最外层

因此，合理的Middleware排列顺序应该是：

1. 最外层：隐私保护、安全检查等基础中间件
2. 中间层：性能优化、对话管理等功能中间件
3. 最内层：业务逻辑、个性化定制等应用中间件

3. LangChain内置中间件详解

LangChain提供了一系列开箱即用的中间件，下面我们详细解析几个最常用的内置中间件。

3.1 个人信息保护中间件(PIIMiddleware)

PII(Personally Identifiable Information)Middleware能够自动检测和隐藏敏感个人信息，如邮箱、电话号码等。这对于处理用户输入的AI应用至关重要。

核心功能

• 自动检测多种敏感信息类型
• 支持多种处理策略：屏蔽(mask)、阻止(block)、替换(replace)
• 可自定义检测规则和替换内容

典型配置示例

python复制from langchain.agents.middleware import PIIMiddleware

# 创建带PII保护的Agent
agent = create_agent(
    middleware=[
        PIIMiddleware("email", strategy="mask", apply_to_input=True),
        PIIMiddleware(
            "phone_number",
            detector=r"(?:\+?\d{1,3}[\s.-]?)?(?:\(?\d{2,4}\)?[\s.-]?)?\d{3,4}[\s.-]?\d{4}",
            strategy="block"
        )
    ]
)

处理效果对比

实际应用建议

1. 对于邮箱等不太敏感的信息，建议使用mask策略，既保护隐私又不影响用户体验
2. 对于手机号、身份证号等高度敏感信息，建议使用block策略，直接阻止处理
3. 可以通过detector参数自定义正则表达式，适应不同的识别需求
4. apply_to_input参数控制是否处理输入内容，通常应该设为True

3.2 对话摘要中间件(SummarizationMiddleware)

当对话轮次过多或内容过长时，SummarizationMiddleware可以自动生成对话摘要，避免上下文过长导致的性能问题。

核心功能

• 基于轮次或token数自动触发摘要
• 可配置保留的原始消息数量
• 支持自定义摘要模型和提示词

典型配置示例

python复制from langchain.agents.middleware import SummarizationMiddleware

middleware = SummarizationMiddleware(
    model=llm,
    trigger=[
        ("messages", 4),  # 对话超过4轮后触发
        ("tokens", 1000)  # 或总Token数超过1000后触发
    ],
    keep=("messages", 2),  # 保留最新的2条原始消息
)

工作流程

1. 监控对话长度和token数
2. 达到触发条件时，调用摘要模型生成摘要
3. 用摘要替换部分历史消息，保留指定数量的最新消息
4. 将摘要作为系统消息插入对话

注意事项

1. 摘要会丢失部分细节信息，不适合需要精确回忆历史对话的场景
2. keep参数不宜设置过小，否则可能导致上下文不连贯
3. 可以结合对话重要性评分等机制，实现更智能的摘要策略
4. 对于专业领域对话，建议使用领域适应的摘要模型

3.3 人工审核中间件(HumanInTheLoopMiddleware)

对于某些敏感操作，HumanInTheLoopMiddleware可以在执行前暂停流程，等待人工审核。

核心功能

• 指定需要审核的工具或操作
• 支持多种审核决策：批准(approve)、编辑(edit)、拒绝(reject)
• 可自定义审核界面和流程

典型配置示例

python复制from langchain.agents.middleware import HumanInTheLoopMiddleware

middleware = HumanInTheLoopMiddleware(
    interrupt_on={
        "send_email": {  # 只有send_email工具需要审核
            "allowed_decisions": ["approve", "edit", "reject"]
        }
    }
)

审核流程

1. Agent准备执行受监控的工具(如send_email)
2. 中间件拦截请求，暂停执行
3. 向审核接口发送审核请求，包含操作详情
4. 等待人工审核决定：
   • approve：继续执行原操作
   • edit：修改后执行
   • reject：取消操作
5. 将审核结果返回给Agent继续处理

最佳实践

1. 只对真正敏感的操作启用审核，避免过度影响用户体验
2. 提供清晰的审核界面，展示完整操作上下文
3. 设置审核超时机制，避免长时间等待
4. 记录完整的审核日志，便于追溯和审计

4. 自定义中间件开发指南

当内置中间件无法满足需求时，我们可以开发自定义中间件。下面通过一个购车推荐场景的案例，演示如何创建业务特定的中间件。

4.1 定义中间件需求

假设我们需要一个根据用户预算动态调整推荐内容的中间件：

• 预算10-20万：只推荐经济型车型
• 预算20-30万：推荐中档车型
• 根据预算自动调整提示词

4.2 实现自定义中间件

python复制from dataclasses import dataclass
from typing import Callable
from langchain.agents.middleware import AgentMiddleware
from langchain.agents.middleware.types import ModelRequest, ModelResponse

@dataclass
class CarContext:
    budget_range: str = "10-20万"

class CarBudgetMiddleware(AgentMiddleware):
    def warp_model_call(
        self,
        model_request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse]
    ) -> ModelResponse:
        # 从上下文中获取预算范围
        budget_range = model_request.runtime.context.budget_range
        
        # 根据预算调整可用工具
        if budget_range == "20-30万":
            model_request.tools = [mid_range_car_recommendation, compare_models]
        else:
            model_request.tools = [economy_car_recommendation, compare_models]
        
        # 添加预算提示
        budget_hint = f"\n[系统提示: 您的购车预算为{budget_range}]"
        model_request.messages[-1].content += budget_hint
        
        return handler(model_request)

4.3 使用自定义中间件

python复制# 创建带自定义中间件的Agent
car_agent = create_agent(
    model=llm,
    tools=[economy_car_recommendation, mid_range_car_recommendation, compare_models],
    middleware=[CarBudgetMiddleware()],
    context_schema=CarContext
)

# 使用不同预算进行测试
response1 = car_agent.invoke({
    "messages": [HumanMessage(content="推荐家庭用车")],
    "runtime": {"context": CarContext(budget_range="10-20万")}
})

response2 = car_agent.invoke({
    "messages": [HumanMessage(content="推荐商务用车")],
    "runtime": {"context": CarContext(budget_range="20-30万")}
})

4.4 自定义中间件开发要点

1. 明确拦截点：确定需要在哪个阶段进行干预(before/after/wrap)
2. 设计上下文结构：使用dataclass等清晰定义中间件需要的上下文数据
3. 保持轻量：中间件逻辑应该尽量简单，复杂业务逻辑应该放在工具或模型中
4. 考虑异常处理：妥善处理可能出现的错误，避免影响主流程
5. 提供配置选项：通过参数化设计提高中间件的复用性

5. Middleware实战：构建安全可靠的AI代理

现在，我们将结合多个Middleware，构建一个既强大又安全的AI代理系统。

5.1 场景需求

假设我们要开发一个客户服务AI代理，需要满足：

1. 自动保护用户隐私信息
2. 长对话自动摘要保持性能
3. 敏感操作需人工审核
4. 根据用户等级提供不同服务

5.2 完整配置示例

python复制from langchain.agents import create_agent
from langchain.agents.middleware import (
    PIIMiddleware,
    SummarizationMiddleware,
    HumanInTheLoopMiddleware
)

# 组合多个中间件
agent = create_agent(
    model=llm,
    tools=[query_order, send_email, upgrade_service],
    middleware=[
        # 第一层：隐私保护
        PIIMiddleware("email", strategy="mask"),
        PIIMiddleware("phone", strategy="block"),
        
        # 第二层：对话管理
        SummarizationMiddleware(
            model=llm,
            trigger=[("messages", 5), ("tokens", 1200)],
            keep=("messages", 3)
        ),
        
        # 第三层：业务逻辑
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {"allowed_decisions": ["approve", "reject"]},
                "upgrade_service": {"allowed_decisions": ["approve", "edit"]}
            }
        ),
        
        # 第四层：权限控制
        UserLevelMiddleware()
    ]
)

5.3 中间件组合最佳实践

1. 分层设计：按照基础→功能→业务的顺序组织中间件
2. 性能考量：将高频中间件放在内层，减少不必要的处理
3. 错误隔离：确保单个中间件失败不会导致整个系统崩溃
4. 监控记录：为关键中间件添加日志和监控
5. 测试覆盖：特别测试中间件组合时的交互效果

6. 常见问题与解决方案

在实际使用Middleware过程中，可能会遇到以下典型问题：

6.1 中间件执行顺序问题

问题现象：多个中间件相互干扰，产生预期外的行为

解决方案：

1. 明确中间件的依赖关系，调整添加顺序
2. 使用调试模式记录中间件执行流程
3. 为中间件添加清晰的命名和日志

6.2 性能瓶颈

问题现象：添加中间件后系统响应明显变慢

优化建议：

1. 评估每个中间件的性能开销
2. 对耗时操作进行异步处理
3. 实现中间件短路机制（满足条件时跳过后续中间件）
4. 考虑缓存中间件处理结果

6.3 上下文管理复杂

问题现象：中间件之间需要共享复杂状态，导致代码混乱

设计建议：

1. 使用统一的上下文对象管理共享状态
2. 定义清晰的接口规范中间件数据访问
3. 避免中间件之间的直接依赖
4. 考虑使用不可变数据结构

6.4 调试困难

问题现象：中间件链式调用导致问题难以追踪

调试技巧：

1. 为每个中间件添加唯一标识和详细日志
2. 实现中间件执行流程图可视化
3. 使用请求ID贯穿整个调用链
4. 开发中间件热插拔机制，便于隔离测试

7. Middleware高级应用模式

除了基本功能外，Middleware还可以实现一些高级应用模式：

7.1 A/B测试框架

通过中间件实现不同策略的A/B测试：

python复制class ABTestMiddleware(AgentMiddleware):
    def __init__(self, variants):
        self.variants = variants
        
    def before_model(self, request, handler):
        # 随机选择测试变体
        variant = random.choice(self.variants)
        # 修改请求参数实现不同策略
        request.model_params.update(variant.params)
        return handler(request)

7.2 渐进式功能发布

使用中间件控制功能开关：

python复制class FeatureToggleMiddleware(AgentMiddleware):
    def __init__(self, features):
        self.features = features
        
    def wrap_tool_call(self, tool_call, handler):
        # 检查工具是否已启用
        if tool_call.tool_name not in self.features.enabled_tools:
            raise Exception("功能暂未开放")
        return handler(tool_call)

7.3 多租户隔离

通过中间件实现租户特定的处理逻辑：

python复制class TenantAwareMiddleware(AgentMiddleware):
    def before_agent(self, request, handler):
        tenant_id = parse_tenant_id(request)
        # 加载租户特定配置
        request.context.tenant_config = load_tenant_config(tenant_id)
        return handler(request)

7.4 合规性检查

自动检查AI行为是否符合法规要求：

python复制class ComplianceMiddleware(AgentMiddleware):
    def after_model(self, response, handler):
        if not check_compliance(response.content):
            response.content = "抱歉，我无法提供该信息"
        return handler(response)

Middleware的设计灵活性使其能够适应各种复杂的业务需求和安全要求，是构建企业级AI应用不可或缺的组件。通过合理设计和组合中间件，开发者可以构建出既强大又安全的AI代理系统。