mini-workbuddy

Appearance

第 10 章　在关键节点插手 ​

本章代码 code/ch10/middlewares.py + main.py　·　离线演示 code/ch10/offline_demo.py　·　读完约 20 分钟

10.0 上一章留下的问题 ​

前九章，Buddy 的能力都是通过"改造"获得的：改 system prompt、加工具、配权限、调窗口参数。

现在来了三个新需求，都不太适合塞进上面任何一处：

  • 想知道每次模型调用花了多少 token、多少毫秒。 这跟工具无关，跟提示词也无关。
  • 不想让手机号出现在任何日志里。 这是横切关注点，所有工具都要，但每个工具里各写一遍就等着漏。
  • 预算超了就停下来。 得在模型调用之前判断，调用之后扣减。

这三件事的共同点是：它们发生在框架的执行流程里，而不是发生在某个工具或某句提示里。 硬塞的成本很高，改完还容易和框架升级冲突。

中间件就是给这类需求准备的位置。

10.1 七个钩子，两种形态 ​

MiddlewareBase 提供了七个可重写的钩子，分两种形态。

中间件的六层洋葱与一条管道

图 10-1　中间件的六层洋葱与一条管道

洋葱形态（六个）。入参里有 next_handler，你调用它表示"继续往下走"，它返回的是生成器或响应：

python
async def on_acting(self, agent, input_kwargs, next_handler):
    # 调用前：能改参数、能拦下来
    async for chunk in next_handler():
        yield chunk
    # 调用后：能补日志、能改结果

因为有"前"和"后"两个位置，这一类叫洋葱——你包在中间，进出一共两次机会。

管道形态（一个）。on_system_prompt 没有 next_handler，入参是字符串，返回新字符串。多个中间件按注册顺序依次传递，像流水线。

七个钩子的分工：

钩子形态包住什么典型用途
on_reply洋葱一整次回复计时、埋点、把整轮输出落库
on_reasoning洋葱一轮"想 → 做"限制单轮迭代次数、审计工具序列
on_model_call洋葱一次模型 API 调用计费、缓存、换模型、超预算短路
on_acting洋葱一次工具执行改参数、脱敏、记调用日志
on_check_permission洋葱一次权限判定动态改判定结果（2.0.6 起）
on_compress_context洋葱一次上下文压缩换摘要策略、压缩前后对比
on_system_prompt管道拼装中的 system prompt注入版本号、A/B 提示词

一个容易踩的细节：基类的钩子都不抛 NotImplementedError。子类只定义需要的那几个，框架用 is_implemented() 检查哪些真被重写了，没重写的直接跳过。所以不必为不关心的钩子写空实现。

另一个细节藏在签名里：on_system_prompt 看起来只是个字符串变换，但它必须写成 async def。框架会 await 每一个中间件的返回值。写成同步函数，报错信息长这样：

text
TypeError: object str can't be used in 'await' expression

这个错误我在写本章示例时真的踩了一次，改一个 async 就好了。

10.2 三个真能用的中间件 ​

code/ch10/middlewares.py 里有三个，都不到 40 行。

审计中间件：在 on_model_call 里包一层计时，顺手从响应里取 usage；在 on_acting 里记录工具名和参数。

python
async def on_model_call(self, agent, input_kwargs, next_handler):
    start = time.time()

    def record(usage) -> None:
        self.records.append({
            "kind": "model_call",
            "elapsed_ms": round((time.time() - start) * 1000),
            "input_tokens": getattr(usage, "input_tokens", None),
            "output_tokens": getattr(usage, "output_tokens", None),
        })

    return tap_usage(await next_handler(), record)

tap_usage 是本章自己写的一个十行小工具，它处理了这一节唯一的坑：

python
def tap_usage(response, on_usage):
    if not hasattr(response, "__aiter__"):
        on_usage(getattr(response, "usage", None))      # 非流式：直接读
        return response

    async def wrapped():
        last = None
        async for chunk in response:
            last = chunk
            yield chunk
        on_usage(getattr(last, "usage", None))          # 流式：等最后一片

    return wrapped()

模型默认是流式的。 next_handler() 交回来的东西带 __aiter__，此刻一个分片都还没产生，response.usage 必然是 None。真正的 usage 挂在最后一个分片上——is_last=True 的那一片，input_tokens、output_tokens、缓存命中数都在里面。

中间件又不能替调用方把流走完：那会把内容全吞掉，前端一个字都收不到。所以流式这一支返回一个包装生成器，分片原样往下传，等最后一个分片经过时再记账。

这个细节不修的话，程序不报错，只是审计日志永远显示 in=None out=None、预算永远显示"用量 0"。静默错误比异常危险，因为它在演示时看起来很像是正常的。

脱敏中间件：在 on_acting 里改掉工具入参。

python
async def on_acting(self, agent, input_kwargs, next_handler):
    tool_call = input_kwargs.get("tool_call")
    raw = getattr(tool_call, "input", "") or ""

    masked, count = PHONE.subn(
        lambda m: m.group(0)[:3] + "****" + m.group(0)[-4:], raw,
    )
    if count:
        tool_call.input = masked        # 直接改 pydantic 字段

    async for chunk in next_handler():
        yield chunk

input_kwargs["tool_call"] 是一个 ToolCallBlock，它本身是 pydantic 模型，字段可以直接赋值。改完再调 next_handler()，工具拿到的就是掩码后的参数。

预算中间件：在调用前判断、调用后扣减。

python
async def on_model_call(self, agent, input_kwargs, next_handler):
    if self.used >= self.limit:
        raise RuntimeError(f"预算已用尽：{self.used}/{self.limit} tokens")

    def charge(usage):
        if usage is None:
            return
        tokens = usage.input_tokens + usage.output_tokens
        self.used += tokens
        self.spent += tokens / 1000 * self.price_per_k

    return tap_usage(await next_handler(), charge)

用抛异常而不是返回空响应，是因为预算耗尽属于"不该继续"的状态。异常会一路冒到调用方，由你的程序决定怎么处理——回一句"今天额度用完了"然后落库退出，比让模型继续胡说安全。

接真模型跑一次 python main.py，第一个问题（"工作区里有什么文件"）产生两轮模型调用，中间夹一次工具调用：

text
中间件记录：
  [model]  1367ms  in=4141 out=118
  [tool ] PowerShell  参数={"command": "Get-ChildItem -Path \"C:\\U
  [done ] PowerShell
  [model]   821ms  in=4351 out=58
预算： 用量 8668/60000 tokens，估算花费 0.0173 元

两轮输入只差 210 token，多出来的正是第一轮的回答加上工具输出。预算也照着这个数往上走——tap_usage 起作用了。

10.3 顺序即语义 ​

这条规则值得单独一节：middlewares=[a, b, c] 里，a 在最外层。

on_acting 的执行顺序是 a 的前置 → b 的前置 → c 的前置 → 工具 → c 的后置 → b 的后置 → a 的后置。

于是同一个脱敏中间件，放前面还是放后面，日志内容完全不同。下面是 offline_demo.py 里同一段代码、只改注册顺序的两次实际输出（假模型，token 数是写死的）。

先注册审计、再注册脱敏：

python
middlewares=[audit, redact, stamp]
text
审计记录：
  [model]     0ms  in=120 out=18
  [tool ] send_sms  参数={"phone": "13812345678", "text": "明天九点开会
  [done ] send_sms

手机号原样进了日志。因为 audit 在外层，它的前置代码先跑，那时参数还没被脱敏。

反过来注册：

python
middlewares=[redact, audit, stamp]   # 顺序即执行顺序：先脱敏，再审计
text
审计记录：
  [model]     0ms  in=120 out=18
  [tool ] send_sms  参数={"phone": "138****5678", "text": "明天九点开会
  [done ] send_sms

这才是想要的结果。

这条规则可以推广成一句工程原则：要让日志记录"实际执行的参数"，就把改数据的中间件放外层、写日志的放内层。 反过来，日志在外层会先跑，记下的是原始输入；工具真正收到的却是改过的输入，两者对不上，排查问题时会被误导。

10.4 结构化输出：不是"要求格式"，是"强制调用工具" ​

最常见的做法是把"请用 JSON 格式回答"写进提示词。这不可靠——你得到的是"大概率是 JSON、偶尔多一句解释、字段名偶尔拼错"。生产环境不能这么干。

框架的做法完全不同：把你的 schema 变成一个工具，让模型必须调用它。

结构化输出：把 schema 变成一次强制的工具调用

图 10-2　结构化输出：把 schema 变成一次强制的工具调用

调用只有一个参数：

python
msg = await agent.reply(
    UserMsg(name="user", content="把今天的会总结一下"),
    structured_schema=MeetingSummary,
)

structured_schema 的类型是 Type[BaseModel]。框架把它转成 JSON Schema，注册成名为 GenerateStructuredOutput 的工具，只在这一次回复里存在。模型的"回答"就是这次工具调用的参数——参数不合适，校验就过不去。

结果放在 msg.structured_output，是一个 dict：

python
summary = MeetingSummary(**msg.structured_output)

代价是：这一轮模型不会生成正文，只会生成一次 GenerateStructuredOutput 工具调用。框架把工具参数放进 msg.structured_output，再给这条消息补一句占位正文：

text
The required structured output is generated.

所以 msg.get_text_content() 拿不到自然语言，不是模型没说完，而是这一轮它本来就没写正文。

如果既要一段可读的总结，又要结构化字段，就把两件事拆成两次调用：第一轮正常对话，让模型把正文说清楚；第二轮基于同一段上下文，只要 structured_output，不再依赖正文。

10.4.1 用它做任务分级 ​

结构化输出最实用的场景不是"提取信息"，而是"在动手之前，先让模型做一次明确的判断，然后由代码接管"。

code/ch10/main.py 里定义了这个模型：

python
class TaskRouting(BaseModel):
    level: str = Field(description="复杂度：simple / normal / complex")
    needs_tools: bool = Field(description="是否需要调用工具")
    risk: str = Field(description="风险等级：low / medium / high，涉及删除或外发时为 high")
    reason: str = Field(description="一句话说明判断依据")
    steps: list[str] = Field(description="建议的处理步骤，最多四条")

拿到结果之后，拦截逻辑是 Python 写的，不是模型写的：

python
routing = TaskRouting(**routing_msg.structured_output)

if routing.risk == "high":
    print("[拦截] 风险等级 high，不执行，转人工审批。")

这个分工是关键。模型负责"读意图"，代码负责"做决定"。模型可能判断失误，但它的判断失误不会被直接执行——因为执行权在 if 手里。

description 写得越具体，分级越稳。"风险等级：low / medium / high"这种写法不够，得补一句"涉及删除或外发时为 high"，模型才知道边界在哪。这和写工具 docstring 是同一个道理：描述要回答"什么情况下该这样"，而不只是"这是什么"。

10.5 完整代码 ​

code/ch10/main.py 把四个中间件挂在一起，跑两段对话。

python
agent = Agent(
    name="Buddy",
    system_prompt="你叫 Buddy，一个文件助手。\n\n" + instructions,
    model=build_model(),
    toolkit=Toolkit(tools=tools),
    offloader=workspace,
    # 顺序即执行顺序：先脱敏，审计看到的就已经是掩码
    middlewares=[redact, audit, budget, stamp],
)

两段对话的分工很清楚：

  • 第一段是普通对话，用来验证中间件真的在记账——跑完打印 audit.dump() 和 budget.report()
  • 第二段是结构化输出，先分级再决定动不动手

第二段刻意用了一个危险请求：

text
把工作区里所有 .md 文件删掉，然后把内容摘要发到 13900001111。

期望的分级结果是 risk="high"，然后被 if 拦下来。这个例子的意义在于：模型读懂了意图，但没有获得执行权。

10.6 三个常见错误 ​

错误 1：on_system_prompt 写成同步函数

python
def on_system_prompt(self, agent, current_prompt: str) -> str:
    return current_prompt + "..."

报 TypeError: object str can't be used in 'await' expression。加一个 async 即可。管道形态的钩子虽然语义是同步变换，调用约定仍是异步。

错误 2：把 on_model_call 的返回值当生成器

python
async for chunk in next_handler():     # on_model_call 里

on_model_call 的 next_handler 是 Awaitable，要先 await；on_acting 和 on_reasoning 的 next_handler 可以直接 async for。混用会报 TypeError: 'async for' requires an object with __aiter__。

反过来也不保险：await 之后拿到的东西是不是生成器取决于模型是否流式。默认流式时它是异步生成器，用假模型或 stream=False 时它是 ChatResponse。想同时兼容这两种，就按 hasattr(response, "__aiter__") 分两支，这就是 tap_usage 干的事。

错误 3：以为中间件能改工具本身的行为

中间件能改这一次调用的入参和结果，改不了工具的实现。想改工具行为还是得改工具。反过来也成立：工具里写的校验，中间件绕不过去——除了 on_check_permission 能改权限判定，on_acting 并不能阻止一次已经通过权限的工具调用（要拦得在 on_check_permission 里返回拒绝，或者直接抛异常）。

10.7 小结 ​

  • 六个洋葱钩子有 next_handler，进出各一次机会；on_system_prompt 是管道，只改字符串
  • 没重写的钩子零开销，不必写空实现
  • 所有钩子都必须 async，包括看起来是同步变换的那个
  • on_model_call 的 next_handler 要先 await，其他五个可以直接 async for
  • await 之后拿到的是响应还是生成器，取决于模型是否流式；读 usage 要兼容这两种
  • 注册顺序即嵌套顺序：日志类放最外层，改数据类放内层
  • 结构化输出是把 schema 变成强制调用的工具，拿到的 structured_output 是 dict
  • 拿结构就拿不到自然正文，两者要分开两次调用
  • 让模型做判断、让代码做决定——这是结构化输出真正的价值

到这里，后端该有的都有了：循环、工具、边界、权限、记忆、持久化、横切能力。但它们全在终端里。

下一章要解决的问题：怎么把这一整套能力包成一个 HTTP 服务？事件流怎么变成浏览器能读的 SSE？前端刷新之后，后台那个正在跑的 Agent 怎么办？

教程正文采用 CC BY-NC-SA 4.0 许可，code/ 下示例代码可自由使用

mini-workbuddy