LLM 调用钩子

了解如何使用 LLM 调用钩子来拦截、修改和控制 CrewAI 中的语言模型交互

LLM 调用钩子在智能体执行期间对语言模型交互 提供细粒度控制。这些钩子让你能够拦截 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:
    # Mutate ctx.messages in place, or
    # raise HookAborted(reason, source) to block the call
    ...

@on(InterceptionPoint.POST_MODEL_CALL)
def after_hook(ctx: LLMCallHookContext) -> str | None:
    # Return a string to replace ctx.response
    # Return None to keep the original response
    ...

与边界点和步骤点不同，模型调用点会将内容丰富的 LLMCallHookContext 直接作为钩子参数传入（没有单独的 ctx.payload）：在调用前就地修改 ctx.messages，并在调用后返回一个 字符串以替换响应。

阻塞调用会在执行器内部抛出 ValueError("LLM call blocked by before_llm_call hook")； HookAborted 的原因和来源会记录到 遥测 中。

LLM 钩子上下文

LLMCallHookContext 对象提供对执行状态的全面访问：

class LLMCallHookContext:
    executor: CrewAgentExecutor | LiteAgent | None  # Executor (None for direct LLM calls)
    messages: list               # Mutable message list
    agent: Agent | None          # Current agent (None for direct LLM calls)
    task: Task | None            # Current task (None for direct calls or LiteAgent)
    crew: Crew | None            # Crew instance (None for direct calls or LiteAgent)
    llm: BaseLLM | None          # LLM instance
    iterations: int              # Current iteration count (0 for direct calls)
    response: str | None         # LLM response (POST_MODEL_CALL only)

该上下文还提供 request_human_input(prompt, default_message)，它会暂停实时控制台更新并从终端收集输入——适用于审批关卡。

修改消息

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

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

# ❌ Wrong - replaces list reference and breaks the 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):
        # Only applies to this crew
        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 a specific hook
unregister_hook(InterceptionPoint.PRE_MODEL_CALL, my_hook)

# Clear one point, or everything (e.g. between tests)
clear_hooks(InterceptionPoint.POST_MODEL_CALL)
clear_all_hooks()

# Inspect what's registered
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 注册的钩子。

旧版装饰器

原有的按拦截点装饰器照常工作，并与 @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  # Block execution
    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 —— 等同于抛出 HookAborted，但没有用于遥测的自定义原因或来源。
• 签名 因拦截点而异：before 钩子返回 bool | None，after 钩子返回 str | None。上下文对象仍是同一个 LLMCallHookContext。
• 过滤器和团队作用域 的工作方式相同：@before_llm_call(agents=[...])， 以及将装饰器应用到 @CrewBase 方法上即可将其限定到该团队。

新代码优先使用 @on；已在使用的旧版风格可继续保留——行为上没有差别。

最佳实践

1. 保持钩子专注且快速 —— 它们在每次 LLM 调用时都会运行
2. 就地修改 —— 始终修改 ctx.messages，绝不要替换整个列表
3. 使用类型提示 —— 用 LLMCallHookContext 标注以获得 IDE 支持
4. 显式中止 —— 抛出带有有意义原因和来源的 HookAborted； 其他任何异常都会被吞掉（fail-open）
5. 在测试中清除钩子 —— 在测试运行之间调用 clear_all_hooks()

故障排查

钩子未执行

• 确认钩子在团队执行之前已注册
• 检查是否有更早的钩子中止了执行（后续钩子不会运行）

消息修改未生效

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

响应修改未生效

• 从 POST_MODEL_CALL 钩子中返回修改后的字符串
• 返回 None 会保留原始响应

相关文档

• 执行钩子概览 →
• 工具调用钩子 →
• 执行边界钩子 →
• 步骤钩子 →