执行钩子

使用 @on 装饰器拦截、修改和控制 CrewAI 的运行时——一个约定覆盖所有拦截点

执行钩子可对你的 CrewAI 智能体的运行时行为提供细粒度控制。 与在团队执行前后运行的 kickoff 钩子不同，执行钩子会在执行期间拦截特定操作——从 一次运行开始的时刻起，经过每次模型调用、工具调用，以及任务或流程方法步骤， 直到最终输出。

钩子使用 @on 装饰器编写：一个注册 API 和一个 约定覆盖框架中的每个拦截点。

from crewai.hooks import on, HookAborted, InterceptionPoint

@on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file"])
def guard_deletes(ctx):
    raise HookAborted(reason="file deletion is not allowed", source="policy")

NOTE

特定于拦截点的装饰器（@before_llm_call、@after_tool_call、...）仍可 保持不变地工作——它们是同一引擎之上的适配器。请参阅 特定于拦截点的装饰器（旧版），见 本页末尾。

约定

每个钩子都是一个接收单个带类型上下文的同步可调用对象：

from crewai.hooks import on, HookAborted, InterceptionPoint

@on(InterceptionPoint.INPUT)
def add_defaults(ctx):
    # 1. Observe: read anything off the context.
    # 2. Mutate in place: change ctx.payload or nested fields directly.
    ctx.payload.setdefault("locale", "en-US")
    # 3. Or replace: return a new value to swap ctx.payload.
    # 4. Or abort: raise HookAborted(reason, source) to stop the operation.
    return None

钩子可以执行以下四种操作之一：

操作	方式	效果
继续	return None（或不返回任何内容）	操作保持不变地继续
修改	原地更改 ctx.payload / 字段	更改在下游可见
替换	return new_payload	非 None 的返回值会替换 ctx.payload
中止	raise HookAborted(reason, source)	操作被停止；原因会向外传播

注册钩子

使用 @on 注册全局钩子。它接受 agents= / tools= 过滤器，以将 钩子限定到特定的智能体角色或工具名称：

from crewai.hooks import on, InterceptionPoint

@on(InterceptionPoint.POST_TOOL_CALL, agents=["researcher"], tools=["web_search"])
def log_search_results(ctx):
    print(f"search returned: {(ctx.tool_result or '')[:80]}")

应用于 @CrewBase 类中的方法时，@on 会注册一个 团队范围的钩子，仅在该团队运行时生效：

from crewai import CrewBase
from crewai.hooks import on, InterceptionPoint

@CrewBase
class MyProjCrew:
    @on(InterceptionPoint.PRE_MODEL_CALL)
    def validate_inputs(self, ctx):
        # Only applies to this crew
        return None

拦截点目录

每个类别都有详细的指南，涵盖其上下文模式、载荷 语义和示例。

执行边界

拦截点	触发时机	ctx.payload
EXECUTION_START	团队或流程即将开始	输入 dict
INPUT	该执行已解析的输入	输入 dict
OUTPUT	最终结果已就绪	输出对象
EXECUTION_END	团队或流程已完成	输出对象

模型边界 与 工具边界

拦截点	触发时机	钩子接收
PRE_MODEL_CALL	LLM 调用之前	LLMCallHookContext
POST_MODEL_CALL	LLM 调用之后	LLMCallHookContext（已设置 response）
PRE_TOOL_CALL	工具运行之前	ToolCallHookContext
POST_TOOL_CALL	工具运行之后	ToolCallHookContext（已设置结果）

在这四个点上，钩子会直接接收丰富的旧版上下文 作为其参数——没有单独的 ctx.payload。就地修改 ctx.messages / ctx.tool_input，并从后置钩子返回字符串以替换 响应 / 工具结果。

步骤点

拦截点	触发时机	ctx.payload
PRE_STEP	任务或流程方法步骤之前	步骤输入
POST_STEP	任务或流程方法步骤之后	步骤输出

PRE_STEP / POST_STEP 携带 ctx.kind（"task" 或 "flow_method"）和 ctx.step_name。

中止操作

HookAborted 携带 reason 和可选的 source。省略时，source 默认 为触发中止的钩子，这对遥测和失败 消息很有用：

@on(InterceptionPoint.EXECUTION_START)
def enforce_policy(ctx):
    if not ctx.payload.get("authorized"):
        raise HookAborted(reason="unauthorized execution", source="access-control")

组合、排序与 fail-open

• 同一拦截点上的多个钩子按注册顺序运行，全局钩子 优先，然后是执行范围内的钩子。为同一拦截点注册的旧版钩子 也参与同一链。
• （可能被修改的）载荷从一个钩子流向下一钩子。
• HookAborted 按设计传播并停止链。
• 钩子抛出的任何其他异常都会被吞掉（fail-open），因此单个 有缺陷的钩子不会使运行崩溃。
• 当某个拦截点没有注册钩子时，分发只是一次字典查找 （空操作快速路径），因此未使用的拦截点实际上没有成本。

常见模式

安全护栏

@on(InterceptionPoint.PRE_TOOL_CALL)
def block_dangerous_tools(ctx):
    dangerous = {"delete_file", "drop_table", "system_shutdown"}
    if ctx.tool_name in dangerous:
        raise HookAborted(reason=f"{ctx.tool_name} is blocked", source="safety-policy")

@on(InterceptionPoint.PRE_MODEL_CALL)
def iteration_limit(ctx):
    if ctx.iterations > 15:
        raise HookAborted(reason="maximum iterations exceeded", source="loop-guard")

人在回路审批

@on(InterceptionPoint.PRE_TOOL_CALL, tools=["send_email", "make_payment"])
def require_approval(ctx):
    response = ctx.request_human_input(
        prompt=f"Approve {ctx.tool_name}?",
        default_message="Type 'yes' to approve:",
    )
    if response.lower() != "yes":
        raise HookAborted(reason="rejected by operator", source="approval-gate")

净化输出

非 None 返回值会替换可拦截的值，因此转换 就是普通的 return 语句：

import re

@on(InterceptionPoint.POST_MODEL_CALL)
def redact_keys(ctx):
    return re.sub(
        r'(api[_-]?key)["\']?\s*[:=]\s*["\']?[\w-]+',
        r"\1: [REDACTED]",
        ctx.response,
        flags=re.IGNORECASE,
    )

观察步骤

@on(InterceptionPoint.POST_STEP)
def trace_steps(ctx):
    print(f"{ctx.kind} '{ctx.step_name}' finished")

遥测

每当某个拦截点实际分发到至少一个钩子时，CrewAI 都会在事件总线上发出一个 HookDispatchedEvent，其中包含该拦截点、结果 （proceeded / modified / aborted）、钩子数量、持续时间，以及——对于 中止——原因和来源。空操作快速路径不会发出任何内容。

在测试中管理钩子

全局钩子在进程的整个生命周期内持续存在。请在测试之间重置它们：

import pytest
from crewai.hooks import clear_all_hooks

@pytest.fixture(autouse=True)
def reset_hooks():
    clear_all_hooks()
    yield
    clear_all_hooks()

最佳实践

1. 保持钩子专注——每个钩子只承担一项明确职责；注册多个 小型钩子，而不是一个包揽所有事情的钩子。
2. 保持钩子快速——钩子会在其拦截点的每次分发时运行；避免繁重 计算，并延迟导入重型依赖。
3. 优先限定作用域——使用 agents= / tools= 过滤器和团队范围的 注册，而不是无条件的全局钩子。
4. 明确中止——抛出带有有意义 reason 和 source 的 HookAborted；该上下文会出现在错误消息和遥测中。请记住 任何其他异常都会被吞掉（fail-open），因此不要依赖抛出 ValueError 来停止运行。

特定拦截点装饰器（旧版）

在 @on 之前，LLM 和工具调用通过专用装饰器对进行挂接。 这些装饰器保持不变并继续工作——它们是同一分发器之上的适配器，因此 它们会与 @on 钩子在同一注册顺序链中组合：

from crewai.hooks import before_llm_call, after_llm_call, before_tool_call, after_tool_call

@before_llm_call
def limit_iterations(context):
    if context.iterations > 10:
        return False  # Block execution

@after_tool_call
def log_tool_result(context):
    print(f"Tool {context.tool_name} completed")

与 @on 的区别：

• 它们仅覆盖四个模型/工具拦截点——没有执行边界，也没有 步骤。
• 阻塞方式是 return False，不附带中止原因或来源。
• 它们接收与 @on 钩子在 模型/工具拦截点接收的相同的丰富上下文——LLMCallHookContext（具有完整 执行器访问权限）和 ToolCallHookContext。
• 团队作用域的工作方式相同：将装饰器应用于 @CrewBase 类中的方法。
• 它们支持相同的 agents= / tools= 过滤器。

对于已经使用 return False 语义的现有代码库，或者当你想要特定拦截点的类型化签名时， 你可能仍然更偏好它们。有关详细指南——上下文属性、模式和管理 API （register_* / unregister_* / clear_*）——请参阅：

• LLM 调用钩子 →
• 工具调用钩子 →

相关文档

• Kickoff 前后钩子 →
• 人在回路 →