告别硬编码！AgentScope钩子系统让智能体行为扩展更优雅

【免费下载链接】agentscope 项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope

你是否还在为修改智能体行为而反复修改核心代码？是否想在不侵入原有逻辑的情况下添加日志、监控或自定义处理？AgentScope的钩子系统正是为解决这些问题而来。本文将通过实例带你掌握实例与类级别钩子的扩展机制，让智能体功能扩展变得灵活高效。

什么是钩子系统？

钩子（Hook）是AgentScope中的扩展点，允许开发者在特定位置自定义智能体行为，提供了一种灵活的方式来修改或扩展智能体的功能，而无需更改其核心实现。

在AgentScope中，钩子围绕智能体的核心函数实现，支持多种类型的钩子：

智能体类	核心函数	钩子类型	描述
AgentBase 及其子类	reply	pre_reply、post_reply	智能体回复消息前/后的钩子
	print	pre_print、post_print	向目标输出打印消息前/后的钩子
	observe	pre_observe、post_observe	从环境观察消息前/后的钩子
ReActAgentBase 及其子类	_reasoning	pre_reasoning、post_reasoning	智能体推理过程前/后的钩子
	_acting	pre_acting、post_acting	智能体行动过程前/后的钩子

更多钩子类型详情可查看官方文档：docs/tutorial/zh_CN/src/task_hook.py

钩子签名规范

AgentScope为所有钩子提供了统一的签名，便于开发者实现自定义逻辑。

前置钩子签名

def pre_hook_template(
    self: AgentBase | ReActAgentBase,
    kwargs: dict[str, Any],
) -> dict[str, Any] | None:  # 修改后的输入
    """前置钩子模板。"""
    pass

后置钩子签名

def post_hook_template(
    self: AgentBase | ReActAgentBase,
    kwargs: dict[str, Any],
    output: Any,  # 目标函数的输出
) -> Any:  # 修改后的输出
    """后置钩子模板。"""
    pass

钩子执行顺序

AgentScope提供实例级（instance）和类级（class）钩子，其区别在于钩子函数的作用范围。它们按以下顺序执行：

钩子管理方法

AgentScope提供了完善的钩子管理方法，方便开发者注册、移除和清除钩子：

级别	方法	描述
实例级	register_instance_hook	为当前对象注册钩子
	remove_instance_hook	移除当前对象的钩子
	clear_instance_hooks	清除当前对象的所有钩子
类级	register_class_hook	为该类的所有对象注册钩子
	remove_class_hook	移除该类所有对象的钩子
	clear_class_hooks	清除该类所有对象的钩子

实战案例：消息转发钩子

下面是一个将消息转发到Studio的实际钩子实现，位于src/agentscope/hooks/_studio_hooks.py：

def as_studio_forward_message_pre_print_hook(
    self: AgentBase,
    kwargs: dict[str, Any],
    studio_url: str,
    run_id: str,
) -> None:
    """The pre-speak hook to forward messages to the studio."""
    msg = kwargs["msg"]

    message_data = msg.to_dict()

    if hasattr(self, "_reply_id"):
        reply_id = getattr(self, "_reply_id")
    else:
        reply_id = shortuuid.uuid()

    n_retry = 0
    while True:
        try:
            res = requests.post(
                f"{studio_url}/trpc/pushMessage",
                json={
                    "runId": run_id,
                    "replyId": reply_id,
                    "name": reply_id,
                    "role": "assistant",
                    "msg": message_data,
                },
            )
            res.raise_for_status()
            break
        except Exception as e:
            if n_retry < 3:
                n_retry += 1
                continue
            raise e from None

实例级与类级别钩子应用

下面通过一个完整示例展示如何使用实例级和类级别钩子：

创建测试智能体

class TestAgent(AgentBase):
    """用于演示钩子的测试智能体。"""
    async def reply(self, msg: Msg) -> Msg:
        """回复消息。"""
        return msg

定义钩子函数

def instance_pre_reply_hook(
    self: AgentBase,
    kwargs: dict[str, Any],
) -> dict[str, Any]:
    """修改消息内容的前置回复钩子。"""
    msg = kwargs["msg"]
    msg.content += "[instance-pre-reply]"
    return {"msg": msg}

def cls_pre_reply_hook(
    self: AgentBase,
    kwargs: dict[str, Any],
) -> dict[str, Any]:
    """修改消息内容的前置回复钩子。"""
    msg = kwargs["msg"]
    msg.content += "[cls-pre-reply]"
    return {"msg": msg}

注册钩子并测试

# 注册类钩子
TestAgent.register_class_hook(
    hook_type="pre_reply",
    hook_name="test_pre_reply",
    hook=cls_pre_reply_hook,
)

# 注册实例钩子
agent = TestAgent()
agent.register_instance_hook(
    hook_type="pre_reply",
    hook_name="test_pre_reply",
    hook=instance_pre_reply_hook,
)

# 测试钩子效果
async def example_test_hook() -> None:
    msg = Msg(name="user", content="Hello, world!", role="user")
    res = await agent(msg)
    print("响应内容：", res.content)  # 输出: Hello, world![cls-pre-reply][instance-pre-reply]

asyncio.run(example_test_hook())

钩子使用注意事项

1. 执行顺序：钩子按注册顺序执行，多个钩子可以链式连接
2. 返回值处理：非None返回值会传递给下一个钩子或核心函数
3. 避免循环调用：不要在钩子内调用核心函数（reply/speak等）

通过钩子系统，我们可以轻松实现智能体的功能扩展，如日志记录、性能监控、消息修改等。更多钩子应用实例可参考：examples/functionality/

总结

AgentScope的钩子系统为智能体行为扩展提供了灵活强大的机制，通过实例级和类级别的钩子管理，开发者可以在不修改核心代码的情况下定制智能体行为。掌握钩子系统将帮助你构建更具扩展性和可维护性的智能体应用。

希望本文对你理解AgentScope钩子系统有所帮助，如有疑问欢迎查阅官方文档或提交issue。

【免费下载链接】agentscope 项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope