主题
步骤钩子
在 CrewAI 中使用 PRE_STEP 和 POST_STEP 钩子拦截任务与流程方法步骤
步骤钩子会拦截一次执行中的每个工作单元：每个团队任务以及每个流程方法。你可以用它们检查或改写进入某个步骤的内容、转换输出的内容，或逐步追踪进度——而无需触及单个 LLM 或工具调用的层面。
概述
两个拦截点覆盖各个步骤：
| 拦截点 | 时机 | ctx.payload |
|---|---|---|
PRE_STEP | 任务或流程方法运行之前 | 步骤输入（见下文） |
POST_STEP | 任务或流程方法运行之后 | 步骤输出（见下文） |
payload 中存放什么取决于 ctx.kind：
ctx.kind | PRE_STEP payload | POST_STEP payload |
|---|---|---|
"task" | 传给智能体的上下文字符串 | TaskOutput 对象 |
"flow_method" | 方法参数组成的 dict | 方法的返回值 |
对于流程方法，位置参数会以 _0、_1……为键出现在参数字典中，关键字参数则以各自的名称作为键；对它们的修改和替换会被映射回实际的调用。
钩子签名
python
from crewai.hooks import on, HookAborted, InterceptionPoint
@on(InterceptionPoint.PRE_STEP)
def step_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 step
return None上下文 Schema
两个拦截点都会收到一个 StepContext：
python
class StepContext(InterceptionContext):
payload: Any # Step input (pre) or step output (post)
kind: str | None # "task" or "flow_method"
step_name: str | None # Task name/description, or flow method name
output: Any # POST_STEP only: same object as payload
agent: Any # Task steps: the executing agent (else None)
agent_role: str | None # Task steps: the agent's role (else None)
task: Any # Task steps: the Task instance (else None)
crew: Any # None for step points
flow: Any # Flow-method steps: the Flow instance (else None)对于任务步骤，step_name 是任务的 name（若没有则回退到其描述）。对于流程方法步骤，它是方法名。
常见用例
步骤追踪
python
@on(InterceptionPoint.POST_STEP)
def trace_steps(ctx):
print(f"{ctx.kind} '{ctx.step_name}' finished")改写任务上下文
python
@on(InterceptionPoint.PRE_STEP)
def inject_disclaimer(ctx):
if ctx.kind != "task":
return None
return f"{ctx.payload}\n\nNote: treat all figures as estimates."转换任务输出
python
@on(InterceptionPoint.POST_STEP)
def normalize_output(ctx):
if ctx.kind != "task":
return None
ctx.payload.raw = ctx.payload.raw.strip()NOTE
POST_STEP 在任务输出被存储之前运行，因此改写会传播到使用该输出的所有地方：下游任务上下文、回调、最终的团队输出，以及磁盘上的任务 output_file。
守护流程方法
python
@on(InterceptionPoint.PRE_STEP)
def guard_publish(ctx):
if ctx.kind == "flow_method" and ctx.step_name == "publish":
if not ctx.flow.state.get("reviewed"):
raise HookAborted(reason="publish requires review", source="review-gate")按智能体过滤
步骤钩子支持与其他拦截点相同的 agents= 过滤器（任务步骤中按执行智能体的角色进行匹配）：
python
@on(InterceptionPoint.POST_STEP, agents=["Researcher"])
def log_research_steps(ctx):
print(f"research step done: {ctx.step_name}")中止步骤
在 PRE_STEP 中抛出 HookAborted 会在任何智能体或方法工作发生之前中止该步骤，并且中止会带着其原因传播到执行之外——不会被吞掉。步骤钩子抛出的任何其他异常都会被吞掉（fail-open），与其他拦截点相同。
在测试中管理钩子
python
from crewai.hooks import clear_all_hooks
clear_all_hooks() # Clears every point, including steps