执行边界钩子

使用 @on 装饰器拦截团队和流程执行的开始、输入、输出和结束

执行边界钩子拦截一次运行的最外层边界——在任何 工作开始之前、输入解析之时、最终结果就绪之时，以及 执行完成之时。它们会为团队和流程触发，并且是进行 运行级策略检查、输入重写和输出清理的合适位置。

概述

四个拦截点覆盖这些边界：

拦截点	时机	ctx.payload
EXECUTION_START	团队或流程即将开始	输入 dict
INPUT	为执行解析后的输入	输入 dict
OUTPUT	最终结果已就绪	输出对象
EXECUTION_END	执行已完成（成功或失败）	输出对象，失败时为 None

对于团队，输出载荷是 CrewOutput。对于流程，它是最终的 流程方法结果。

钩子签名

from crewai.hooks import on, HookAborted, InterceptionPoint

@on(InterceptionPoint.EXECUTION_START)
def boundary_hook(ctx) -> Any | None:
    # Mutate ctx.payload in place, or
    # return a non-None value to replace it, or
    # raise HookAborted(reason, source) to stop the run
    return None

边界钩子遵循标准约定：继续（return None）、原地 修改、通过返回来替换，或通过抛出 HookAborted 来中止。任何边界上的 中止都会携带其原因从 kickoff() 向外传播。

上下文 Schema

每个拦截点都会接收一个带类型的上下文。所有上下文共享基础字段：

class InterceptionContext:
    payload: Any            # The interceptable value (see table above)
    agent: Any = None       # Not populated at execution boundaries
    agent_role: str | None  # Not populated at execution boundaries
    task: Any = None        # Not populated at execution boundaries
    crew: Any = None        # The Crew instance (crew runs only)
    flow: Any = None        # The Flow instance (flow runs only)

各拦截点的上下文为载荷添加了一个命名别名：

class ExecutionStartContext(InterceptionContext):
    inputs: dict            # Same dict as payload

class InputContext(InterceptionContext):
    inputs: dict            # Same dict as payload

class OutputContext(InterceptionContext):
    output: Any             # The output object

class ExecutionEndContext(InterceptionContext):
    output: Any                    # The output object (None when status == "failed")
    status: str                    # "completed" or "failed"
    error: BaseException | None    # The exception when status == "failed"

NOTE

ctx.inputs 是原始输入 dict 的别名，因此通过 任一名称进行原地编辑都是等价的。如果较早的钩子通过 返回新 dict 替换了载荷，则只会重新绑定 ctx.payload——当钩子可能链式执行时，请始终读写 ctx.payload。

团队运行与流程运行

边界钩子在两种运行时上都会触发，而团队执行在内部依赖 流程运行时。因此，在 crew.kickoff() 期间，全局边界钩子会为 团队边界（设置了 ctx.crew，ctx.flow 为 None）以及 内部流程（设置了 ctx.flow，ctx.crew 为 None）触发。按运行时进行区分：

@on(InterceptionPoint.OUTPUT)
def crew_output_only(ctx):
    if ctx.crew is None:
        return None  # Skip the internal flow (or a bare flow)
    ctx.payload.raw = ctx.payload.raw.strip()

常见用例

开始时的策略检查

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

输入重写

@on(InterceptionPoint.INPUT)
def add_defaults(ctx):
    if ctx.crew is None:
        return None
    ctx.payload.setdefault("locale", "en-US")
    ctx.payload["topic"] = ctx.payload["topic"].strip().lower()

重写后的输入会流入任务插值，因此这次运行的行为就如同 使用修改后的 dict 启动一样。

重写时优先使用 INPUT，并将 EXECUTION_START 视为允许/拒绝 关口。在 EXECUTION_START 处的重写仍会生效——在团队中，它们还会 馈入 before_kickoff 回调；在流程中，它们的落地方式与 INPUT 重写完全相同。

输出清理

import re

@on(InterceptionPoint.OUTPUT)
def redact_emails(ctx):
    if ctx.crew is None:
        return None
    ctx.payload.raw = re.sub(
        r"\b[\w.+-]+@[\w-]+\.[\w.]+\b", "[EMAIL-REDACTED]", ctx.payload.raw
    )

OUTPUT 在 EXECUTION_END 之前运行，并且两者都会看到来自较早钩子的（可能已被替换的） 载荷；最终重写后的值就是 kickoff() 返回的内容。

观察失败

EXECUTION_END 每次执行恰好触发一次，无论成功还是失败 都是如此。当运行抛出异常时——任务错误、流程方法异常，或来自较早拦截点的 HookAborted——该钩子会收到 status="failed"，并在 ctx.error 中收到该异常，而原始异常仍会保持不变地从 kickoff() 向外传播：

@on(InterceptionPoint.EXECUTION_END)
def report_outcome(ctx):
    if ctx.status == "failed":
        notify_policy_engine(status="failed", error=repr(ctx.error))
    else:
        notify_policy_engine(status="completed")

有两点需要注意：当 EXECUTION_START 从未被分派时，EXECUTION_END 不会 触发（在开始时中止意味着边界从未打开，因此没有 可配对的结束），并且从失败路径的 EXECUTION_END 分派中抛出 HookAborted 会被忽略——已经没有可中止的内容，原始错误 胜出。

顺序

对于团队运行，边界顺序为：

EXECUTION_START → before_kickoff callbacks → INPUT → tasks execute → OUTPUT → EXECUTION_END

对于流程运行，边界钩子会在生命周期 事件开始之前解析输入：

EXECUTION_START → INPUT → FlowStartedEvent → flow methods execute → OUTPUT → EXECUTION_END → FlowFinishedEvent

FlowStartedEvent 携带钩子解析后的输入，而在边界钩子中重写 inputs["id"] 会重定向状态恢复。在 EXECUTION_START 处的中止仍会表现为先出现 FlowStartedEvent，随后出现 FlowFailedEvent；后者在中止时发出，其载荷由在该中止之前运行的 钩子解析得出。

同一拦截点上的钩子按注册顺序运行，先运行全局钩子，然后运行 团队作用域钩子。每次分派都会发出遥测（HookDispatchedEvent）。

在测试中管理钩子

from crewai.hooks import clear_all_hooks

clear_all_hooks()  # Clears every point, including boundaries

相关文档

• 执行钩子概览 →
• 步骤钩子 →
• LLM 调用钩子 →
• 工具调用钩子 →