执行 Hook 概览

执行 Hook 能让你对 CrewAI 智能体在运行时的行为进行细粒度控制。与在 crew 执行前后运行的 kickoff hook 不同，execution hook 会在智能体执行期间拦截特定操作，从而允许你修改行为、加入安全检查，以及添加全面监控。

执行 Hook 的类型

CrewAI 提供两大类执行 Hook：

1. LLM 调用 Hook

控制并监控语言模型交互：

• LLM 调用前：修改提示词、校验输入、实现审批闸门
• LLM 调用后：转换响应、净化输出、更新对话历史

使用场景：

• 限制迭代次数
• 成本追踪与 token 用量监控
• 响应净化与内容过滤
• 对 LLM 调用进行人类在环审批
• 添加安全指南或上下文
• 调试日志和请求/响应检查

查看 LLM Hooks 文档 →

2. 工具调用 Hook

控制并监控工具执行：

• 工具调用前：修改输入、校验参数、阻止危险操作
• 工具调用后：转换结果、净化输出、记录执行细节

使用场景：

• 针对破坏性操作的安全护栏
• 对敏感动作进行人工审批
• 输入校验与净化
• 结果缓存与速率限制
• 工具使用分析
• 调试日志和监控

查看 Tool Hooks 文档 →

Hook 注册方式

1. 基于装饰器的 Hook（推荐）

这是注册 Hook 最简洁、最符合 Python 风格的方式：

from crewai.hooks import before_llm_call, after_llm_call, before_tool_call, after_tool_call

@before_llm_call
def limit_iterations(context):
    """Prevent infinite loops by limiting iterations."""
    if context.iterations > 10:
        return False  # Block execution
    return None

@after_llm_call
def sanitize_response(context):
    """Remove sensitive data from LLM responses."""
    if "API_KEY" in context.response:
        return context.response.replace("API_KEY", "[REDACTED]")
    return None

@before_tool_call
def block_dangerous_tools(context):
    """Block destructive operations."""
    if context.tool_name == "delete_database":
        return False  # Block execution
    return None

@after_tool_call
def log_tool_result(context):
    """Log tool execution."""
    print(f"Tool {context.tool_name} completed")
    return None

2. Crew 作用域 Hook

只把 Hook 应用到特定 crew 实例：

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):
        # Only applies to this crew
        print(f"LLM call in {self.__class__.__name__}")
        return None

    @after_tool_call_crew
    def log_results(self, context):
        # Crew-specific logging
        print(f"Tool result: {context.tool_result[:50]}...")
        return None

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

Hook 执行流程

LLM 调用流程

Agent needs to call LLM
    ↓
[Before LLM Call Hooks Execute]
    ├→ Hook 1: Validate iteration count
    ├→ Hook 2: Add safety context
    └→ Hook 3: Log request
    ↓
If any hook returns False:
    ├→ Block LLM call
    └→ Raise ValueError
    ↓
If all hooks return True/None:
    ├→ LLM call proceeds
    └→ Response generated
    ↓
[After LLM Call Hooks Execute]
    ├→ Hook 1: Sanitize response
    ├→ Hook 2: Log response
    └→ Hook 3: Update metrics
    ↓
Final response returned

工具调用流程

Agent needs to execute tool
    ↓
[Before Tool Call Hooks Execute]
    ├→ Hook 1: Check if tool is allowed
    ├→ Hook 2: Validate inputs
    └→ Hook 3: Request approval if needed
    ↓
If any hook returns False:
    ├→ Block tool execution
    └→ Return error message
    ↓
If all hooks return True/None:
    ├→ Tool execution proceeds
    └→ Result generated
    ↓
[After Tool Call Hooks Execute]
    ├→ Hook 1: Sanitize result
    ├→ Hook 2: Cache result
    └→ Hook 3: Log metrics
    ↓
Final result returned

Hook 上下文对象

LLMCallHookContext

提供对 LLM 执行状态的访问：

class LLMCallHookContext:
    executor: CrewAgentExecutor  # Full executor access
    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
    response: str | None         # LLM response (after hooks)

ToolCallHookContext

提供对工具执行状态的访问：

class ToolCallHookContext:
    tool_name: str               # Tool being called
    tool_input: dict             # Mutable input parameters
    tool: CrewStructuredTool     # Tool instance
    agent: Agent | None          # Agent executing
    task: Task | None            # Current task
    crew: Crew | None            # Crew instance
    tool_result: str | None      # Agent-facing result string (after hooks)
    raw_tool_result: Any | None  # Raw Python result (after hooks)

对于带类型的工具输出，tool_result 是智能体看到的字符串。默认情况下这是 JSON。如果工具使用了自定义格式，也可能是 Markdown 或其他字符串。raw_tool_result 则是工具返回的原始 Python 值。

常见模式

安全与校验

@before_tool_call
def safety_check(context):
    """Block destructive operations."""
    dangerous = ['delete_file', 'drop_table', 'system_shutdown']
    if context.tool_name in dangerous:
        print(f"🛑 Blocked: {context.tool_name}")
        return False
    return None

@before_llm_call
def iteration_limit(context):
    """Prevent infinite loops."""
    if context.iterations > 15:
        print("⛔ Maximum iterations exceeded")
        return False
    return None

人在环中

@before_tool_call
def require_approval(context):
    """Require approval for sensitive operations."""
    sensitive = ['send_email', 'make_payment', 'post_message']

    if context.tool_name in sensitive:
        response = context.request_human_input(
            prompt=f"Approve {context.tool_name}?",
            default_message="Type 'yes' to approve:"
        )

        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

# View metrics
def print_metrics():
    for tool, data in metrics.items():
        avg = data['total_time'] / data['count']
        print(f"{tool}: {data['count']} calls, {avg:.2f}s avg")

响应净化

import re

@after_llm_call
def sanitize_llm_response(context):
    """Remove sensitive data from LLM responses."""
    if not context.response:
        return None

    result = context.response
    result = re.sub(r'(api[_-]?key)["\']?\s*[:=]\s*["\']?[\w-]+',
                   r'\1: [REDACTED]', result, flags=re.IGNORECASE)
    return result

@after_tool_call
def sanitize_tool_result(context):
    """Remove sensitive data from tool results."""
    if not context.tool_result:
        return None

    result = context.tool_result
    result = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
                   '[EMAIL-REDACTED]', result)
    return result

Hook 管理

清除全部 Hook

from crewai.hooks import clear_all_global_hooks

# Clear all hooks at once
result = clear_all_global_hooks()
print(f"Cleared {result['total']} hooks")
# Output: {'llm_hooks': (2, 1), 'tool_hooks': (1, 2), 'total': (3, 3)}

清除特定类型 Hook

from crewai.hooks import (
    clear_before_llm_call_hooks,
    clear_after_llm_call_hooks,
    clear_before_tool_call_hooks,
    clear_after_tool_call_hooks
)

# Clear specific types
llm_before_count = clear_before_llm_call_hooks()
tool_after_count = clear_after_tool_call_hooks()

取消注册单个 Hook

from crewai.hooks import (
    unregister_before_llm_call_hook,
    unregister_after_tool_call_hook
)

def my_hook(context):
    ...

# Register
register_before_llm_call_hook(my_hook)

# Later, unregister
success = unregister_before_llm_call_hook(my_hook)
print(f"Unregistered: {success}")

最佳实践

1. 保持 Hook 聚焦

每个 Hook 都应该只承担一个清晰职责：

# ✅ Good - focused responsibility
@before_tool_call
def validate_file_path(context):
    if context.tool_name == 'read_file':
        if '..' in context.tool_input.get('path', ''):
            return False
    return None

# ❌ Bad - too many responsibilities
@before_tool_call
def do_everything(context):
    # Validation + logging + metrics + approval...
    ...

2. 优雅处理错误

@before_llm_call
def safe_hook(context):
    try:
        # Your logic
        if some_condition:
            return False
    except Exception as e:
        print(f"Hook error: {e}")
        return None  # Allow execution despite error

3. 原地修改上下文

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

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

4. 使用类型提示

from crewai.hooks import LLMCallHookContext, ToolCallHookContext

def my_llm_hook(context: LLMCallHookContext) -> bool | None:
    # IDE autocomplete and type checking
    return None

def my_tool_hook(context: ToolCallHookContext) -> str | None:
    return None

5. 在测试中清理

import pytest
from crewai.hooks import clear_all_global_hooks

@pytest.fixture(autouse=True)
def clean_hooks():
    """Reset hooks before each test."""
    yield
    clear_all_global_hooks()

何时使用哪种 Hook

使用 LLM Hook 的场景：

• 实现迭代限制
• 向提示词添加上下文或安全指南
• 追踪 token 用量和成本
• 净化或转换响应
• 为 LLM 调用实现审批闸门
• 调试提示词/响应交互

使用工具 Hook 的场景：

• 阻止危险或破坏性操作
• 在执行前校验工具输入
• 为敏感操作实现审批闸门
• 缓存工具结果
• 追踪工具使用和性能
• 净化工具输出
• 对工具调用进行速率限制

同时使用两者的场景：

构建需要监控所有智能体操作的综合可观测性、安全或审批系统。

其他注册方式

程序化注册（进阶）

对于动态 Hook 注册，或者当你需要通过代码注册 Hook 时：

from crewai.hooks import (
    register_before_llm_call_hook,
    register_after_tool_call_hook
)

def my_hook(context):
    return None

# Register programmatically
register_before_llm_call_hook(my_hook)

# Useful for:
# - Loading hooks from configuration
# - Conditional hook registration
# - Plugin systems

注意： 对大多数场景来说，装饰器更简洁，也更易维护。

性能考量

1. 保持 Hook 轻量：Hook 会在每次调用时执行，避免重计算
2. 能缓存就缓存：存储代价高的校验或查找结果
3. 有选择地使用：当不需要全局 Hook 时，使用 crew 作用域 Hook
4. 监控开销：在生产环境中分析 Hook 执行时间
5. 延迟导入：只在需要时导入重量级依赖

Hook 调试

启用调试日志

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

@before_llm_call
def debug_hook(context):
    logger.debug(f"LLM call: {context.agent.role}, iteration {context.iterations}")
    return None

Hook 执行顺序

Hook 会按注册顺序执行。如果某个 before hook 返回 False，后续 Hook 就不会继续执行：

# Register order matters!
register_before_tool_call_hook(hook1)  # Executes first
register_before_tool_call_hook(hook2)  # Executes second
register_before_tool_call_hook(hook3)  # Executes third

# If hook2 returns False:
# - hook1 executed
# - hook2 executed and returned False
# - hook3 NOT executed
# - Tool call blocked

相关文档

• LLM Call Hooks → - 详细的 LLM hook 文档
• Tool Call Hooks → - 详细的工具 hook 文档
• Before and After Kickoff Hooks → - crew 生命周期 hook
• Human-in-the-Loop → - 人类输入模式

结论

执行 Hook 为智能体运行时行为提供了强大的控制能力。你可以用它们实现安全护栏、审批流程、全面监控和自定义业务逻辑。结合适当的错误处理、类型安全和性能考量，Hook 能帮助你构建适合生产环境的安全、可观测智能体系统。