Agent中间件核心钩子：before_model与after_model的工程实践

1. 从一次生产事故说起：模型调用前后为什么要保留两个事件钩子

1.1 一次毫不起眼的偶发超时事故

我维护的一个智能体服务，在某次版本上线后开始出现一种很折磨人的现象：白天高峰期，大概每三四百次模型调用里会有一次超时，报错提示是"上游模型调用超时"。最初所有人的第一反应都是模型服务侧出问题了，但排查了一整天，模型服务那边的监控曲线平平整整，完全不像有异常的样子。

后来我把单次请求的完整链路日志捞出来，逐帧回放，才发现问题根本不在模型调用本身，而是卡在了一个我之前几乎没正眼看过的环节——某个 before_model 拦截器里有一段同步写日志的代码，高峰期排队加重试，硬生生把一个本应 800 毫秒的模型请求拖到了 15 秒以上。那一刻我才彻底明白：在一个完整的智能体链路里，模型调用前与调用后的那两个钩子函数，从来不是可有可无的"旁路小功能"，而是决定整个系统稳定性、可观测性和可控性的关键关卡。

1.2 安检闸门和后处理流水线

如果你熟悉生产流水线的概念，就很容易理解 before_model 和 after_model 的定位。一条流水线上，产品进入核心加工设备之前，会有一道安检闸门，检查产品是否符合加工条件，不符合的要么剔除要么改写；产品加工完成后，又会有一个质检工位，检查成品是否合格、是否要做二次处理、记录产量。把这条流水线映射到 Agent 框架里，核心加工设备就是模型调用，安检闸门就是 before_model 事件，质检工位就是 after_model 事件。

这两个事件的存在有非常现实的理由。我见过不少初期不太成熟的 Agent 项目，业务逻辑直接往框架内部代码里扎，想拦截请求就改调用点，想处理响应就改返回解析处，结果框架一升级就废掉一大片。而通过中间件事件，框架把"模型调用前后"这两个关键节点开放成稳定的钩子，让开发者在不侵入内部调用流程的前提下，对请求和响应进行任何定制化处理。这带来的好处非常实在：你可以把日志、鉴权、限流、缓存、格式修正、成本统计这些横切关注点，全部从业务代码里剥离出来，统一收敛到中间件里管理。

很多刚开始接触 Agent 中间件的人会有一个误解，觉得 before_model 和 after_model 不过是两个"通知类"的回调，模型调用之前告诉我一声，调用之后告诉我一声，仅此而已。但实际上，它们远比通知更强大：你可以直接修改即将发送给模型的消息内容，可以决定这次调用是否要下发，甚至可以在响应返回后替换掉整个响应内容。这种能力如果掌握不好，也会带来巨大的隐患。后面我会用一整节的篇幅专门讲踩坑，先说清楚能力，再讲清楚边界。

2. before_model：请求进入模型前的那道闸门

2.1 触发时机与可操作对象

before_model 的触发时机非常明确：在框架收集完当前 Agent 的多轮上下文、组装好完整请求体、准备调用模型 API 的那一刻触发。也就是说，你在这个钩子里看到的数据，和真正发给模型的数据只隔着一层内部结构体的距离。

触发时，钩子函数通常是下面这样的签名（不同框架写法略有差异，但核心结构一致）：

python复制def before_model(self, agent, request, context):
    # agent：当前正在运行的Agent实例
    # request：即将发往模型的请求对象
    # context：跨事件共享的上下文容器
    return request

关键点在于返回值。如果你直接返回传入的 request，表示不做任何修改；如果你返回一个新的请求对象，或者原地修改 request 的字段再返回，后续流程就会按你的新请求去执行。这给了你在最后一刻"扭转乾坤"的机会。

request 对象上通常能看到这些字段：

2.2 高频用途：改写、注入、拦截、熔断

我实际项目中用得最多的四类 before_model 逻辑，按出现频率排是这样的：

第一类是上下文注入。智能体跑在业务系统里，很多隐式状态并不会被塞进对话消息里，比如当前用户的权限等级、语言偏好、会员有效期。直接在业务代码里每次组装消息太容易遗漏，放在 before_model 里统一注入，可以保证每一个模型请求体里都带上通用信息。

第二类是规则拦截。有些场景根本不需要调用模型。比如用户在短时间内重复提交了相同指令，或者命中了一些明令禁止的敏感词，直接在 before_model 里终止调用，返回一个预先写好的答复，比把请求发给模型再等一个不确定的答复要快得多，也省钱得多。

第三类是请求改写与兜底。比如多租户场景下，需要根据租户配置动态调整模型参数；或者主模型当前处于灰度阶段，灰度比例外的请求需要落到备用模型上。这类逻辑放在调用前处理，既能保证策略统一，又不会污染业务代码。

第四类是限流与熔断。如果上游模型服务当前处于过载状态，没必要再往里面塞请求。在 before_model 里读取熔断状态，直接返回缓存结果或降级答复，可以避免整个系统被拖垮。

2.3 一个可复用的请求拦截与改写示例

假设我们需要实现三个能力：给所有请求自动补充系统提示词、对特定请求直接拦截不调模型、按用户等级切换模型。代码可以这样写：

python复制import datetime

class BusinessBeforeModelMiddleware(AgentMiddleware):
    def __init__(self, user_level_retriever):
        self.user_level_retriever = user_level_retriever
        self.deduplicator = set()

    def before_model(self, agent, request, context):
        user_id = context.get("user_id")
        if not user_id:
            return request

        # 1. 拦截重复请求：同一用户10秒内的相同问题直接返回缓存答复
        cache_key = f"{user_id}:{request.messages[-1].content}"
        if cache_key in self.deduplicator:
            context.set("skip_model_call", True)
            return None  # 返回None代表终止调用
        
        # 2. 按用户等级切换模型
        user_level = self.user_level_retriever.get_level(user_id)
        if user_level == "free":
            request.model = "tier-small"
            request.max_tokens = 2048
        elif user_level == "vip":
            request.model = "tier-large"
            request.max_tokens = 8192

        # 3. 注入统一系统提示词
        language = context.get("user_lang", "zh-CN")
        request.messages.insert(0, {
            "role": "system",
            "content": (
                f"当前用户语言偏好：{language}。"
                f"当前时间：{datetime.datetime.now().isoformat()}"
            )
        })
        return request

这里的细节有两个值得留意。第一，返回 None 的语义是"终止本次模型调用"，框架会在 before_model 返回 None 后自动走降级流程，可能是取缓存、可能是返回预设话术、也可能是直接跳过，具体取决于框架约定。第二，通过 context 共享状态，可以让同一个中间件在不同的钩子之间传递信息，后面我会重点展开。

3. after_model：响应的后处理环节，藏着多少价值

3.1 触发时机与 after 阶段的数据视角

after_model 的触发时机是在模型 API 返回响应之后、框架把响应交给上层业务之前。如果 before_model 是"发送前最后一道关卡"，那 after_model 就是"交出去前最后一道关卡"。

这里要强调一个视角差异：after_model 里拿到的 request 已经被 before_model 修改过了，是真正发出去的那一份；同时你会拿到一个全新的 response 对象，里面装着模型返回的内容。签名大致是这样：

python复制def after_model(self, agent, request, response, context):
    # request：实际发送给模型的请求（已被before_model改写）
    # response：模型返回的响应对象
    return response

response 对象上通常包含的内容有：最终生成的文本内容、模型返回的原始输出、token 用量统计（输入 token 数、输出 token 数）、模型名称、可能的终止原因（比如因为 token 上限被截断）。这些数据在业务层往往很容易被忽略，但在 after_model 里它们就像流水线上的产品合格证一样齐全。

3.2 后处理能做的六类经典操作

我在 after_model 里最常用的逻辑，粗算下来有六类：

第一是结构化修正。很多模型返回的内容会有多余的空白、重复的标点、偶尔的格式跑偏，直接在 after_model 里做一次正则清洗，比让业务侧每次解析时处理要省心得多。

第二是缓存写入。如果这是一次可以复用的通用问题，可以把模型返回写入缓存，下次遇到相同请求就能在 before_model 阶段直接命中拦截。

第三是成本与用量统计。把每次调用的 token 用量累计到某个指标服务中，用于账单分析和配额控制。这个数据在业务层可能只关心最终文本，但平台层必须关心。

第四是质量拦截与重试。如果模型返回的是空字符串、或明显格式错误、或触发了内容安全策略，after_model 可以自动触发一次重试，而不是把坏结果直接交给用户。

第五是日志沉淀。把完整的请求与响应摘要写入日志链路，方便事后问题追踪。比如我在最开始提到的那次事故，如果日志模板从一开始就设计合理，问题排查时间至少能缩短一半。

第六是响应模板化替换。某些特殊业务场景，比如模型返回的是内部代号，需要在返回给用户前替换成友好的展示文案。

3.3 一个响应清理与重试示例

下面是一个压制格式噪声并进行质量检查的示例，逻辑非常简单但非常实用：

python复制import re

class ResponseCleaningMiddleware(AgentMiddleware):
    def after_model(self, agent, request, response, context):
        # 1. 提取模型文本
        raw_text = response.content or ""

        # 2. 清理常见格式噪声
        cleaned = re.sub(r"\n{3,}", "\n\n", raw_text)
        cleaned = re.sub(r"[ \t]{2,}", " ", cleaned)

        # 3. 空内容检测
        if not cleaned.strip():
            context.set("needs_retry", True)
            return None

        # 4. 截断场景：如果模型因为长度限制中断，在响应上打个标记
        if getattr(response, "finish_reason", "") == "length":
            cleaned += "\n\n[内容已达长度上限，如需继续请告诉我]"
            
        response.content = cleaned
        return response

同样地，返回 None 表示"这个响应不合格，我不认"，框架会按约定进入重试或异常处理流程。但这里我提前给一个忠告：在 after_model 里返回 None 实现重试，一定要加重的次数限制，否则模型连续出现格式错误时，你的系统会在一瞬间被重试请求打爆。这个点我在第 6 章会再详细说。

4. 事件上下文与参数传递：中间件到底在哪一层工作

4.1 事件对象的结构拆解

很多刚上手的人会问：我在中间件里改了一个对象，后面的业务代码能感知到吗？答案取决于你改的是事件对象本身，还是改的是跨钩子共享的上下文。这两个东西必须分清楚。

事件参数里的 agent 是当前运行实例，它承载了这个智能体所有内部状态，比如当前 Agent 的记忆、工具列表、执行计划等。request 和 response 是单次模型调用的输入输出，生命周期只在这一次调用范围内。真正能跨阶段传递信息的，是那个 context 对象。

4.2 共享上下文：跨钩子状态的可靠通道

context 的本质是一个作用域贯穿整个 Agent 运行周期的共享容器。它通常以字典或键值对接口存在，比如 context.get(key)、context.set(key, value)。它的价值在于：一个中间件的 before_model 里写入的数据，可以被同一个中间件的 after_model 读取，也可以被后续其他中间件读取，甚至可以被业务层在整体调用结束后读取。

我举一个真实场景。某个内部运营助手，每次用户提问都要统计查询耗时和 token 总消耗。

python复制import time

class UsageTrackingMiddleware(AgentMiddleware):
    def before_model(self, agent, request, context):
        context.set("call_start_time", time.time())
        context.set("call_model", request.model)
        context.set("call_agent", agent.agent_id)
        return request

    def after_model(self, agent, request, response, context):
        elapsed = time.time() - context.get("call_start_time", 0)
        usage = {
            "agent_id": context.get("call_agent"),
            "model": context.get("call_model"),
            "cost_ms": round(elapsed * 1000, 2),
            "prompt_tokens": response.usage.get("prompt_tokens", 0) if response.usage else 0,
            "completion_tokens": response.usage.get("completion_tokens", 0) if response.usage else 0,
        }
        self.metrics_reporter.report(usage)
        return response

这里我们做的事情其实很简单：before_model 里记录起始时间和关键参数，after_model 里读取这些值并计算耗时。如果没有共享上下文，这种跨钩子的状态传递就得靠全局变量或额外存储，既脏又不安全，还会在并发请求下互相污染。

4.3 并发隔离与上下文污染

说到上下文，必须提醒一个容易踩的坑：并发场景下，同一个 context 实例必须是"按请求隔离"的。也就是说，每次新的用户请求或每次 Agent 执行循环，框架都应该创建一个新的 context，而不是所有请求共用一份。我之前见过一个项目，为了省事把 context 定义为类变量，结果两个用户同时提问时，A 用户的上下文被 B 用户的并发请求覆盖，最终出现张冠李戴的答非所问。如果你的中间件代码依赖 context 传递状态，第一步就要确认框架的 context 生命周期到底绑定在哪一级。这决定了你的中间件是线程安全的，还是隐患缠身的。

5. 多中间件串行编排时的顺序规则

5.1 注册顺序决定执行路径

几乎所有的中间件机制都会遵循一个共同规则：注册顺序就是执行顺序。系统会按照中间件注册列表的顺序，依次调用每个中间件的 before_model；调用完第一个，把返回结果作为第二个的输入，依次串联。

需要注意的是，after_model 的顺序有两种常见约定：一种是"与注册顺序相同"，另一种是"与注册顺序相反"。反序更贴近“后进先出”的语义，形似入栈出栈：最早注册的中间件最后处理响应。无论你的框架采用哪种，原则都一样——你必须先明确这个顺序，再设计中间件的依赖关系。

我这里给一张简单的对比表，帮助你理解顺序差异带来的实际影响：

5.2 多个中间件之间的协作与冲突

顺序之所以重要，根本原因是中间件之间会互相影响。举个例子：中间件 A 在 before_model 里把用户消息里的敏感信息脱敏了，中间件 B 再对消息做关键字匹配，它看到的就已经是脱敏后的内容。如果 B 本来希望基于原始内容做风控，就会产生错误判断。反过来，如果 A 在 after_model 里把响应里的占位符替换成真实数据，而 B 在 after_model 的注册顺序在 A 之前，B 看到的响应里就还残留着占位符。

因此，在大量中间件并存的情况下，我的实践经验是三条原则：第一，同类操作尽量合并到同一个中间件里，避免多个中间件做类似的事却互相覆盖；第二，凡是对请求或响应有"读后写"依赖的中间件，要明确依赖关系，绝不能靠运气排序；第三，不要在中间件里依赖别的中间件的某个副作用，除非你非常确信框架保证执行顺序。

5.3 覆盖语义：返回新对象与原地改字段的差异

还有一个经常被忽略的细节：中间件的 before_model 既可以原地修改 request 对象再返回，也可以直接构造一个全新的 request 对象返回。在大部分框架中，原地修改的语义是"后续中间件仍然看到这个对象"；但如果返回的是一个全新对象，后续中间件看到的则是新对象。这听起来简单，实际写错的人不少——有人在一个中间件里原地改了字段，又在另一个中间件里返回了一个旧引用，导致前面的修改全部丢失。

我建议的写法是：除非你有明确理由要返回完整新对象，否则一律原地修改字段再返回原对象。这个约定虽然不够"函数式"，但在工程上最不容易出错。

6. 踩坑实录：流式响应、异常吞噬、超时阻塞

6.1 流式响应在 after_model 里拿不到完整内容

这是我在实际项目里踩过的最烦人的坑之一。当你开启流式模式时，模型返回不是一次性到达的，而是分块推送。如果在 after_model 里试图读取 response.content 拿完整文本，在某些框架实现里你会只拿到最后一片内容，甚至拿到的只是一个迭代器对象。

如果业务必须拿到完整内容再做处理，有两个解决方案：要么在调用模型之前通过 before_model 把 request.stream 强制设为 False，拿到完整响应后自己再模拟流式输出；要么在 after_model 里把流式数据完整聚合成一个缓冲区，再统一处理。后者的实现方式大致是这样：

python复制class StreamAggregationMiddleware(AgentMiddleware):
    def after_model(self, agent, request, response, context):
        # 假设response.content是流式分片迭代器
        if hasattr(response.content, "__iter__") and not isinstance(response.content, str):
            chunks = []
            for chunk in response.content:
                if hasattr(chunk, "content") and chunk.content:
                    chunks.append(chunk.content)
            full_text = "".join(chunks)
            response.content = full_text
        return response

但这里要警惕：如果同时注册了多个中间件，聚合中间件必须排在最前面执行，否则其他中间件看到的也是分片内容。

6.2 after_model 里抛异常会引发重试风暴

前面提过在 after_model 里返回 None 触发重试的机制。如果框架约定"返回 None 表示重试"，并且你没有在抛出处做重试次数限制，当模型连续多次返回格式错误时，重试会像滚雪球一样越滚越大。更危险的是 after_model 内部抛出的异常：此时模型调用已经成功，钱已经花出去了，异常如果被框架吞掉，你会损失一次有效响应且毫无感知；异常如果向上抛出，业务层就会看到一个非常难理解的"模型调用失败"。

我的建议是：after_model 里的逻辑要遵循"灰度容忍"原则——轻量校验和修正操作可以直接做；任何可能失败的重操作（写数据库、调用外部 API、上报指标）都要用 try-except 包起来，并给足重试上限；宁可丢弃局部逻辑，也不能污染正常响应返回链路。

6.3 before_model 里做重活导致模型调用超时

我还见过另一种情况：有人在 before_model 里同步调用了外部鉴权服务，外部服务一次耗时 2 秒，直接把整个模型调用的预算吃掉了大半。本质上 before_model 的代码运行时间紧跟着模型调用的倒计时，它做得越重，留给模型生成的时间就越少。

如果你必须在 before_model 阶段做耗时操作，至少要考虑这三个优化方案：第一，能并行的操作就并行，比如多个上下文注入内容可以并发拉取再合并；第二，能缓存的就缓存，鉴权结果、用户画像这类变化不频繁的信息，本地缓存十分钟对业务体验几乎没有影响；第三，把非核心操作挪到 after_model 或其他更合适的阶段，不要全挤在 before_model 里。

6.4 最后分享三个调试技巧

调试中间件事件，我常用的方法有三个，都不神秘但非常有效。

第一个是链路日志定位。在中间件里给每次模型调用生成一个全局唯一的 call_id，写入 context，在 before_model 和 after_model 里分别打印关键字段摘要。这样出问题时，只需要按 call_id 搜索日志，就能还原模型请求一路上被谁改过、改成了什么样、最终返回了什么。

第二个是中间件卸载对照。怀疑某个中间件有问题时，先把它从注册列表里摘掉，跑一组基准请求对比行为和性能。如果质量下降，说明这个中间件在起作用；如果行为没区别，说明它早就在帮倒忙，可以名正言顺地删掉。

第三个是务必给对外协处理器加超时。所有在 before_model 和 after_model 中发起的网络调用、数据库查询，都要设置明确的超时时间，超时后按降级逻辑走，而不是无限阻塞等待。这个习惯救了我很多次，也建议你尽早养成。

这两个钩子看着不起眼，但在我的项目里，它们承担了重试、限流、日志、统计、脱敏、格式修正等十几个职责。把它们的语义彻底吃透，Agent 系统的稳定性和可观测性会肉眼可见地上一个台阶。