1. 为什么要在 LangChain Agent 里引入切面钩子

1.1 从一个真实痛点说起

做过 LangChain Agent 项目的人大概率都经历过这样一个阶段：第一版 Demo 跑通很快，工具调用、记忆管理、提示词拼装全都塞在一个AgentExecutor或者自定义 Chain 里，看起来挺清爽。等到业务开始膨胀，问题就来了——日志要加、耗时统计要加、敏感词过滤要加、Token 用量要统计、异常要重试、调用链要追踪。每加一个通用能力，就得回头改一遍核心业务逻辑，改到最后那个run方法里全是横切关注点，业务代码被淹没在try/except、logger.info、time.time()的海洋里。

这就是典型的横切关注点污染问题。在传统后端开发里，我们早就用 AOP（面向切面编程）解决了它——Spring 的@Around、Django 的 middleware、Gin 的 handler chain，本质都是同一套思路：把日志、鉴权、限流、监控这些和业务无关但每个请求都要走一遍的逻辑，抽成独立的切面，挂到执行链的固定位置上。

LangChain 的 Agent 执行链路天然适合做这件事。它内部有明确的回调机制（Callback）、工具调用入口（Tool）、执行器循环（AgentExecutor），这些位置就是天然的“切面挂载点”。把这套思路落地成一套可复用的钩子体系，业务代码就能彻底和通用能力解耦——业务只管“做什么”，钩子负责“怎么记录、怎么保护、怎么统计”。

1.2 切面钩子到底解决什么问题

先把话说直白一点：切面钩子不是银弹，它解决的是**“多个 Agent 共享同一批通用能力”**这个场景。如果你只有一个 Agent、一个工具、一个业务，那直接在函数里写日志也没啥问题。但只要出现下面任意一种情况，切面钩子的价值就立刻显现：

• 你有 3 个以上的 Agent，每个都要加日志、加耗时统计、加异常兜底；
• 你的工具调用需要统一做参数校验、权限判断、结果脱敏；
• 你想在不改业务代码的前提下，动态开启/关闭某个能力（比如灰度上线一个“调用链追踪”）；
• 你希望把 Token 统计、成本核算、限流这些运维级能力从业务里剥离出去。

我自己的项目里，最早是 5 个 Agent 各自维护一套日志和重试逻辑，后来一次需求变更要统一改重试策略，改了 5 个文件还漏了一个，线上直接出问题。从那之后我就把切面钩子这套东西固化下来了，现在新增一个 Agent，通用能力基本是“零代码接入”。

1.3 和中间件、装饰器的区别在哪

很多人会问：这不就是装饰器吗？或者用中间件不就行了？这里得区分清楚。

装饰器是 Python 语言层面的能力，它作用在函数上，适合做单点的、静态的增强。但 Agent 的执行链路是动态的、多阶段的——一次invoke会经历“输入解析 → 提示词组装 → LLM 调用 → 工具选择 → 工具执行 → 结果回填 → 输出解析”这么多环节，装饰器只能包住最外层，包不住中间每个环节。

中间件这个词在不同框架里含义不同。在 Web 框架里它通常指请求-响应链上的处理单元，是线性的。而 Agent 的切面钩子是事件驱动的——在“工具调用前”“工具调用后”“LLM 返回后”这些具体事件点上触发，粒度更细，位置更精准。

所以更准确的说法是：切面钩子是基于事件回调的、可插拔的、按执行阶段挂载的增强机制。LangChain 本身提供的BaseCallbackHandler就是这套机制的雏形，我们要做的是把它工程化、体系化，变成一套真正能复用的钩子框架。

1.4 整体架构设计思路

我采用的方案是五层解耦架构，从外到内依次是：

这个分层的核心思想是：业务层对切面层完全无感知。业务代码里看不到一行日志、一行统计，所有增强都通过事件总线在钩子调度层完成。这样带来的直接好处是，业务代码可以单独测试、单独复用，切面能力可以跨项目迁移。

2. 核心机制拆解：事件模型与钩子调度

2.1 标准事件模型怎么设计

切面钩子能不能复用，关键看事件模型设计得好不好。事件模型是钩子和业务之间的“契约”，契约稳定了，钩子才能跨 Agent 复用。

我定义的事件模型包含这几个核心字段：

from dataclasses import dataclass, field from typing import Any, Dict, Optional from enum import Enum class EventType(Enum): AGENT_START = "agent_start" AGENT_END = "agent_end" LLM_START = "llm_start" LLM_END = "llm_end" TOOL_START = "tool_start" TOOL_END = "tool_end" TOOL_ERROR = "tool_error" CHAIN_ERROR = "chain_error" @dataclass class AgentEvent: event_type: EventType agent_name: str trace_id: str timestamp: float payload: Dict[str, Any] = field(default_factory=dict) error: Optional[Exception] = None metadata: Dict[str, Any] = field(default_factory=dict)

这里有几个设计决策值得说明。

为什么用trace_id而不是session_id？因为一次用户请求可能触发多次 Agent 调用、多次工具调用，trace_id是贯穿整条调用链的唯一标识，而session_id是会话级别的。用trace_id才能把一次请求里所有事件串起来，做全链路追踪。

为什么payload用字典而不是强类型？因为不同事件携带的数据结构差异很大——LLM_START带的是提示词，TOOL_END带的是工具返回结果。用字典保持灵活性，具体解析交给各个钩子自己处理。但要注意，字典的 key 命名要统一规范，我一般约定用input、output、tool_name、tool_args、duration、token_usage这几个标准字段。

metadata和payload的区别是什么？payload是事件本身的数据，metadata是钩子之间传递的上下文。比如一个“计时钩子”在TOOL_START时往metadata里塞一个开始时间，在TOOL_END时取出来算耗时。这样钩子之间可以协作，而不需要共享全局状态。

2.2 钩子调度器怎么实现

调度器是整个体系的心脏，它负责三件事：注册钩子、按优先级排序、在正确的事件点触发。

class HookDispatcher: def __init__(self): self._hooks: Dict[EventType, List[Tuple[int, BaseHook]]] = {} def register(self, event_type: EventType, hook: "BaseHook", priority: int = 100): self._hooks.setdefault(event_type, []).append((priority, hook)) self._hooks[event_type].sort(key=lambda x: x[0]) def dispatch(self, event: AgentEvent): for priority, hook in self._hooks.get(event.event_type, []): if not hook.enabled: continue try: hook.handle(event) except Exception as e: # 钩子异常不能影响主流程 logger.error(f"Hook {hook.name} failed: {e}")

优先级设计：数字越小越先执行。我一般这样分配：

• 0-49：安全类钩子（敏感词过滤、权限校验），必须最先执行；
• 50-99：观测类钩子（日志、追踪），在业务前后记录；
• 100-149：统计类钩子（Token 统计、耗时统计）；
• 150-199：后处理类钩子（结果脱敏、格式化）。

异常隔离：这是很多人会踩的坑。钩子本身出问题，绝对不能把主流程搞挂。所以dispatch里必须用try/except包住每个钩子的执行。我见过有人的日志钩子因为序列化失败抛异常，直接把整个 Agent 调用搞崩了，这种问题在生产环境是致命的。

同步还是异步：默认同步执行，因为大部分钩子很快。但像“上报监控数据”这种 IO 密集的钩子，同步执行会拖慢主流程。我的做法是给钩子加一个async_mode标记，异步钩子丢到线程池里执行，主流程不等待。

2.3 钩子基类与生命周期

所有钩子继承同一个基类，保证接口统一：

class BaseHook: name: str = "base_hook" enabled: bool = True async_mode: bool = False def handle(self, event: AgentEvent): raise NotImplementedError def on_register(self): """钩子注册时的初始化，比如建立连接、加载配置""" pass def on_unregister(self): """钩子卸载时的清理""" pass

on_register和on_unregister这两个生命周期方法很容易被忽略，但在实际项目里很有用。比如一个“数据库审计钩子”需要在注册时建立连接池，在卸载时关闭连接；一个“限流钩子”需要在注册时加载限流配置。没有这两个钩子，资源管理就会散落在各处。

2.4 与 LangChain 原生 Callback 的对接

LangChain 本身有BaseCallbackHandler，提供了on_llm_start、on_tool_start、on_tool_end等回调点。我们的切面钩子体系不需要重新造轮子，而是包装原生 Callback，把原生回调转换成标准事件，再分发给我们的钩子。

from langchain.callbacks.base import BaseCallbackHandler class HookBridgeCallback(BaseCallbackHandler): def __init__(self, dispatcher: HookDispatcher, agent_name: str): self.dispatcher = dispatcher self.agent_name = agent_name self.trace_id = None def on_chain_start(self, serialized, inputs, **kwargs): self.trace_id = kwargs.get("run_id", str(uuid.uuid4())) self.dispatcher.dispatch(AgentEvent( event_type=EventType.AGENT_START, agent_name=self.agent_name, trace_id=self.trace_id, timestamp=time.time(), payload={"input": inputs} )) def on_tool_start(self, serialized, input_str, **kwargs): self.dispatcher.dispatch(AgentEvent( event_type=EventType.TOOL_START, agent_name=self.agent_name, trace_id=self.trace_id, timestamp=time.time(), payload={"tool_name": serialized.get("name"), "tool_args": input_str} ))

这样做的好处是：钩子体系与 LangChain 版本解耦。LangChain 的 Callback 接口偶尔会变，但我们只需要改 Bridge 这一层，所有钩子实现不用动。这是典型的适配器模式应用。

3. 通用能力钩子的实战实现

3.1 日志钩子：结构化日志与调用链追踪

日志钩子是最基础也最常用的。但“打日志”这件事，做得好和做得差差距很大。我见过太多项目日志就是print(f"tool called: {tool_name}")，线上出问题根本查不到上下文。

我的日志钩子输出结构化 JSON，包含trace_id、agent_name、event_type、duration、payload摘要等字段：

import json import logging class StructuredLogHook(BaseHook): name = "structured_log" def __init__(self, logger_name: str = "agent"): self.logger = logging.getLogger(logger_name) self._timers: Dict[str, float] = {} def handle(self, event: AgentEvent): if event.event_type == EventType.TOOL_START: self._timers[f"{event.trace_id}:{event.payload['tool_name']}"] = event.timestamp duration = None if event.event_type == EventType.TOOL_END: key = f"{event.trace_id}:{event.payload['tool_name']}" start = self._timers.pop(key, None) if start: duration = round(event.timestamp - start, 3) log_record = { "trace_id": event.trace_id, "agent": event.agent_name, "event": event.event_type.value, "ts": event.timestamp, "duration": duration, "payload": self._truncate(event.payload) } self.logger.info(json.dumps(log_record, ensure_ascii=False)) def _truncate(self, payload: dict, max_len: int = 500) -> dict: result = {} for k, v in payload.items(): s = str(v) result[k] = s[:max_len] + "..." if len(s) > max_len else s return result

关键点一：日志截断。工具返回结果可能非常长（比如爬虫返回整个网页），直接打进日志会撑爆日志文件。我一般截断到 500 字符，完整内容存到对象存储或数据库，日志里只留摘要和引用 ID。

关键点二：耗时计算用 metadata 还是内部字典。上面代码用的是钩子内部字典，因为同一个钩子实例处理同一批事件，用内部字典更简单。但如果钩子可能被并发调用，就要用event.metadata来传递，避免 key 冲突。我一般建议用metadata，更安全。

关键点三：日志级别。AGENT_START、AGENT_END用 INFO，TOOL_START、TOOL_END用 DEBUG（量大），TOOL_ERROR、CHAIN_ERROR用 ERROR。这样线上可以按级别过滤，避免日志量爆炸。

3.2 耗时统计钩子：P99 监控与慢调用告警

耗时统计钩子关注的是性能。它和日志钩子的区别是：日志钩子记录每一次调用，耗时钩子做聚合统计。

from collections import defaultdict import statistics class LatencyStatsHook(BaseHook): name = "latency_stats" def __init__(self, window_size: int = 1000, slow_threshold: float = 5.0): self.window_size = window_size self.slow_threshold = slow_threshold self._samples: Dict[str, list] = defaultdict(list) self._timers: Dict[str, float] = {} def handle(self, event: AgentEvent): if event.event_type == EventType.TOOL_START: self._timers[event.trace_id] = event.timestamp elif event.event_type == EventType.TOOL_END: start = self._timers.pop(event.trace_id, None) if start is None: return duration = event.timestamp - start tool_name = event.payload.get("tool_name", "unknown") samples = self._samples[tool_name] samples.append(duration) if len(samples) > self.window_size: samples.pop(0) if duration > self.slow_threshold: self._alert_slow_call(tool_name, duration, event.trace_id) def _alert_slow_call(self, tool_name, duration, trace_id): logger.warning(f"SLOW_CALL tool={tool_name} duration={duration:.2f}s trace={trace_id}") def get_stats(self, tool_name: str) -> dict: samples = self._samples.get(tool_name, []) if not samples: return {} sorted_samples = sorted(samples) return { "count": len(samples), "avg": round(statistics.mean(samples), 3), "p50": round(sorted_samples[len(sorted_samples) // 2], 3), "p99": round(sorted_samples[int(len(sorted_samples) * 0.99)], 3), "max": round(max(samples), 3) }

为什么用滑动窗口而不是全量统计：全量统计内存会无限增长，而且历史数据对当前性能判断意义不大。滑动窗口保留最近 N 次调用，既能反映当前状态，又不会内存泄漏。

P99 比平均值更重要：平均值会被大量快调用拉低，掩盖慢调用问题。P99 能反映“最慢的那 1% 有多慢”，这才是用户体验的关键指标。我一般会同时看 P50、P95、P99 三个分位数。

慢调用告警阈值怎么定：没有统一标准，取决于业务。我的经验是：LLM 调用超过 10 秒、工具调用超过 5 秒、整个 Agent 超过 30 秒，就值得关注。阈值要可配置，不同环境（开发/生产）用不同值。

3.3 敏感词过滤钩子：输入输出双向防护

Agent 应用的安全问题越来越受重视。敏感词过滤钩子要在两个位置拦截：用户输入进入 Agent 之前，以及 Agent 输出返回给用户之前。

import re class SensitiveFilterHook(BaseHook): name = "sensitive_filter" def __init__(self, patterns: list, action: str = "mask"): self.patterns = [re.compile(p) for p in patterns] self.action = action # mask / block / log def handle(self, event: AgentEvent): if event.event_type == EventType.AGENT_START: text = str(event.payload.get("input", "")) self._check(text, event, "input") elif event.event_type == EventType.AGENT_END: text = str(event.payload.get("output", "")) self._check(text, event, "output") def _check(self, text: str, event: AgentEvent, direction: str): for pattern in self.patterns: if pattern.search(text): if self.action == "block": raise SensitiveContentError(f"Blocked {direction}: {pattern.pattern}") elif self.action == "mask": event.payload[f"{direction}_masked"] = pattern.sub("***", text) logger.warning(f"Sensitive hit in {direction}, trace={event.trace_id}")

注意点一：过滤位置。输入过滤在AGENT_START，输出过滤在AGENT_END。但工具返回结果也可能包含敏感内容，所以TOOL_END也要过滤。我一般会在三个位置都挂上。

注意点二：block 还是 mask。block直接抛异常中断流程，适合强合规场景；mask替换敏感词后继续，适合宽松场景。生产环境我建议默认mask，同时记录告警，人工复核后再决定是否升级为block。

注意点三：正则性能。敏感词正则如果写得不好，可能成为性能瓶颈。避免用.*这种贪婪匹配，尽量用精确的字符类。如果敏感词库很大（上千条），考虑用 AC 自动机替代正则。

3.4 Token 统计与成本核算钩子

LLM 调用是要花钱的，Token 统计钩子帮你看清钱花在哪了。

class TokenCostHook(BaseHook): name = "token_cost" PRICING = { "gpt-4": {"input": 0.03, "output": 0.06}, "gpt-3.5-turbo": {"input": 0.0015, "output": 0.002}, } def __init__(self): self._usage: Dict[str, dict] = defaultdict(lambda: {"input": 0, "output": 0, "cost": 0.0}) def handle(self, event: AgentEvent): if event.event_type != EventType.LLM_END: return usage = event.payload.get("token_usage", {}) model = event.payload.get("model", "unknown") input_tokens = usage.get("prompt_tokens", 0) output_tokens = usage.get("completion_tokens", 0) pricing = self.PRICING.get(model, {"input": 0, "output": 0}) cost = (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1000 record = self._usage[event.agent_name] record["input"] += input_tokens record["output"] += output_tokens record["cost"] += cost

为什么按 Agent 维度统计：因为不同 Agent 的成本差异很大，按 Agent 拆分才能定位“哪个 Agent 最烧钱”。如果发现某个 Agent 成本异常，可以进一步按trace_id下钻，看是哪个用户、哪类请求导致的。

价格表要可配置：模型价格经常变，硬编码在代码里迟早过时。我一般把价格表放到配置文件或配置中心，支持热更新。

3.5 重试与降级钩子

工具调用失败是常态——网络抖动、第三方 API 限流、超时。重试钩子负责在失败时自动重试，降级钩子负责在重试耗尽后返回兜底结果。

class RetryHook(BaseHook): name = "retry" def __init__(self, max_retries: int = 3, backoff: float = 1.0): self.max_retries = max_retries self.backoff = backoff self._attempts: Dict[str, int] = defaultdict(int) def handle(self, event: AgentEvent): if event.event_type == EventType.TOOL_ERROR: key = f"{event.trace_id}:{event.payload.get('tool_name')}" self._attempts[key] += 1 if self._attempts[key] <= self.max_retries: delay = self.backoff * (2 ** (self._attempts[key] - 1)) event.metadata["retry_after"] = delay event.metadata["should_retry"] = True logger.info(f"Retry {self._attempts[key]}/{self.max_retries} after {delay}s") else: event.metadata["should_retry"] = False logger.error(f"Retry exhausted for {key}")

指数退避：delay = backoff * 2^(attempt-1)，第一次等 1 秒，第二次 2 秒，第三次 4 秒。这样避免在第三方服务已经过载时继续高频重试，加重对方负担。

重试要幂等：不是所有工具都能重试。查询类工具（搜索、读数据库）可以重试，写入类工具（发消息、下单）重试可能导致重复操作。所以重试钩子要能识别工具类型，只对幂等工具重试。我一般给工具打标签idempotent=True/False，重试钩子根据标签决定是否重试。

4. 完整接入流程与代码组织

4.1 项目目录结构

一套可复用的切面钩子体系，目录结构要清晰。我推荐这样组织：

agent_hooks/ ├── core/ │ ├── event.py # 事件模型定义 │ ├── dispatcher.py # 钩子调度器 │ ├── base.py # 钩子基类 │ └── bridge.py # LangChain Callback 适配器 ├── hooks/ │ ├── logging_hook.py │ ├── latency_hook.py │ ├── sensitive_hook.py │ ├── token_hook.py │ └── retry_hook.py ├── config/ │ └── hooks.yaml # 钩子配置 └── factory.py # 钩子工厂，根据配置创建钩子

core目录是框架层，跨项目复用；hooks目录是具体实现，按需增删；config目录让钩子的启停和参数可配置，不用改代码。

4.2 配置文件驱动钩子注册

钩子的注册不应该硬编码在代码里，而是通过配置文件驱动：

hooks: - name: structured_log class: agent_hooks.hooks.logging_hook.StructuredLogHook enabled: true priority: 50 params: logger_name: agent - name: latency_stats class: agent_hooks.hooks.latency_hook.LatencyStatsHook enabled: true priority: 100 params: window_size: 1000 slow_threshold: 5.0 - name: sensitive_filter class: agent_hooks.hooks.sensitive_hook.SensitiveFilterHook enabled: true priority: 10 params: patterns: ["badword1", "badword2"] action: mask

工厂根据配置动态导入类、实例化、注册到调度器：

import importlib import yaml def load_hooks(config_path: str, dispatcher: HookDispatcher): with open(config_path) as f: config = yaml.safe_load(f) for item in config["hooks"]: if not item.get("enabled", True): continue module_path, class_name = item["class"].rsplit(".", 1) module = importlib.import_module(module_path) hook_cls = getattr(module, class_name) hook = hook_cls(**item.get("params", {})) hook.on_register() for event_type in hook.supported_events: dispatcher.register(event_type, hook, item.get("priority", 100))

为什么用配置驱动：因为不同环境需要不同的钩子组合。开发环境可能只开日志，测试环境开日志+统计，生产环境全开。用配置驱动，同一份代码在不同环境加载不同配置，不用改代码。

4.3 接入到 Agent 的完整示例

把钩子体系接入一个 LangChain Agent，完整流程如下：

from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.tools import tool from agent_hooks.core.dispatcher import HookDispatcher from agent_hooks.core.bridge import HookBridgeCallback from agent_hooks.factory import load_hooks # 1. 初始化调度器并加载钩子 dispatcher = HookDispatcher() load_hooks("config/hooks.yaml", dispatcher) # 2. 定义业务工具（纯业务，无任何通用能力代码） @tool def query_order(order_id: str) -> str: """根据订单号查询订单状态""" return order_service.get(order_id) @tool def send_notification(user_id: str, message: str) -> str: """给用户发送通知""" return notification_service.send(user_id, message) # 3. 构建 Agent llm = ChatOpenAI(model="gpt-4") tools = [query_order, send_notification] agent = create_openai_tools_agent(llm, tools, prompt) executor = AgentExecutor(agent=agent, tools=tools) # 4. 挂载钩子桥接器 callback = HookBridgeCallback(dispatcher, agent_name="order_agent") result = executor.invoke( {"input": "帮我查一下订单 12345 的状态"}, config={"callbacks": [callback]} )

注意看业务工具的定义——query_order和send_notification里没有任何日志、统计、重试代码，就是纯粹的业务逻辑。所有通用能力都通过HookBridgeCallback挂载。这就是解耦的价值。

4.4 多 Agent 共享钩子的实践

当你有多个 Agent 时，钩子体系的价值最大化。所有 Agent 共用同一个dispatcher，钩子只注册一次：

# 全局调度器，所有 Agent 共享 dispatcher = HookDispatcher() load_hooks("config/hooks.yaml", dispatcher) # Agent A executor_a = build_agent("order_agent", tools_a) callback_a = HookBridgeCallback(dispatcher, agent_name="order_agent") # Agent B executor_b = build_agent("support_agent", tools_b) callback_b = HookBridgeCallback(dispatcher, agent_name="support_agent")

事件里的agent_name字段区分了不同 Agent，钩子内部可以按 Agent 维度做差异化处理。比如日志钩子可以给不同 Agent 打不同标签，统计钩子可以按 Agent 分别聚合。

5. 常见问题与排查技巧实录

5.1 钩子不生效的排查思路

钩子不生效是最常见的问题，排查按这个顺序走：

我踩过最坑的一次是：钩子注册了，事件也触发了，但钩子没执行。查了半天发现是enabled属性在基类里定义成了类属性，子类实例化时没覆盖，导致所有钩子共享同一个enabled值，一个钩子被禁用，全部都被禁用了。后来改成实例属性就好了。

5.2 钩子性能拖慢主流程

钩子多了之后，主流程可能被拖慢。排查和优化思路：

• 先测量：在dispatch里记录每个钩子的执行耗时，找出最慢的钩子；
• 异步化：IO 密集的钩子（上报监控、写数据库）改成异步执行；
• 采样：日志钩子不需要记录每一次调用，可以按 10% 采样；
• 批量：统计类钩子不要每次事件都上报，攒够一批再上报。

我一般会加一个“钩子耗时监控”，如果某个钩子平均耗时超过 10ms，就考虑优化。超过 100ms 的钩子必须异步化。

5.3 钩子之间的依赖与顺序问题

有些钩子有依赖关系。比如“脱敏钩子”必须在“日志钩子”之前执行，否则日志里会记录未脱敏的敏感数据。这种依赖通过优先级来保证：

• 脱敏钩子 priority=10
• 日志钩子 priority=50

但优先级只能保证顺序，不能保证“前一个钩子成功执行”。如果脱敏钩子失败了，日志钩子还是会记录原始数据。所以关键钩子要加“失败阻断”机制——脱敏失败时，直接中断流程，不继续执行后续钩子。

5.4 生产环境钩子配置管理

生产环境的钩子配置要支持动态调整，不能改一次配置就重启服务。我的做法是：

• 配置存到配置中心（如 Nacos、Apollo），支持热更新；
• 钩子调度器监听配置变更，动态增删钩子；
• 关键钩子（如安全过滤）不允许动态关闭，防止误操作。

另外，钩子的启停要有审计日志——谁在什么时候开启了哪个钩子，都要记录。这在合规场景下很重要。

5.5 常见问题速查表

6. 一些实操心得与扩展方向

6.1 钩子粒度怎么把握

钩子不是越细越好。我见过有人把钩子拆到“每次字符输出”级别，结果性能直接崩了。我的经验是：钩子粒度对齐业务语义。工具调用、LLM 调用、Agent 开始结束，这三个粒度足够了。再细就是过度设计。

6.2 钩子测试怎么做

钩子本身也要测试。我一般写三类测试：

• 单元测试：给定一个事件，验证钩子的输出；
• 集成测试：跑一个真实 Agent，验证钩子被正确触发；
• 性能测试：模拟高并发，验证钩子不成为瓶颈。

钩子的单元测试特别好写，因为它是纯函数式的——输入事件，输出副作用。构造一个AgentEvent对象，调用handle，断言结果即可。

6.3 后续可以扩展的方向

这套体系跑通之后，可以往几个方向扩展：

• 钩子市场：把通用钩子（日志、统计、限流）打包成独立包，跨项目复用；
• 可视化：把事件流可视化，做成调用链追踪面板；
• A/B 测试：通过钩子动态切换提示词或模型，做 A/B 测试；
• 熔断降级：基于统计钩子的数据，自动触发熔断。

我个人在实际操作中的体会是，切面钩子这套东西前期投入不大，但后期收益极高。尤其是当 Agent 数量超过 3 个之后，没有这套体系，通用能力的维护成本会指数级上升。早点把这套基础设施搭好，后面加 Agent 就是“复制粘贴改配置”的事。