第 15 章：元类与 Hook——方法调用的拦截

发表于 分类于 阅读次数： 本文字数： 6.9k 阅读时长 ≈ 6 分钟
深入解析 Hook 系统的元类实现：_AgentMeta 在类定义时自动包装 reply 等方法，_wrap_with_hooks 提供防重入保护与链式执行，参数归一化让 Hook 能安全修改调用参数，_ReActAgentMeta 进一步扩展 reasoning 与 acting 钩子点。
个人公众号

难度：进阶

你想在 Agent 每次 reply 前后自动执行一些逻辑（比如日志记录、参数校验），但不想修改 Agent 的源码。Hook 系统就是为这个设计的——它是怎么实现的？

上一章：第 14 章 继承体系

知识补全：元类（Metaclass）

在 Python 中，类也是对象。元类是”创建类的类”。

1
2
3
4
5
class MyClass:        # 普通类，创建实例
    pass

obj = MyClass()       # obj 是 MyClass 的实例
# MyClass 是 type 的实例！

当你写 class MyClass: 时，Python 实际上调用 type.__new__() 来创建这个类。元类允许你拦截类的创建过程，在类被创建之前修改它的属性和方法。

1
2
3
4
5
6
7
8
9
10
11
class MyMeta(type):
    def __new__(mcs, name, bases, attrs):
        # 在类被创建前，修改 attrs（方法字典）
        attrs['extra_method'] = lambda self: "added by metaclass"
        return super().__new__(mcs, name, bases, attrs)

class MyClass(metaclass=MyMeta):
    pass

obj = MyClass()
obj.extra_method()  # "added by metaclass"

AgentScope 用这个机制在类定义时就自动给 reply、observe、print 方法包上 Hook 逻辑。

_AgentMeta：元类的实现

打开 src/agentscope/agent/_agent_meta.py，找到第 159 行：

1
2
3
4
5
6
7
# _agent_meta.py:159
class _AgentMeta(type):
    def __new__(mcs, name, bases, attrs):
        for func_name in ["reply", "print", "observe"]:
            if func_name in attrs:
                attrs[func_name] = _wrap_with_hooks(attrs[func_name])
        return super().__new__(mcs, name, bases, attrs)

只有 8 行代码。它做的事情：

  1. 遍历 ["reply", "print", "observe"] 三个方法名
  2. 如果类定义了这个方法（func_name in attrs），用 _wrap_with_hooks 包装它
  3. 用包装后的函数替换原来的方法

关键点：这个包装发生在类被创建时，不是在实例方法被调用时。所有 AgentBase 的子类自动获得 Hook 能力。

_wrap_with_hooks：Hook 的执行逻辑

_wrap_with_hooks（第 55 行）是一个装饰器工厂。它返回的 async_wrapper 做这些事：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
async def async_wrapper(self, *args, **kwargs):
    # 1. 防重入保护（避免继承链中重复执行 Hook）
    if getattr(self, hook_guard_attr, False):
        return await original_func(self, *args, **kwargs)

    # 2. 归一化参数（把位置参数转成关键字参数）
    normalized_kwargs = _normalize_to_kwargs(original_func, self, *args, **kwargs)

    # 3. 执行 pre-hooks（先实例级，后类级别）
    pre_hooks = list(self._instance_pre_reply_hooks.values()) + \
                list(self.__class__._class_pre_reply_hooks.values())
    for pre_hook in pre_hooks:
        modified = await pre_hook(self, deepcopy(normalized_kwargs))
        if modified is not None:
            normalized_kwargs = modified

    # 4. 执行原始函数
    setattr(self, hook_guard_attr, True)
    try:
        output = await original_func(self, **normalized_kwargs)
    finally:
        setattr(self, hook_guard_attr, False)

    # 5. 执行 post-hooks
    post_hooks = list(self._instance_post_reply_hooks.values()) + \
                 list(self.__class__._class_post_reply_hooks.values())
    for post_hook in post_hooks:
        modified = await post_hook(self, deepcopy(normalized_kwargs), deepcopy(output))
        if modified is not None:
            output = modified

    return output

防重入保护

hook_guard_attr（如 _hook_running_reply）是一个标志位。当继承链中多个类都定义了 reply 时，每个都被 _AgentMeta 包装了，但 Hook 只执行一次——最外层设置标志，内层检测到标志就跳过。

执行顺序

1
pre_hook(实例级) → pre_hook(类级别) → 原始函数 → post_hook(实例级) → post_hook(类级别)

_ReActAgentMeta：扩展 Hook 类型

ReActAgentBase 使用了 _ReActAgentMeta 元类（在 _react_agent_base.py 中），它扩展了 Hook 类型：

1
2
3
# _react_agent_base.py:28-31
"pre_reasoning", "post_reasoning",
"pre_acting",   "post_acting",

所以 ReAct Agent 支持 6 个 Hook 点：reply、observe、print、reasoning、acting。

1
2
3
4
flowchart TD
    A["AgentBase (metaclass=_AgentMeta)"] --> B["reply / observe / print"]
    C["ReActAgentBase (metaclass=_ReActAgentMeta)"] --> D["reply / observe / print / reasoning / acting"]
    E["ReActAgent"] --> F["继承 ReActAgentBase 的所有 Hook"]

Hook 的注册

Hook 分两种级别：

类级别：所有实例共享。在类上直接注册：

1
2
3
4
@AgentBase.register_class_hook("post_reply", "log")
def log_hook(self, kwargs, output):
    print(f"Agent {self.name} replied")
    return output

实例级别：只影响单个实例。在实例上注册：

1
agent.register_instance_hook("pre_reply", "validate", my_validator)

AgentScope 官方文档的 Building Blocks > Hooking Functions 页面展示了 Hook 的注册方式（register_instance_hook / register_class_hook）和支持的 Hook 类型表格：reply/observe/print 各有 pre 和 post 两个时机，ReAct Agent 额外支持 reasoning/acting 的 pre/post hooks。本章解释了这些 Hook 是如何通过元类在类定义时自动注入的。

AgentScope 1.0 论文对 Hook 机制的设计说明是：

“we ground agent behaviors in the ReAct paradigm and offer advanced agent-level infrastructure based on a systematic asynchronous design”

— AgentScope 1.0: A Comprehensive Framework for Building Agentic Applications, arXiv:2508.16279, Section 2.2

Hook 系统是这个”高级 Agent 基础设施”的一部分——让开发者在不修改源码的情况下，在 Agent 的关键方法执行前后插入自定义逻辑。

_normalize_to_kwargs：参数归一化

在 Hook 的执行链中，有一个容易被忽略但很重要的步骤：_normalize_to_kwargs（第 35 行）。

它的作用是把 reply(self, msg, structured_model=None) 的位置参数转成关键字参数字典：

1
2
3
4
5
6
7
8
# 调用方式 1：位置参数
await agent.reply(msg)

# 调用方式 2：关键字参数
await agent.reply(msg=msg)

# _normalize_to_kwargs 把两者都转成：
{"msg": msg}

为什么要这样做？因为 Hook 函数需要修改参数——如果用位置参数，Hook 无法知道第几个参数是什么。统一转成 kwargs 字典后，Hook 可以自由地增删改参数。

Hook 类型的完整列表

元类Hook 类型对应方法
_AgentMetapre_reply / post_replyreply()
_AgentMetapre_print / post_printprint()
_AgentMetapre_observe / post_observeobserve()
_ReActAgentMeta以上全部 + pre_reasoning / post_reasoning_reasoning()
_ReActAgentMeta以上全部 + pre_acting / post_acting_acting()

每个 Hook 类型有实例级和类级别两种注册方式。实例级只影响单个 Agent，类级别影响所有该类的实例。

两个元类的继承关系

1
2
3
4
5
6
7
8
9
10
11
12
# _agent_meta.py:159
class _AgentMeta(type): ...

# _react_agent_base.py:12
class _ReActAgentMeta(_AgentMeta):  # 继承 _AgentMeta
    def __new__(mcs, name, bases, attrs):
        # 扩展 Hook 列表：加入 reasoning 和 acting
        attrs = super().__new__(mcs, name, bases, attrs)  # 先执行父类的包装
        for func_name in ["_reasoning", "_acting"]:
            if func_name in attrs:
                attrs[func_name] = _wrap_with_hooks(attrs[func_name])
        return super(type, mcs).__new__(mcs, name, bases, attrs)

_ReActAgentMeta 继承 _AgentMeta，先让父类包装 reply/observe/print，然后自己再包装 _reasoning/_acting。这样 ReAct Agent 自动获得全部 6 个 Hook 点。

调试实践：观察 Hook 的执行顺序

目标：亲眼看到 pre-hook → 原始函数 → post-hook 的执行链。

步骤：

  1. 在 src/agentscope/agent/_agent_meta.py 的 async_wrapper 函数中（约第 70 行），加 print：
1
2
3
4
5
6
7
8
async def async_wrapper(self, *args, **kwargs):
    if getattr(self, hook_guard_attr, False):
        print(f"[DEBUG] {self.name}: 跳过 Hook（防重入）")
        return await original_func(self, *args, **kwargs)

    print(f"[DEBUG] {self.name}: 开始执行 Hook 链 for {func_name}")
    normalized_kwargs = _normalize_to_kwargs(...)
    ...
  1. 创建测试脚本：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import asyncio
from agentscope.agent import UserAgent
from agentscope.message import Msg

async def main():
    agent = UserAgent(name="test")
    agent.register_instance_hook(
        "post_reply",
        "log",
        lambda self, kwargs, output: print(f"  [HOOK] {self.name} 回复完成")
    )
    # UserAgent.reply 直接返回用户输入，不需要 API key

asyncio.run(main())
  1. 观察 Hook 链的执行顺序

改完后恢复：

1
git checkout src/agentscope/agent/

试一试：注册一个日志 Hook

目标：用 Hook 记录 Agent 每次调用 reply 的参数。

步骤：

  1. 打开任意使用 ReActAgent 的脚本，在创建 Agent 之后加：
1
2
3
4
5
6
7
import time

def log_reply(self, kwargs, output):
    print(f"[HOOK] {self.name} replied at {time.strftime('%H:%M:%S')}")
    return output

agent.register_instance_hook("post_reply", "log", log_reply)
  1. 如果没有 API key，可以在 src/agentscope/agent/_agent_meta.py 的 _wrap_with_hooks 中加 print 观察 Hook 执行顺序：
1
2
# 在 pre-hooks 循环前加
print(f"[DEBUG] 执行 pre-hooks for {func_name}, 共 {len(pre_hooks)} 个")

改完后恢复：

1
git checkout src/agentscope/agent/

检查点

你现在理解了：

  • 元类 _AgentMeta 在类定义时自动包装 reply/observe/print 方法
  • _wrap_with_hooks 实现 pre-hook → 原始函数 → post-hook 的执行链
  • 防重入保护确保继承链中 Hook 只执行一次
  • Hook 分实例级和类级别，执行顺序是实例级先于类级别
  • _ReActAgentMeta 扩展了 reasoning 和 acting 的 Hook

自检练习：

  1. 如果你在 AgentBase 的子类中定义了 reply 方法但没有使用 _AgentMeta 元类，Hook 还会生效吗？
  2. _normalize_to_kwargs 的作用是什么？为什么 pre-hook 接收的是 kwargs 字典而不是原始参数？

下一章预告

Hook 是在特定方法前后插入逻辑。还有一种更通用的模式——策略模式：同一个接口，根据不同的情况选择不同的实现。下一章我们看 Formatter 如何使用策略模式适配不同的模型 API。

下一章：第 16 章 策略模式