执行钩子（Execution Hooks）概述

执行钩子（Execution Hooks）为您提供了对 CrewAI 智能体运行时行为的细粒度控制。与在团队（Crew）执行前后运行的启动钩子（kickoff hooks）不同，执行钩子会拦截智能体执行过程中的特定操作，从而允许您修改行为、实施安全检查并添加全面的监控。

执行钩子的类型

CrewAI 提供两大类执行钩子

1. LLM 调用钩子

控制和监控语言模型交互

LLM 调用前：修改提示词、验证输入、实现批准门控
LLM 调用后：转换响应、清理输出、更新对话历史

用例

迭代限制
成本跟踪与 Token 使用量监控
响应清理与内容过滤
针对 LLM 调用的“人在回路”（Human-in-the-loop）审批
添加安全准则或上下文
调试日志与请求/响应检查

查看 LLM 钩子文档 →

2. 工具调用钩子

控制和监控工具执行

工具调用前：修改输入、验证参数、拦截危险操作
工具调用后：转换结果、清理输出、记录执行详情

用例

针对破坏性操作的安全护栏
敏感操作的人工审批
输入验证与清理
结果缓存与速率限制
工具使用分析
调试日志与监控

查看工具钩子文档 →

钩子注册方法

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

最整洁、最符合 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）作用域钩子

仅将钩子应用于特定的团队实例

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
        )

钩子执行流程

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

钩子上下文对象

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      # Tool result (after hooks)

常见模式

安全与验证

@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

人在回路（Human-in-the-Loop）

@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

钩子管理

清除所有钩子

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)}

清除特定类型的钩子

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()

注销单个钩子

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. 保持钩子职责单一

每个钩子应只有一个明确的职责

# ✅ 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. 使用类型提示（Type Hints）

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()

何时使用哪种钩子

在以下情况使用 LLM 钩子

实施迭代限制
向提示词添加上下文或安全准则
跟踪 Token 使用量和成本
清理或转换响应
为 LLM 调用实施批准门控
调试提示词/响应交互

在以下情况使用工具钩子

拦截危险或破坏性操作
在执行前验证工具输入
为敏感操作实施批准门控
缓存工具结果
跟踪工具使用情况与性能
清理工具输出
对工具调用进行速率限制

在以下情况两者皆用

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

替代注册方法

程序化注册（高级）

用于动态钩子注册，或当您需要通过编程方式注册钩子时

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

注意：对于大多数用例，使用装饰器更加简洁且易于维护。

性能注意事项

保持钩子高效：钩子在每次调用时都会执行 - 避免繁重的计算
尽量缓存：存储昂贵的验证或查找结果
有选择地使用：当不需要全局钩子时，使用团队作用域钩子
监控钩子开销：在生产环境中分析钩子执行时间
延迟导入：仅在需要时导入大型依赖项

调试钩子

启用调试日志

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

钩子执行顺序

钩子按注册顺序执行。如果前置钩子返回 False，后续钩子将不会执行

# 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 调用钩子 → - 详细的 LLM 钩子文档
工具调用钩子 → - 详细的工具钩子文档
启动前后钩子 → - 团队生命周期钩子
人在回路 → - 人工输入模式

结论

执行钩子为智能体运行时行为提供了强大的控制能力。利用它们来实施安全护栏、审批工作流、全面监控和自定义业务逻辑。结合正确的错误处理、类型安全和性能考量，钩子能够助力构建生产就绪、安全且可观测的智能体系统。

文档索引

​执行钩子的类型

​1. LLM 调用钩子

​2. 工具调用钩子

​钩子注册方法

​1. 基于装饰器的钩子（推荐）

​2. 团队（Crew）作用域钩子

​钩子执行流程

​LLM 调用流程

​工具调用流程

​钩子上下文对象

​LLMCallHookContext

​ToolCallHookContext

​常见模式

​安全与验证

​人在回路（Human-in-the-Loop）

​监控与分析

​响应清理

​钩子管理

​清除所有钩子

​清除特定类型的钩子

​注销单个钩子

​最佳实践

​1. 保持钩子职责单一

​2. 优雅地处理错误

​3. 原地修改上下文

​4. 使用类型提示（Type Hints）

​5. 在测试中进行清理

​何时使用哪种钩子

​在以下情况使用 LLM 钩子

​在以下情况使用工具钩子

​在以下情况两者皆用

​替代注册方法

​程序化注册（高级）

​性能注意事项

​调试钩子

​启用调试日志

​钩子执行顺序

​相关文档

​结论