훅

훅 (Hooks)

에이전트가 실행되는 각 단계 — 모델 요청, 도구 호출, 스트리밍 이벤트 — 에서 동작을 가로채고 수정하고 싶을 때가 있어요. Pydantic AI의 훅(Hooks)은 그걸 서브클래싱 없이 간단한 데코레이터나 생성자 인자로 가능하게 해줘요. 로깅·메트릭·가벼운 검증 같은 애플리케이션 수준 관심사를 다룰 때 권장되는 방법이에요.

출처: 공식문서

빠른 시작

Hooks 인스턴스를 만들고 @hooks.on.* 데코레이터로 훅을 등록한 뒤 에이전트에 넘기면 돼요.

from pydantic_ai import Agent, ModelRequestContext, RunContext
from pydantic_ai.capabilities import Hooks

hooks = Hooks()


@hooks.on.before_model_request
async def log_request(ctx: RunContext, request_context: ModelRequestContext) -> ModelRequestContext:
    print(f'Sending {len(request_context.messages)} messages to the model')
    #> Sending 1 messages to the model
    return request_context


agent = Agent('test', capabilities=[hooks])
result = agent.run_sync('Hello!')
print(result.output)
#> success (no tool calls)

훅 등록하기

데코레이터 등록

hooks.on 네임스페이스가 모든 수명 주기 훅에 대한 데코레이터 메서드를 제공해요. 베어 데코레이터나 파라미터와 함께 쓸 수 있어요.

# Bare decorator
@hooks.on.before_model_request
async def my_hook(ctx, request_context):
    return request_context

# With parameters (timeout, tool filter)
@hooks.on.before_model_request(timeout=5.0)
async def my_timed_hook(ctx, request_context):
    return request_context

같은 이벤트에 여러 훅을 등록할 수 있고, 등록 순서대로 발화해요.

생성자 kwargs

훅 함수를 Hooks 생성자에 직접 넘길 수도 있어요.

agent = Agent('test', capabilities=[Hooks(before_model_request=log_request)])

sync·async 훅 함수 둘 다 받아요. sync 함수는 쓰레드 풀에서 실행되므로 느려도 나머지 실행을 막지 않아요. 주의: Pydantic AI는 sync 훅에 블로킹 코드가 있다고 가정해서 워커 쓰레드에서 실행해요. 그래서 훅이 contextvars.ContextVar에 설정한 값은 훅 밖에서 안 보이고, asyncio.get_running_loop() 같은 API는 워커 쓰레드에 이벤트 루프가 없어 오류를 내요. 이런 게 필요하면 훅을 async로 만드세요.

온디맨드 훅

Hooks는 capability라서 다른 capability처럼 필요할 때 로드할 수 있어요. 선택적·사용자 요청 동작(예: 상세 요청 로깅)에 유용해요.

request_logging_hooks = Hooks(
    id='request-logging',
    description='Use when the user asks for verbose request diagnostics.',
    defer_loading=True,
)

Pydantic AI는 capability가 로드되기 전까지 deferred Hooks 인스턴스가 소유한 훅을 건너뛰어요.

훅 종류

훅은 어느 시점에 발화하느냐에 따라 여러 계열로 나뉘어요.

  • Run 훅 — 에이전트 실행당 한 번 발화: before_run, after_run, run(전체 실행을 감싸는 wrap_run), run_error. (실시간 세션도 하나의 run이라 세션을 감싸며 네 훅이 한 번씩 발화해요.)
  • Node 훅 — 각 그래프 단계(UserPromptNode, ModelRequestNode, CallToolsNode)마다 발화: before_node_run, after_node_run, node_run, node_run_error. agent.run(), agent_run.next(), agent.iter()로 돌려도 똑같이 진행돼요. 예외: agent.run_stream()은 최종 출력이 스트림 중간에서 찾아지면 결과를 바로 주므로, 그 요청은 before_node_run만 받고 wrap_node_run·after_node_run은 안 받아요.
  • 모델 요청 훅 — 각 LLM 호출 주변: before_model_request, after_model_request, model_request, model_request_error. ModelRequestContext가 model, messages, model_settings, model_request_parameters를 묶어줘요. 특정 요청에 모델을 바꾸려면 request_context.model을 다른 Model 인스턴스로 설정하면 돼요. 모델 호출을 완전히 건너뛰려면 before_model_request·model_request에서 SkipModelRequest(response)를 raise해요.
  • 도구 검증 훅 — 모델의 JSON 인자가 파싱·검증될 때 발화: before_tool_validate, after_tool_validate, tool_validate, tool_validate_error. 모든 도구 훅은 call(ToolCallPart)과 tool_def(ToolDefinition) 파라미터를 받아요. 검증을 건너뛰려면 SkipToolValidation(args)를 raise해요. 도구 호출이 deferred될 수 있는 건 인자가 검증된 뒤부터예요(누가 지연을 해결하든 그 인자를 보여주므로). 그래서 ApprovalRequired·CallDeferred는 after_tool_validate(그리고 tool_validate의 handler() 반환 후)에서 raise할 수 있고, before_tool_validate·before handler()·tool_validate_error에서는 UserError예요.
  • 도구 실행 훅 — 도구 함수가 실행될 때 발화: before_tool_execute, after_tool_execute, tool_execute, tool_execute_error. args는 항상 검증된 dict[str, Any]. 실행을 건너뛰려면 SkipToolExecution(result)를 raise해요.
  • 출력 검증 훅 — 구조화 출력이 출력 스키마에 대해 파싱될 때 발화: before_output_validate, after_output_validate, output_validate, output_validate_error. 순수 텍스트·이미지 출력엔 발화하지 않아요. 모든 출력 훅은 output_context(OutputContext) 파라미터를 받아요.
  • 출력 처리 훅 — 출력이 처리될 때(값 추출, 출력 함수 호출, 출력 검증기 실행) 발화: before_output_process, after_output_process, output_process, output_process_error.
  • 도구 준비 — prepare_tools(함수 도구)과 prepare_output_tools(출력 도구)가 모델이 각 단계에서 보는 도구 정의를 필터·수정해요.
  • Deferred 도구 호출 훅 — deferred_tool_calls가 실행 중 deferred 도구 호출(승인 필요·외부 실행)을 인라인으로 해결해요. DeferredToolRequests를 받아 DeferredToolResults(또는 거절 시 None)를 반환해요.
  • 이벤트 스트림 훅 — run_event_stream이 전체 이벤트 스트림을 async 제네레이터로 감싸고, event가 개별 이벤트를 관찰해요. 이벤트 클래스를 넘겨 콜백을 필터할 수 있어요.

이벤트만 관찰하고 싶을 때

이벤트만 보고 싶다면 Hooks capability 없이 @agent.on_event가 에이전트에 직접 리스너를 등록해요. 같은 필터링·타이핑·timeout=을 지원해요.

from pydantic_ai import Agent, FunctionToolCallEvent, RunContext

agent = Agent('test')
called_tools: list[str] = []


@agent.on_event(FunctionToolCallEvent)
async def track_tools(ctx: RunContext, event: FunctionToolCallEvent) -> None:
    called_tools.append(event.part.tool_name)

도구 훅 필터링

도구 훅(검증·실행)은 tools 파라미터로 특정 도구만 대상으로 삼을 수 있어요.

@hooks.on.before_tool_execute(tools=['send_email'])
async def audit_dangerous_tools(
    ctx: RunContext,
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
) -> ValidatedToolArgs:
    call_log.append(f'audit: {call.tool_name}')
    return args

일치하는 도구에만 발화하고 다른 도구 호출은 그대로 통과해요.

타임아웃

각 훅은 선택적으로 초 단위 timeout을 지원해요. 훅이 그보다 오래 걸리면 HookTimeoutError가 raise돼요. 데코레이터 파라미터(@hooks.on.before_model_request(timeout=5.0))나 kwargs 생성자로 설정해요.

Wrap 훅

Wrap 훅은 한 작업을 setup/teardown 로직으로 감싸요. hooks.on 네임스페이스에선 wrap 훅이 wrap_ 접두사를 버려요 — hooks.on.model_request는 wrap_model_request에 해당해요. 핸들러는 첫 파라미터로 handler를 받아 그 안에서 실제 작업을 수행해요.

@hooks.on.model_request
async def log_request(
    ctx: RunContext, *, request_context: ModelRequestContext, handler: WrapModelRequestHandler
) -> ModelResponse:
    wrap_log.append('before')
    response = await handler(request_context)
    wrap_log.append('after')
    return response

훅 순서

단일 Hooks 인스턴스 안에서 before_*, after_*, on_*_error는 등록 순서대로 발화하고, wrap_*는 미들웨어처럼 중첩되며 첫 등록 래퍼가 가장 바깥쪽이 돼요. 여러 capability에 걸쳐선 구성 규칙이 적용돼요 — before_*는 capability 순서로, after_*는 역순으로, wrap_*는 첫 capability가 가장 바깥쪽으로 중첩돼요.

오류 훅

오류 훅(hooks.on 네임스페이스의 *_error, AbstractCapability의 on_*_error)은 raise-to-propagate, return-to-recover 의미를 써요: 원래 오류를 raise하면 그대로 전파(기본), 다른 예외를 raise하면 오류를 변환, 결과를 return하면 오류를 억제.

ModelRetry로 재시도 트리거, ToolFailed로 실패 보고

훅은 ModelRetry를 raise해 커스텀 메시지로 모델에 재시도를 요청할 수 있어요(도구 함수·출력 검증기에서 쓰는 것과 같은 예외).

  • 모델 요청 훅(after_model_request, wrap_model_request, on_model_request_error)에서: 재시도 메시지가 RetryPromptPart로 모델에 전송되고, 재시도 횟수는 에이전트 재시도 예산의 출력 쪽에 계상돼요.
  • 도구 훅에서: 도구 재시도 프롬프트로 변환되고 재시도 횟수는 도구의 max_retries 한도에 계상돼요.
  • 출력 훅에서: 재시도 프롬프트로 변환. 도구 출력이면 도구의 max_retries, 텍스트 출력이면 출력 쪽 예산에 계상돼요.

ModelRetry는 일시적인 오류(같은 호출을 다시 시도하면 성공할 수 있음)에, ToolFailed는 확정적인 실패(재시도해도 소용없고 모델이 결과를 보고 적응해야 함)에 써요. 예를 들어 업스트림 상태 코드로 판단하는 훅은 500·429면 재시도를, 404·403이면 실패 보고를 하게 할 수 있어요. ToolFailed는 도구의 재시도 예산을 소모하지 않으면서 실패를 모델에 알려줘요.

Hooks vs AbstractCapability

  • Hooks: 애플리케이션 수준 훅(로깅·메트릭), 빠른 일회성 인터셉터, 상태 설정 불필요, 단일 파일 스크립트에 적합.
  • AbstractCapability: 재사용·패키지화된 capability, 도구+훅+지시+설정을 합친 것, 복잡한 per-run 상태 관리, 멀티에이전트 공유 동작에 적합.

더 알아보기 (Learn more)