LLM 调用 Hook

LLM Call Hooks 能在智能体执行期间对语言模型交互进行细粒度控制。通过这些 Hook，你可以拦截 LLM 调用、修改提示词、转换响应、实现审批闸门，以及添加自定义日志或监控。

概览

LLM Hook 会在两个关键点执行：

• LLM 调用前：修改消息、校验输入或阻止执行
• LLM 调用后：转换响应、净化输出或修改对话历史

Hook 类型

1. LLM 调用前 Hook

在每次 LLM 调用前执行，这类 Hook 可以：

• 检查并修改发送给 LLM 的消息
• 根据条件阻止 LLM 执行
• 实现速率限制或审批闸门
• 添加上下文或系统消息
• 记录请求细节

签名：

def before_hook(context: LLMCallHookContext) -> bool | None:
    # Return False to block execution
    # Return True or None to allow execution
    ...

2. LLM 调用后 Hook

在每次 LLM 调用后执行，这类 Hook 可以：

• 修改或净化 LLM 响应
• 添加元数据或格式化
• 记录响应细节
• 更新对话历史
• 实现内容过滤

签名：

def after_hook(context: LLMCallHookContext) -> str | None:
    # Return modified response string
    # Return None to keep original response
    ...

LLM Hook 上下文

LLMCallHookContext 对象提供对执行状态的完整访问：

class LLMCallHookContext:
    executor: CrewAgentExecutor  # Full executor reference
    messages: list               # Mutable message list
    agent: Agent                 # Current agent
    task: Task                   # Current task
    crew: Crew                   # Crew instance
    llm: BaseLLM                 # LLM instance
    iterations: int              # Current iteration count
    response: str | None         # LLM response (after hooks only)

修改消息

重要： 始终原地修改消息：

# ✅ Correct - modify in-place
def add_context(context: LLMCallHookContext) -> None:
    context.messages.append({"role": "system", "content": "Be concise"})

# ❌ Wrong - replaces list reference
def wrong_approach(context: LLMCallHookContext) -> None:
    context.messages = [{"role": "system", "content": "Be concise"}]

注册方式

1. 全局 Hook 注册

注册适用于所有 crew 的所有 LLM 调用的 Hook：

from crewai.hooks import register_before_llm_call_hook, register_after_llm_call_hook

def log_llm_call(context):
    print(f"LLM call by {context.agent.role} at iteration {context.iterations}")
    return None  # Allow execution

register_before_llm_call_hook(log_llm_call)

2. 基于装饰器的注册

使用装饰器获得更简洁的语法：

from crewai.hooks import before_llm_call, after_llm_call

@before_llm_call
def validate_iteration_count(context):
    if context.iterations > 10:
        print("⚠️ Exceeded maximum iterations")
        return False  # Block execution
    return None

@after_llm_call
def sanitize_response(context):
    if context.response and "API_KEY" in context.response:
        return context.response.replace("API_KEY", "[REDACTED]")
    return None

3. Crew 作用域 Hook

为特定 crew 实例注册 Hook：

@CrewBase
class MyProjCrew:
    @before_llm_call_crew
    def validate_inputs(self, context):
        # Only applies to this crew
        if context.iterations == 0:
            print(f"Starting task: {context.task.description}")
        return None

    @after_llm_call_crew
    def log_responses(self, context):
        # Crew-specific response logging
        print(f"Response length: {len(context.response)}")
        return None

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

常见用途

1. 限制迭代次数

@before_llm_call
def limit_iterations(context: LLMCallHookContext) -> bool | None:
    max_iterations = 15
    if context.iterations > max_iterations:
        print(f"⛔ Blocked: Exceeded {max_iterations} iterations")
        return False  # Block execution
    return None

2. 人工审批闸门

@before_llm_call
def require_approval(context: LLMCallHookContext) -> bool | None:
    if context.iterations > 5:
        response = context.request_human_input(
            prompt=f"Iteration {context.iterations}: Approve LLM call?",
            default_message="Press Enter to approve, or type 'no' to block:"
        )
        if response.lower() == "no":
            print("🚫 LLM call blocked by user")
            return False
    return None

3. 添加系统上下文

@before_llm_call
def add_guardrails(context: LLMCallHookContext) -> None:
    # Add safety guidelines to every LLM call
    context.messages.append({
        "role": "system",
        "content": "Ensure responses are factual and cite sources when possible."
    })
    return None

4. 响应净化

@after_llm_call
def sanitize_sensitive_data(context: LLMCallHookContext) -> str | None:
    if not context.response:
        return None

    # Remove sensitive patterns
    import re
    sanitized = context.response
    sanitized = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[SSN-REDACTED]', sanitized)
    sanitized = re.sub(r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b', '[CARD-REDACTED]', sanitized)

    return sanitized

5. 成本追踪

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"📊 Input tokens: ~{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"📊 Response tokens: ~{tokens}")
    return None

6. 调试日志

@before_llm_call
def debug_request(context: LLMCallHookContext) -> None:
    print(f"""
    🔍 LLM Call Debug:
    - Agent: {context.agent.role}
    - Task: {context.task.description[:50]}...
    - Iteration: {context.iterations}
    - Message Count: {len(context.messages)}
    - Last Message: {context.messages[-1] if context.messages else 'None'}
    """)
    return None

@after_llm_call
def debug_response(context: LLMCallHookContext) -> None:
    if context.response:
        print(f"✅ Response Preview: {context.response[:100]}...")
    return None

Hook 管理

取消注册 Hook

from crewai.hooks import (
    unregister_before_llm_call_hook,
    unregister_after_llm_call_hook
)

# Unregister specific hook
def my_hook(context):
    ...

register_before_llm_call_hook(my_hook)
# Later...
unregister_before_llm_call_hook(my_hook)  # Returns True if found

清除 Hook

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

# Clear specific hook type
count = clear_before_llm_call_hooks()
print(f"Cleared {count} before hooks")

# Clear all LLM hooks
before_count, after_count = clear_all_llm_call_hooks()
print(f"Cleared {before_count} before and {after_count} after hooks")

列出已注册 Hook

from crewai.hooks import (
    get_before_llm_call_hooks,
    get_after_llm_call_hooks
)

# Get current hooks
before_hooks = get_before_llm_call_hooks()
after_hooks = get_after_llm_call_hooks()

print(f"Registered: {len(before_hooks)} before, {len(after_hooks)} after")

高级模式

条件式 Hook 执行

@before_llm_call
def conditional_blocking(context: LLMCallHookContext) -> bool | None:
    # Only block for specific agents
    if context.agent.role == "researcher" and context.iterations > 10:
        return False

    # Only block for specific tasks
    if "sensitive" in context.task.description.lower() and context.iterations > 5:
        return False

    return None

感知上下文的修改

@before_llm_call
def adaptive_prompting(context: LLMCallHookContext) -> None:
    # Add different context based on iteration
    if context.iterations == 0:
        context.messages.append({
            "role": "system",
            "content": "Start with a high-level overview."
        })
    elif context.iterations > 3:
        context.messages.append({
            "role": "system",
            "content": "Focus on specific details and provide examples."
        })
    return None

Hook 链式执行

# Multiple hooks execute in registration order

@before_llm_call
def first_hook(context):
    print("1. First hook executed")
    return None

@before_llm_call
def second_hook(context):
    print("2. Second hook executed")
    return None

@before_llm_call
def blocking_hook(context):
    if context.iterations > 10:
        print("3. Blocking hook - execution stopped")
        return False  # Subsequent hooks won't execute
    print("3. Blocking hook - execution allowed")
    return None

最佳实践

1. 保持 Hook 聚焦：每个 Hook 只应有一个职责
2. 避免重计算：Hook 会在每次 LLM 调用时执行
3. 优雅处理错误：使用 try-except 防止 Hook 失败中断执行
4. 使用类型提示：利用 LLMCallHookContext 获得更好的 IDE 支持
5. 记录 Hook 行为：尤其是阻塞条件
6. 独立测试 Hook：在生产使用前进行单元测试
7. 在测试中清理 Hook：测试运行之间使用 clear_all_llm_call_hooks()
8. 原地修改：始终原地修改 context.messages，不要替换它

错误处理

@before_llm_call
def safe_hook(context: LLMCallHookContext) -> bool | None:
    try:
        # Your hook logic
        if some_condition:
            return False
    except Exception as e:
        print(f"⚠️ Hook error: {e}")
        # Decide: allow or block on error
        return None  # Allow execution despite error

类型安全

from crewai.hooks import LLMCallHookContext, BeforeLLMCallHookType, AfterLLMCallHookType

# Explicit type annotations
def my_before_hook(context: LLMCallHookContext) -> bool | None:
    return None

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

# Type-safe registration
register_before_llm_call_hook(my_before_hook)
register_after_llm_call_hook(my_after_hook)

故障排查

Hook 没有执行

• 确认 Hook 在 crew 执行前已注册
• 检查前一个 Hook 是否返回了 False（会阻止后续 Hook）
• 确认 Hook 签名与预期类型一致

消息修改没有保留

• 使用原地修改：context.messages.append()
• 不要替换列表：context.messages = []

响应修改没有生效

• 从 after hook 返回修改后的字符串
• 返回 None 会保留原始响应

结论

LLM 调用 Hook 为控制和监控 CrewAI 中的语言模型交互提供了强大能力。你可以用它们实现安全护栏、审批闸门、日志记录、成本追踪和响应净化。结合适当的错误处理和类型安全，Hook 能帮助你构建稳健且可直接用于生产的智能体系统。