실행 훅 개요

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

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 호출 전 훅 실행]
    ├→ 훅 1: 반복 횟수 검증
    ├→ 훅 2: 안전 컨텍스트 추가
    └→ 훅 3: 요청 로깅
    ↓
훅이 False를 반환하는 경우:
    ├→ LLM 호출 차단
    └→ ValueError 발생
    ↓
모든 훅이 True/None을 반환하는 경우:
    ├→ LLM 호출 진행
    └→ 응답 생성
    ↓
[LLM 호출 후 훅 실행]
    ├→ 훅 1: 응답 정제
    ├→ 훅 2: 응답 로깅
    └→ 훅 3: 메트릭 업데이트
    ↓
최종 응답 반환

에이전트가 도구를 실행해야 함
    ↓
[도구 호출 전 훅 실행]
    ├→ 훅 1: 도구 허용 여부 확인
    ├→ 훅 2: 입력 검증
    └→ 훅 3: 필요시 승인 요청
    ↓
훅이 False를 반환하는 경우:
    ├→ 도구 실행 차단
    └→ 오류 메시지 반환
    ↓
모든 훅이 True/None을 반환하는 경우:
    ├→ 도구 실행 진행
    └→ 결과 생성
    ↓
[도구 호출 후 훅 실행]
    ├→ 훅 1: 결과 정제
    ├→ 훅 2: 결과 캐싱
    └→ 훅 3: 메트릭 로깅
    ↓
최종 결과 반환

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

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

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

# ✅ 올바름 - 제자리에서 수정
@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": "간결하게"}]

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

import pytest
from crewai.hooks import clear_all_global_hooks

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