LLM 호출 훅

def before_hook(context: LLMCallHookContext) -> bool | None:
    # 실행을 차단하려면 False 반환
    # 실행을 허용하려면 True 또는 None 반환
    ...

def after_hook(context: LLMCallHookContext) -> str | None:
    # 수정된 응답 문자열 반환
    # 원본 응답을 유지하려면 None 반환
    ...

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

# ✅ 올바름 - 제자리에서 수정
def add_context(context: LLMCallHookContext) -> None:
    context.messages.append({"role": "system", "content": "간결하게 작성하세요"})

# ❌ 잘못됨 - 리스트 참조를 교체
def wrong_approach(context: LLMCallHookContext) -> None:
    context.messages = [{"role": "system", "content": "간결하게 작성하세요"}]

from crewai.hooks import before_llm_call, after_llm_call

@before_llm_call
def validate_iteration_count(context):
    """반복 횟수를 검증합니다."""
    if context.iterations > 10:
        print("⚠️ 최대 반복 횟수 초과")
        return False  # 실행 차단
    return None

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

from crewai import CrewBase
from crewai.project import crew
from crewai.hooks import before_llm_call_crew, after_llm_call_crew

@CrewBase
class MyProjCrew:
    @before_llm_call_crew
    def validate_inputs(self, context):
        # 이 크루에만 적용됩니다
        if context.iterations == 0:
            print(f"작업 시작: {context.task.description}")
        return None
    
    @after_llm_call_crew
    def log_responses(self, context):
        # 크루별 응답 로깅
        print(f"응답 길이: {len(context.response)}")
        return None
    
    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True
        )

@before_llm_call
def limit_iterations(context: LLMCallHookContext) -> bool | None:
    """무한 루프를 방지하기 위해 반복을 제한합니다."""
    max_iterations = 15
    if context.iterations > max_iterations:
        print(f"⛔ 차단됨: {max_iterations}회 반복 초과")
        return False  # 실행 차단
    return None

@before_llm_call
def require_approval(context: LLMCallHookContext) -> bool | None:
    """특정 반복 후 승인을 요구합니다."""
    if context.iterations > 5:
        response = context.request_human_input(
            prompt=f"반복 {context.iterations}: LLM 호출을 승인하시겠습니까?",
            default_message="승인하려면 Enter를 누르고, 차단하려면 'no'를 입력하세요:"
        )
        if response.lower() == "no":
            print("🚫 사용자에 의해 LLM 호출이 차단되었습니다")
            return False
    return None

@before_llm_call
def add_guardrails(context: LLMCallHookContext) -> None:
    """모든 LLM 호출에 안전 가이드라인을 추가합니다."""
    context.messages.append({
        "role": "system",
        "content": "응답이 사실에 기반하고 가능한 경우 출처를 인용하도록 하세요."
    })
    return None

@after_llm_call
def sanitize_sensitive_data(context: LLMCallHookContext) -> str | None:
    """민감한 데이터 패턴을 제거합니다."""
    if not context.response:
        return None
    
    import re
    sanitized = context.response
    sanitized = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[주민번호-수정됨]', sanitized)
    sanitized = re.sub(r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b', '[카드번호-수정됨]', sanitized)
    
    return sanitized

import tiktoken

@before_llm_call
def track_token_usage(context: LLMCallHookContext) -> None:
    """입력 토큰을 추적합니다."""
    encoding = tiktoken.get_encoding("cl100k_base")
    total_tokens = sum(
        len(encoding.encode(msg.get("content", ""))) 
        for msg in context.messages
    )
    print(f"📊 입력 토큰: ~{total_tokens}")
    return None

@after_llm_call
def track_response_tokens(context: LLMCallHookContext) -> None:
    """응답 토큰을 추적합니다."""
    if context.response:
        encoding = tiktoken.get_encoding("cl100k_base")
        tokens = len(encoding.encode(context.response))
        print(f"📊 응답 토큰: ~{tokens}")
    return None

@before_llm_call
def debug_request(context: LLMCallHookContext) -> None:
    """LLM 요청을 디버그합니다."""
    print(f"""
    🔍 LLM 호출 디버그:
    - 에이전트: {context.agent.role}
    - 작업: {context.task.description[:50]}...
    - 반복: {context.iterations}
    - 메시지 수: {len(context.messages)}
    - 마지막 메시지: {context.messages[-1] if context.messages else 'None'}
    """)
    return None

@after_llm_call
def debug_response(context: LLMCallHookContext) -> None:
    """LLM 응답을 디버그합니다."""
    if context.response:
        print(f"✅ 응답 미리보기: {context.response[:100]}...")
    return None

from crewai.hooks import (
    unregister_before_llm_call_hook,
    unregister_after_llm_call_hook
)

# 특정 훅 등록 해제
def my_hook(context):
    ...

register_before_llm_call_hook(my_hook)
# 나중에...
unregister_before_llm_call_hook(my_hook)  # 찾으면 True 반환

from crewai.hooks import (
    clear_before_llm_call_hooks,
    clear_after_llm_call_hooks,
    clear_all_llm_call_hooks
)

# 특정 훅 타입 지우기
count = clear_before_llm_call_hooks()
print(f"{count}개의 전(before) 훅이 지워졌습니다")

# 모든 LLM 훅 지우기
before_count, after_count = clear_all_llm_call_hooks()
print(f"{before_count}개의 전(before) 훅과 {after_count}개의 후(after) 훅이 지워졌습니다")

@before_llm_call
def conditional_blocking(context: LLMCallHookContext) -> bool | None:
    """특정 조건에서만 차단합니다."""
    # 특정 에이전트에 대해서만 차단
    if context.agent.role == "researcher" and context.iterations > 10:
        return False
    
    # 특정 작업에 대해서만 차단
    if "민감한" in context.task.description.lower() and context.iterations > 5:
        return False
    
    return None

@before_llm_call
def adaptive_prompting(context: LLMCallHookContext) -> None:
    """반복에 따라 다른 컨텍스트를 추가합니다."""
    if context.iterations == 0:
        context.messages.append({
            "role": "system",
            "content": "높은 수준의 개요부터 시작하세요."
        })
    elif context.iterations > 3:
        context.messages.append({
            "role": "system",
            "content": "구체적인 세부사항에 집중하고 예제를 제공하세요."
        })
    return None

# 여러 훅은 등록 순서대로 실행됩니다

@before_llm_call
def first_hook(context):
    print("1. 첫 번째 훅 실행됨")
    return None

@before_llm_call
def second_hook(context):
    print("2. 두 번째 훅 실행됨")
    return None

@before_llm_call
def blocking_hook(context):
    if context.iterations > 10:
        print("3. 차단 훅 - 실행 중지")
        return False  # 후속 훅은 실행되지 않습니다
    print("3. 차단 훅 - 실행 허용")
    return None

@before_llm_call
def safe_hook(context: LLMCallHookContext) -> bool | None:
    try:
        # 훅 로직
        if some_condition:
            return False
    except Exception as e:
        print(f"⚠️ 훅 오류: {e}")
        # 결정: 오류 발생 시 허용 또는 차단
        return None  # 오류에도 불구하고 실행 허용

from crewai.hooks import LLMCallHookContext, BeforeLLMCallHookType, AfterLLMCallHookType

# 명시적 타입 주석
def my_before_hook(context: LLMCallHookContext) -> bool | None:
    return None

def my_after_hook(context: LLMCallHookContext) -> str | None:
    return None

# 타입 안전 등록
register_before_llm_call_hook(my_before_hook)
register_after_llm_call_hook(my_after_hook)