LLM 콜 훅
LLM 콜 훅 (LLM Call Hooks)
LLM 콜 훅은 에이전트 실행 중 언어모델 상호작용을 세밀하게 제어할 수 있게 해줘요. LLM 호출을 가로채고, 프롬프트를 수정하고, 응답을 변환하고, 승인 게이트를 구현하고, 커스텀 로깅·모니터링을 추가할 수 있습니다. 이 페이지는 @on 데코레이터 기반의 LLM 콜 훅을 다룹니다.
출처: 공식문서
본문
개요
LLM 훅은 두 인터셉션 지점에서 실행됩니다.
| 지점 | 시점 | 훅이 받는 것 |
|---|---|---|
PRE_MODEL_CALL |
모든 LLM 호출 전 | LLMCallHookContext |
POST_MODEL_CALL |
모든 LLM 호출 후 | LLMCallHookContext(response 설정됨) |
@on 데코레이터로 작성합니다. 레거시 @before_llm_call / @after_llm_call 데코레이터도 그대로 동작하며, 두 스타일 모두 같은 엔진에 등록되어 하나의 순서 체인에서 실행됩니다.
훅 시그니처
from crewai.hooks import on, HookAborted, InterceptionPoint, LLMCallHookContext
@on(InterceptionPoint.PRE_MODEL_CALL)
def before_hook(ctx: LLMCallHookContext) -> None:
# ctx.messages 를 제자리에서 수정하거나,
# raise HookAborted(reason, source) 로 호출을 차단
...
@on(InterceptionPoint.POST_MODEL_CALL)
def after_hook(ctx: LLMCallHookContext) -> str | None:
# 문자열을 반환하면 ctx.response 를 교체
# None 을 반환하면 원래 응답 유지
...
경계(execution boundary)·스텝 지점과 달리, 모델 콜 지점은 풍부한 LLMCallHookContext 를 훅 인자로 직접 받습니다(별도의 ctx.payload 는 없음). 호출 전에 ctx.messages 를 제자리에서 수정하고, 호출 후에는 응답을 교체할 문자열을 반환하면 됩니다.
호출을 차단하면 executor 안에서 ValueError("LLM call blocked by before_llm_call hook") 가 발생합니다. HookAborted 의 reason 과 source 는 텔레메트리에 기록됩니다.
LLM 훅 컨텍스트
LLMCallHookContext 객체는 실행 상태에 대한 포괄적인 접근을 제공합니다.
class LLMCallHookContext:
executor: CrewAgentExecutor | LiteAgent | None # Executor (직접 LLM 호출이면 None)
messages: list # 변경 가능한 메시지 목록
agent: Agent | None # 현재 에이전트 (직접 LLM 호출이면 None)
task: Task | None # 현재 태스크 (직접 호출/LiteAgent면 None)
crew: Crew | None # 크루 인스턴스 (직접 호출/LiteAgent면 None)
llm: BaseLLM | None # LLM 인스턴스
iterations: int # 현재 반복 횟수 (직접 호출이면 0)
response: str | None # LLM 응답 (POST_MODEL_CALL에서만)
컨텍스트는 또한 request_human_input(prompt, default_message) 를 노출하는데, 라이브 콘솔 업데이트를 멈추고 터미널에서 입력을 수집합니다. 승인 게이트에 유용해요.
메시지 수정
중요: 메시지는 항상 제자리에서(in-place) 수정하세요.
# ✅ 올바름 - 제자리 수정
@on(InterceptionPoint.PRE_MODEL_CALL)
def add_context(ctx: LLMCallHookContext) -> None:
ctx.messages.append({"role": "system", "content": "Be concise"})
# ❌ 오류 - 목록 참조를 교체해 executor 를 깨뜨림
@on(InterceptionPoint.PRE_MODEL_CALL)
def wrong_approach(ctx: LLMCallHookContext) -> None:
ctx.messages = [{"role": "system", "content": "Be concise"}]
등록 방법
1. 전역 훅
모든 크루의 모든 LLM 호출에 적용됩니다. agents= 필터로 훅을 특정 에이전트 역할로 제한할 수 있어요.
from crewai.hooks import on, InterceptionPoint
@on(InterceptionPoint.PRE_MODEL_CALL)
def log_llm_call(ctx):
print(f"LLM call by {ctx.agent.role} at iteration {ctx.iterations}")
@on(InterceptionPoint.POST_MODEL_CALL, agents=["Researcher"])
def log_researcher_responses(ctx):
print(f"Response length: {len(ctx.response)}")
2. 크루 범위 훅
@CrewBase 클래스 안의 메서드에 같은 데코레이터를 적용하면 그 크루에만 적용되는 훅이 됩니다.
from crewai.hooks import on, InterceptionPoint
@CrewBase
class MyProjCrew:
@on(InterceptionPoint.PRE_MODEL_CALL)
def validate_inputs(self, ctx):
# 이 크루에만 적용
if ctx.iterations == 0:
print(f"Starting task: {ctx.task.description}")
@crew
def crew(self) -> Crew:
return Crew(agents=self.agents, tasks=self.tasks, process=Process.sequential)
흔한 사용 사례
1. 반복 횟수 제한
@on(InterceptionPoint.PRE_MODEL_CALL)
def limit_iterations(ctx: LLMCallHookContext) -> None:
if ctx.iterations > 15:
raise HookAborted(reason="exceeded 15 iterations", source="loop-guard")
2. 인간 승인 게이트
@on(InterceptionPoint.PRE_MODEL_CALL)
def require_approval(ctx: LLMCallHookContext) -> None:
if ctx.iterations > 5:
response = ctx.request_human_input(
prompt=f"Iteration {ctx.iterations}: Approve LLM call?",
default_message="Press Enter to approve, or type 'no' to block:",
)
if response.lower() == "no":
raise HookAborted(reason="blocked by user", source="approval-gate")
3. 시스템 컨텍스트 추가
@on(InterceptionPoint.PRE_MODEL_CALL)
def add_guardrails(ctx: LLMCallHookContext) -> None:
ctx.messages.append({
"role": "system",
"content": "Ensure responses are factual and cite sources when possible."
})
4. 응답 정제
import re
@on(InterceptionPoint.POST_MODEL_CALL)
def sanitize_sensitive_data(ctx: LLMCallHookContext) -> str | None:
if not ctx.response:
return None
sanitized = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[SSN-REDACTED]', ctx.response)
return re.sub(r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b', '[CARD-REDACTED]', sanitized)
5. 디버그 로깅
@on(InterceptionPoint.PRE_MODEL_CALL)
def debug_request(ctx: LLMCallHookContext) -> None:
print(f"Agent: {ctx.agent.role}, iteration {ctx.iterations}, "
f"{len(ctx.messages)} messages")
@on(InterceptionPoint.POST_MODEL_CALL)
def debug_response(ctx: LLMCallHookContext) -> None:
if ctx.response:
print(f"Response preview: {ctx.response[:100]}...")
훅 관리
from crewai.hooks import (
InterceptionPoint,
clear_all_hooks,
clear_hooks,
get_hooks,
unregister_hook,
)
# 특정 훅 등록 해제
unregister_hook(InterceptionPoint.PRE_MODEL_CALL, my_hook)
# 한 지점만, 또는 전부 클리어 (예: 테스트 사이)
clear_hooks(InterceptionPoint.POST_MODEL_CALL)
clear_all_hooks()
# 등록된 것 확인
print(len(get_hooks(InterceptionPoint.PRE_MODEL_CALL)))
레거시 관리 API(register_before_llm_call_hook, unregister_before_llm_call_hook, clear_before_llm_call_hooks, clear_all_llm_call_hooks, get_before_llm_call_hooks 및 after_ 대응 API)는 같은 기본 레지스트리에서 동작하므로, 어느 API 로든 다른 API 가 등록한 훅을 관리할 수 있습니다.
레거시 데코레이터
원래의 지점별 데코레이터는 그대로 동작하며 @on 훅과 같은 등록 순서 체인에서 실행됩니다.
from crewai.hooks import before_llm_call, after_llm_call
@before_llm_call
def validate_iteration_count(context):
if context.iterations > 10:
return False # 실행 차단
return None
@after_llm_call(agents=["Researcher"])
def sanitize_response(context):
if context.response and "API_KEY" in context.response:
return context.response.replace("API_KEY", "[REDACTED]")
return None
@on 과의 차이점:
- 차단은 before 훅에서
return False로, 커스텀 reason·source 없이HookAborted를 일으키는 것과 동일합니다. - 시그니처는 지점별: before 훅은
bool | None, after 훅은str | None을 반환. 컨텍스트 객체는 같은LLMCallHookContext입니다. - 필터와 크루 범위 지정은 동일하게 동작:
@before_llm_call(agents=[...]), 그리고@CrewBase메서드에 데코레이터를 적용하면 그 크루로 한정됩니다. - 새 코드에는
@on을 선호하고, 이미 쓰이는 곳에서는 레거시 스타일을 유지하세요. 동작상 불이익은 없습니다.
베스트프랙티스
- 훅을 집중적이고 빠르게 유지 — 모든 LLM 호출마다 실행됩니다
- 제자리에서 수정 — 항상
ctx.messages를 변경하고 목록을 교체하지 마세요 - 타입 힌트 사용 — IDE 지원을 위해
LLMCallHookContext로 주석 - 크게 중단 — 의미 있는 reason·source 와 함께
HookAborted를 발생시키세요. 다른 예외는 삼켜집니다(fail-open) - 테스트에서 훅 정리 — 테스트 실행 사이
clear_all_hooks()호출
문제 해결
훅이 실행되지 않을 때
- 크루 실행 전에 훅이 등록됐는지 확인
- 이전 훅이 중단(abort)했는지 확인(이후 훅은 실행되지 않음)
메시지 수정이 반영되지 않을 때
- 제자리 수정 사용:
ctx.messages.append(...) - 목록을 교체하지 마세요:
ctx.messages = []
응답 수정이 동작하지 않을 때
POST_MODEL_CALL훅에서 수정된 문자열을 반환None을 반환하면 원래 응답 유지