스텝 훅

스텝 훅 (Step Hooks)

스텝 훅은 실행 안의 각 작업 단위 — 모든 크루 태스크와 모든 플로우 메서드 — 를 가로챕니다. 개별 LLM·도구 호출 수준에는 손대지 않고, 스텝에 들어가는 것을 검사·재작성하고, 나오는 것을 변환하고, 단계별 진행을 추적하는 데 사용해요.

출처: 공식문서

본문

개요

스텝을 다루는 인터셉션 지점은 두 개입니다.

지점 시점 ctx.payload
PRE_STEP 태스크나 플로우 메서드가 실행되기 전 스텝 입력(아래 참조)
POST_STEP 태스크나 플로우 메서드가 실행된 후 스텝 출력(아래 참조)

payload 가 무엇이냐는 ctx.kind 에 달려 있습니다.

ctx.kind PRE_STEP payload POST_STEP payload
"task" 에이전트에게 전달된 컨텍스트 문자열 TaskOutput 객체
"flow_method" 메서드의 파라미터 dict 메서드의 반환 값

플로우 메서드의 경우 위치 인자는 params dict 에 _0, _1, … 키로, 키워드 인자는 자기 이름으로 나타납니다. 수정·교체는 실제 호출에 다시 매핑됩니다.

훅 시그니처

from crewai.hooks import on, HookAborted, InterceptionPoint

@on(InterceptionPoint.PRE_STEP)
def step_hook(ctx) -> Any | None:
    # ctx.payload 를 제자리에서 수정하거나,
    # None 이 아닌 값을 반환해 교체하거나,
    # raise HookAborted(reason, source) 로 스텝 중지
    return None

컨텍스트 스키마

두 지점 모두 StepContext 를 받습니다.

class StepContext(InterceptionContext):
    payload: Any            # 스텝 입력(사전) 또는 스텝 출력(사후)
    kind: str | None        # "task" 또는 "flow_method"
    step_name: str | None   # 태스크 이름/설명, 또는 플로우 메서드 이름
    output: Any             # POST_STEP만: payload와 같은 객체
    agent: Any              # 태스크 스텝: 실행 중인 에이전트(그 외엔 None)
    agent_role: str | None  # 태스크 스텝: 에이전트 역할(그 외엔 None)
    task: Any               # 태스크 스텝: Task 인스턴스(그 외엔 None)
    crew: Any               # 스텝 지점에서는 None
    flow: Any               # 플로우 메서드 스텝: Flow 인스턴스(그 외엔 None)

태스크 스텝에서 step_name 은 태스크의 name(없으면 설명으로 대체)입니다. 플로우 메서드 스텝에서는 메서드 이름입니다.

흔한 사용 사례

스텝 추적

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

태스크 컨텍스트 재작성

@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."

태스크 출력 변환

@on(InterceptionPoint.POST_STEP)
def normalize_output(ctx):
    if ctx.kind != "task":
        return None
    ctx.payload.raw = ctx.payload.raw.strip()

POST_STEP 은 태스크 출력이 저장되기 전에 실행되므로, 재작성은 출력이 사용되는 모든 곳(하위 태스크 컨텍스트, 콜백, 최종 크루 출력, 태스크의 output_file 디스크 파일)에 전파됩니다.

플로우 메서드 가드

@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= 필터를 지원합니다(태스크 스텝에서는 실행 에이전트의 역할과 대조).

@on(InterceptionPoint.POST_STEP, agents=["Researcher"])
def log_research_steps(ctx):
    print(f"research step done: {ctx.step_name}")

스텝 중단

PRE_STEP 에서 HookAborted 를 발생시키면 에이전트나 메서드 작업이 일어나기 전에 스텝이 멈추고, 중단은 그 reason 과 함께 실행 밖으로 전파됩니다(삼켜지지 않습니다). 스텝 훅이 발생시킨 다른 예외는 다른 모든 지점처럼 삼켜집니다(fail-open).

테스트에서 훅 관리

from crewai.hooks import clear_all_hooks

clear_all_hooks()  # 스텝 포함 모든 지점을 클리어

더 알아보기