메인 콘텐츠로 건너뛰기
실행 훅(Execution Hooks)은 CrewAI 에이전트의 런타임 동작을 세밀하게 제어할 수 있게 해줍니다. 크루 실행 전후에 실행되는 킥오프 훅과 달리, 실행 훅은 에이전트 실행 중 특정 작업을 가로채서 동작을 수정하고, 안전성 검사를 구현하며, 포괄적인 모니터링을 추가할 수 있습니다.

실행 훅의 유형

CrewAI는 두 가지 주요 범주의 실행 훅을 제공합니다:

1. LLM 호출 훅

언어 모델 상호작용을 제어하고 모니터링합니다:
  • LLM 호출 전: 프롬프트 수정, 입력 검증, 승인 게이트 구현
  • LLM 호출 후: 응답 변환, 출력 정제, 대화 기록 업데이트
사용 사례:
  • 반복 제한
  • 비용 추적 및 토큰 사용량 모니터링
  • 응답 정제 및 콘텐츠 필터링
  • LLM 호출에 대한 사람의 승인
  • 안전 가이드라인 또는 컨텍스트 추가
  • 디버그 로깅 및 요청/응답 검사
LLM 훅 문서 보기 →

2. 도구 호출 훅

도구 실행을 제어하고 모니터링합니다:
  • 도구 호출 전: 입력 수정, 매개변수 검증, 위험한 작업 차단
  • 도구 호출 후: 결과 변환, 출력 정제, 실행 세부사항 로깅
사용 사례:
  • 파괴적인 작업에 대한 안전 가드레일
  • 민감한 작업에 대한 사람의 승인
  • 입력 검증 및 정제
  • 결과 캐싱 및 속도 제한
  • 도구 사용 분석
  • 디버그 로깅 및 모니터링
도구 훅 문서 보기 →

훅 등록 방법

1. 데코레이터 기반 훅 (권장)

훅을 등록하는 가장 깔끔하고 파이썬스러운 방법:
from crewai.hooks import before_llm_call, after_llm_call, before_tool_call, after_tool_call

@before_llm_call
def limit_iterations(context):
    """반복 횟수를 제한하여 무한 루프를 방지합니다."""
    if context.iterations > 10:
        return False  # 실행 차단
    return None

@after_llm_call
def sanitize_response(context):
    """LLM 응답에서 민감한 데이터를 제거합니다."""
    if "API_KEY" in context.response:
        return context.response.replace("API_KEY", "[수정됨]")
    return None

@before_tool_call
def block_dangerous_tools(context):
    """파괴적인 작업을 차단합니다."""
    if context.tool_name == "delete_database":
        return False  # 실행 차단
    return None

@after_tool_call
def log_tool_result(context):
    """도구 실행을 로깅합니다."""
    print(f"도구 {context.tool_name} 완료")
    return None

2. 크루 범위 훅

특정 크루 인스턴스에만 훅을 적용합니다:
from crewai import CrewBase
from crewai.project import crew
from crewai.hooks import before_llm_call_crew, after_tool_call_crew

@CrewBase
class MyProjCrew:
    @before_llm_call_crew
    def validate_inputs(self, context):
        # 이 크루에만 적용됩니다
        print(f"{self.__class__.__name__}에서 LLM 호출")
        return None

    @after_tool_call_crew
    def log_results(self, context):
        # 크루별 로깅
        print(f"도구 결과: {context.tool_result[:50]}...")
        return None

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential
        )

훅 실행 흐름

LLM 호출 흐름

에이전트가 LLM을 호출해야 함

[LLM 호출 전 훅 실행]
    ├→ 훅 1: 반복 횟수 검증
    ├→ 훅 2: 안전 컨텍스트 추가
    └→ 훅 3: 요청 로깅

훅이 False를 반환하는 경우:
    ├→ LLM 호출 차단
    └→ ValueError 발생

모든 훅이 True/None을 반환하는 경우:
    ├→ LLM 호출 진행
    └→ 응답 생성

[LLM 호출 후 훅 실행]
    ├→ 훅 1: 응답 정제
    ├→ 훅 2: 응답 로깅
    └→ 훅 3: 메트릭 업데이트

최종 응답 반환

도구 호출 흐름

에이전트가 도구를 실행해야 함

[도구 호출 전 훅 실행]
    ├→ 훅 1: 도구 허용 여부 확인
    ├→ 훅 2: 입력 검증
    └→ 훅 3: 필요시 승인 요청

훅이 False를 반환하는 경우:
    ├→ 도구 실행 차단
    └→ 오류 메시지 반환

모든 훅이 True/None을 반환하는 경우:
    ├→ 도구 실행 진행
    └→ 결과 생성

[도구 호출 후 훅 실행]
    ├→ 훅 1: 결과 정제
    ├→ 훅 2: 결과 캐싱
    └→ 훅 3: 메트릭 로깅

최종 결과 반환

훅 컨텍스트 객체

LLMCallHookContext

LLM 실행 상태에 대한 액세스를 제공합니다:
class LLMCallHookContext:
    executor: CrewAgentExecutor  # 전체 실행자 액세스
    messages: list               # 변경 가능한 메시지 목록
    agent: Agent                 # 현재 에이전트
    task: Task                   # 현재 작업
    crew: Crew                   # 크루 인스턴스
    llm: BaseLLM                 # LLM 인스턴스
    iterations: int              # 현재 반복 횟수
    response: str | None         # LLM 응답 (후 훅용)

ToolCallHookContext

도구 실행 상태에 대한 액세스를 제공합니다:
class ToolCallHookContext:
    tool_name: str               # 호출되는 도구
    tool_input: dict             # 변경 가능한 입력 매개변수
    tool: CrewStructuredTool     # 도구 인스턴스
    agent: Agent | None          # 실행 중인 에이전트
    task: Task | None            # 현재 작업
    crew: Crew | None            # 크루 인스턴스
    tool_result: str | None      # 도구 결과 (후 훅용)

일반적인 패턴

안전 및 검증

@before_tool_call
def safety_check(context):
    """파괴적인 작업을 차단합니다."""
    dangerous = ['delete_file', 'drop_table', 'system_shutdown']
    if context.tool_name in dangerous:
        print(f"🛑 차단됨: {context.tool_name}")
        return False
    return None

@before_llm_call
def iteration_limit(context):
    """무한 루프를 방지합니다."""
    if context.iterations > 15:
        print("⛔ 최대 반복 횟수 초과")
        return False
    return None

사람의 개입

@before_tool_call
def require_approval(context):
    """민감한 작업에 대한 승인을 요구합니다."""
    sensitive = ['send_email', 'make_payment', 'post_message']

    if context.tool_name in sensitive:
        response = context.request_human_input(
            prompt=f"{context.tool_name} 승인하시겠습니까?",
            default_message="승인하려면 'yes'를 입력하세요:"
        )

        if response.lower() != 'yes':
            return False

    return None

모니터링 및 분석

from collections import defaultdict
import time

metrics = defaultdict(lambda: {'count': 0, 'total_time': 0})

@before_tool_call
def start_timer(context):
    context.tool_input['_start'] = time.time()
    return None

@after_tool_call
def track_metrics(context):
    start = context.tool_input.get('_start', time.time())
    duration = time.time() - start

    metrics[context.tool_name]['count'] += 1
    metrics[context.tool_name]['total_time'] += duration

    return None

훅 관리

모든 훅 지우기

from crewai.hooks import clear_all_global_hooks

# 모든 훅을 한 번에 지웁니다
result = clear_all_global_hooks()
print(f"{result['total']} 훅이 지워졌습니다")

특정 훅 유형 지우기

from crewai.hooks import (
    clear_before_llm_call_hooks,
    clear_after_llm_call_hooks,
    clear_before_tool_call_hooks,
    clear_after_tool_call_hooks
)

# 특정 유형 지우기
llm_before_count = clear_before_llm_call_hooks()
tool_after_count = clear_after_tool_call_hooks()

모범 사례

1. 훅을 집중적으로 유지

각 훅은 단일하고 명확한 책임을 가져야 합니다.

2. 오류를 우아하게 처리

@before_llm_call
def safe_hook(context):
    try:
        if some_condition:
            return False
    except Exception as e:
        print(f"훅 오류: {e}")
        return None  # 오류에도 불구하고 실행 허용

3. 컨텍스트를 제자리에서 수정

# ✅ 올바름 - 제자리에서 수정
@before_llm_call
def add_context(context):
    context.messages.append({"role": "system", "content": "간결하게"})

# ❌ 잘못됨 - 참조를 교체
@before_llm_call
def wrong_approach(context):
    context.messages = [{"role": "system", "content": "간결하게"}]

4. 타입 힌트 사용

from crewai.hooks import LLMCallHookContext, ToolCallHookContext

def my_llm_hook(context: LLMCallHookContext) -> bool | None:
    return None

def my_tool_hook(context: ToolCallHookContext) -> str | None:
    return None

5. 테스트에서 정리

import pytest
from crewai.hooks import clear_all_global_hooks

@pytest.fixture(autouse=True)
def clean_hooks():
    """각 테스트 전에 훅을 재설정합니다."""
    yield
    clear_all_global_hooks()

어떤 훅을 사용해야 할까요

LLM 훅을 사용하는 경우:

  • 반복 제한 구현
  • 프롬프트에 컨텍스트 또는 안전 가이드라인 추가
  • 토큰 사용량 및 비용 추적
  • 응답 정제 또는 변환
  • LLM 호출에 대한 승인 게이트 구현
  • 프롬프트/응답 상호작용 디버깅

도구 훅을 사용하는 경우:

  • 위험하거나 파괴적인 작업 차단
  • 실행 전 도구 입력 검증
  • 민감한 작업에 대한 승인 게이트 구현
  • 도구 결과 캐싱
  • 도구 사용 및 성능 추적
  • 도구 출력 정제
  • 도구 호출 속도 제한

둘 다 사용하는 경우:

모든 에이전트 작업을 모니터링해야 하는 포괄적인 관찰성, 안전 또는 승인 시스템을 구축하는 경우.

관련 문서

결론

실행 훅은 에이전트 런타임 동작에 대한 강력한 제어를 제공합니다. 이를 사용하여 안전 가드레일, 승인 워크플로우, 포괄적인 모니터링 및 사용자 정의 비즈니스 로직을 구현하세요. 적절한 오류 처리, 타입 안전성 및 성능 고려사항과 결합하면, 훅을 통해 프로덕션 준비가 된 안전하고 관찰 가능한 에이전트 시스템을 구축할 수 있습니다.