钩子的"形状"用 Protocol 定义（hooks/types.py:17）：

# hooks/types.py:17
ContextT = TypeVar("ContextT", contravariant=True)
ReturnT  = TypeVar("ReturnT", covariant=True)

@runtime_checkable
class Hook(Protocol, Generic[ContextT, ReturnT]):
    """Generic protocol for hook functions."""
    def __call__(self, context: ContextT) -> ReturnT: ...

# :47 before hooks 都返回 bool|None（False 阻断）
class BeforeLLMCallHook(Hook["LLMCallHookContext", bool | None], Protocol):
    def __call__(self, context: LLMCallHookContext) -> bool | None: ...

# :67 after hooks 都返回 str|None（str 替换）
class AfterLLMCallHook(Hook["LLMCallHookContext", str | None], Protocol):
    def __call__(self, context: LLMCallHookContext) -> str | None: ...

# :128 便捷类型别名
BeforeLLMCallHookType = Hook["LLMCallHookContext", bool | None]
AfterLLMCallHookType  = Hook["LLMCallHookContext", str | None]

钩子能干活，全靠这个上下文对象（hooks/llm_hooks.py:24）：

# hooks/llm_hooks.py:24
class LLMCallHookContext:
    """Context object passed to LLM call hooks.
    messages: Direct reference to messages (mutable list).
        IMPORTANT: Modify messages in-place (append/extend/remove).
        Do NOT replace the list (context.messages = []), as this will break the executor."""
    executor: ...
    messages: list[LLMMessage]     # ★对执行器 messages 的直接引用（不是拷贝）
    agent: Any; task: Any; crew: Any
    llm: BaseLLM | None
    iterations: int                # 当前第几轮
    response: str | None           # 只有 after 钩子才有值

    def __init__(self, executor=None, response=None, messages=None, ...):
        if executor is not None:
            self.executor   = executor
            self.messages   = executor.messages     # 直接指向执行器的那份 messages
            self.llm        = executor.llm
            self.iterations = executor.iterations
            ...
        self.response = response

它还带一个"请求人工输入"的方法，用于审批场景（hooks/llm_hooks.py:109）：

# hooks/llm_hooks.py:109
def request_human_input(self, prompt, default_message="Press Enter to continue..."):
    """暂停实时输出、向用户展示 prompt、等待输入、再恢复。用于审批/调试。"""

四个装饰器由一个工厂统一生产（hooks/decorators.py:18）：

# hooks/decorators.py:18
def _create_hook_decorator(hook_type, register_function, marker_attribute):
    def decorator_factory(func=None, *, tools=None, agents=None):
        if tools:
            tools = [sanitize_tool_name(t) for t in tools]
        def decorator(f):
            setattr(f, marker_attribute, True)          # 打标记（如 is_before_llm_call_hook）
            sig = inspect.signature(f)
            params = list(sig.parameters.keys())
            is_method = len(params) >= 2 and params[0] == "self"   # 判断是不是类方法
            ...
            if not is_method:
                register_function(f)                    # 普通函数→立即注册到全局表
            return f
        if func is None:
            return decorator                            # @before_llm_call(agents=[...]) 带参用法
        return decorator(func)                          # @before_llm_call 裸用法
    return decorator_factory

注册就是往一个模块级列表里 append（hooks/llm_hooks.py:153）：

# hooks/llm_hooks.py:153
_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = []

# :157
def register_before_llm_call_hook(hook):
    ...
    _before_llm_call_hooks.append(hook)    # :190 存进全局表

# :221
def get_before_llm_call_hooks():
    return _before_llm_call_hooks.copy()   # 返回副本，防止外部篡改原表

钩子在执行前被批量调用（utilities/agent_utils.py:1668）：

# utilities/agent_utils.py:1668
def _setup_before_llm_call_hooks(executor_context, printer, verbose=True) -> bool:
    if executor_context and executor_context.before_llm_call_hooks:
        original_messages = executor_context.messages
        hook_context = LLMCallHookContext(executor_context)   # 造上下文（内含 messages 引用）
        try:
            for hook in executor_context.before_llm_call_hooks:
                result = hook(hook_context)                   # 逐个调用
                if result is False:                           # ★任一返回 False → 阻断
                    if verbose:
                        printer.print("LLM call blocked by before_llm_call hook", color="yellow")
                    return False                              # 告诉调用方：别问 LLM 了
        except Exception as e:
            printer.print(f"Error in before_llm_call hook: {e}", color="yellow")  # 钩子出错不崩主流程
        if not isinstance(executor_context.messages, list):   # ★防御 L03 的坑
            executor_context.messages = original_messages if isinstance(original_messages, list) else []
    return True   # 放行

而执行循环里是这样用它的（utilities/agent_utils.py:422）：

# utilities/agent_utils.py:422
if not _setup_before_llm_call_hooks(executor_context, printer, verbose=verbose):
    ...   # 返回 False → 跳过这次真正的 LLM 调用

@before_llm_call
def gate(ctx):
    if ctx.iterations > 5:
        fb = ctx.request_human_input("已经 5 轮了，继续吗？")
        if fb.strip().lower() == "stop":
            return False        # 阻断：不再问 LLM
    ctx.messages.append({"role":"system","content":"请更简洁"})  # 原地改 prompt

after 钩子的返回值语义不同——返回 str 就替换结果（utilities/agent_utils.py:1724）：

# utilities/agent_utils.py:1724
def _setup_after_llm_call_hooks(executor_context, answer, printer, verbose=True):
    if executor_context and executor_context.after_llm_call_hooks:
        if isinstance(answer, BaseModel):                  # 结构化输出：先转成 json 字符串给钩子
            pydantic_answer = answer
            hook_response = pydantic_answer.model_dump_json()
            original_json = hook_response
        else:
            pydantic_answer = None
            hook_response = str(answer)

        hook_context = LLMCallHookContext(executor_context, response=hook_response)
        try:
            for hook in executor_context.after_llm_call_hooks:
                modified_response = hook(hook_context)
                if modified_response is not None and isinstance(modified_response, str):
                    hook_response = modified_response        # ★返回 str → 用它替换
        except Exception as e:
            printer.print(f"Error in after_llm_call hook: {e}", color="yellow")
        ...
        if pydantic_answer is not None and hook_response != original_json:
            answer = type(pydantic_answer).model_validate_json(hook_response)  # 改完再校验回模型
    ...

带 agents=/tools= 参数时，装饰器额外包一层过滤（hooks/decorators.py:57）：

# hooks/decorators.py:57
if tools or agents:
    @wraps(f)
    def filtered_hook(context):
        if tools and hasattr(context, "tool_name"):
            if context.tool_name not in tools:
                return None                       # 不是我关心的工具 → 直接不干预
        if agents and hasattr(context, "agent"):
            if context.agent and context.agent.role not in agents:
                return None                       # 不是我关心的 agent → 跳过
        return f(context)                         # 命中才真的执行你的钩子
    if not is_method:
        register_function(filtered_hook)
    return f

而写在 @CrewBase 类里的方法钩子，靠标记类的 __get__ 描述符绑到实例（hooks/wrappers.py:29）：

# hooks/wrappers.py:29
class BeforeLLMCallHookMethod:
    is_before_llm_call_hook: bool = True
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return lambda context: self._meth(obj, context)   # 绑定 self → 变成"只吃 context"的钩子