실행 경계 훅

실행 경계 훅 (Execution Boundary Hooks)

실행 경계 훅(execution boundary hooks)은 실행(run)의 가장 바깥쪽 경계를 가로채요 — 어떤 작업도 시작되기 전, 입력이 결정될 때, 최종 결과가 준비됐을 때, 그리고 실행이 끝날 때 말이죠. 이 훅은 crew와 flow 모두에서 동작하며, run 레벨의 정책 검사, 입력 재작성, 출력 정화(sanitization)를 넣기에 딱 좋은 자리예요.

출처: 문서

본문

Overview (개요)

경계를 덮는 네 가지 가로채기 지점이 있어요:

crew의 경우 출력 payload는 CrewOutput이에요. flow의 경우 최종 flow-method 결과예요.

Hook Signature (훅 시그니처)

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), 제자리에서 수정(mutate in place), 반환하여 교체(replace by returning), 또는 HookAborted를 raise해서 중단해요. 어떤 경계에서든 중단은 kickoff() 밖으로 reason과 함께 전파돼요.

Context 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)

지점별 컨텍스트는 payload에 대한 명명된 별칭(alias)을 추가해요:

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"

ctx.inputs는 원본 inputs dict의 별칭이에요. 그래서 어느 이름을 쓰든 제자리 수정은 동일해요. 이전 훅이 새 dict를 반환해서 payload를 교체했다면 ctx.payload만 다시 바인딩돼요 — 훅이 연결될 수 있는 상황이니 항상 ctx.payload를 읽고 써야 해요.

Crew Runs vs. Flow Runs

경계 훅은 두 런타임 모두에서 동작하며, crew 실행은 내부적으로 flow 런타임 위에서 돌아요. 따라서 crew.kickoff() 동안 전역 경계 훅은 crew 경계(ctx.crew 설정, ctx.flow는 None) 그리고 내부 flow(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()

Common Use Cases (일반적 사용 사례)

Policy Check at Start (시작 시 정책 검사)

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

Input Rewriting (입력 재작성)

@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()

재작성된 입력은 태스크 보간(task interpolation)으로 흘러들어가서, 마치 수정된 dict로 kickoff된 것처럼 run이 동작해요.

재작성에는 INPUT을 선호하고, EXECUTION_START는 허용/거부(allow/deny) 게이트로 취급하세요. EXECUTION_START에서의 재작성도 여전히 존중돼요 — crew에서는 before_kickoff 콜백에도 공급되고, flow에서는 INPUT 재작성과 정확히 동일하게 동작해요.

Output Sanitization (출력 정화)

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보다 먼저 실행되고, 둘 다 이전 훅에서 (아마 교체된) payload를 봐요. 최종 재작성된 값이 kickoff()이 반환하는 값이에요.

Observing Failures (실패 관찰)

EXECUTION_END는 실행당 정확히 한 번, 성공과 실패 모두에서 동작해요. run이 raise할 때 — 태스크 에러, flow-method 예외, 또는 이전 지점에서의 HookAborted — 훅은 ctx.error에 예외를 담은 status="failed"를 받고, 원래 예외는 변경 없이 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가 dispatch되지 않았다면 EXECUTION_END는 동작하지 않아요 (시작에서 중단되면 경계가 열리지 않으니 짝을 이룰 끝이 없는 거죠). 그리고 실패 경로의 EXECUTION_END dispatch에서 HookAborted를 raise하는 것은 무시돼요 — 중단할 것이 남아 있지 않고, 원래 에러가 우선하기 때문이에요.

Ordering (순서)

crew run의 경계 순서는:

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

flow run의 경우, 경계 훅이 라이프사이클 이벤트가 시작되기 전에 입력을 결정해요:

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

FlowStartedEvent는 훅에서 결정된 입력을 담고, 경계 훅에서 inputs["id"]를 재작성하면 상태 복원(state restoration)이 리다이렉트돼요. EXECUTION_START에서의 중단은 여전히 FlowStartedEvent 다음에 FlowFailedEvent로 표면화되며, 그 앞에서 실행된 훅들이 결정한 payload를 담아 중단 시점에 emit돼요.

같은 지점의 훅은 등록 순서대로 실행되고, 전역 훅이 먼저, 그다음 crew 범위 훅이에요. 텔레메트리(HookDispatchedEvent)는 dispatch마다 emit돼요.

Managing Hooks in Tests (테스트에서 훅 관리)

from crewai.hooks import clear_all_hooks

clear_all_hooks()  # Clears every point, including boundaries

Related Documentation (관련 문서)

• Execution Hooks Overview
• Step Hooks
• LLM Call Hooks
• Tool Call Hooks

더 알아보기 (Learn more)

• CrewAI 학습 개요