★ FEATURED ARTICLE 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 整体架构设计思路我采用的方案是五层解耦架构从外到内依次是层级名称职责L1接入层统一入口负责 Agent 实例化、钩子注册L2钩子调度层管理钩子的注册、优先级、启停L3切面层具体钩子实现日志、统计、限流、脱敏等L4事件总线层定义标准事件模型负责事件分发L5业务层纯业务逻辑不含任何通用能力代码这个分层的核心思想是业务层对切面层完全无感知。业务代码里看不到一行日志、一行统计所有增强都通过事件总线在钩子调度层完成。这样带来的直接好处是业务代码可以单独测试、单独复用切面能力可以跨项目迁移。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_factorydict) error: Optional[Exception] None metadata: Dict[str, Any] field(default_factorydict)这里有几个设计决策值得说明。为什么用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(keylambda 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(fHook {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): 钩子卸载时的清理 passon_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_typeEventType.AGENT_START, agent_nameself.agent_name, trace_idself.trace_id, timestamptime.time(), payload{input: inputs} )) def on_tool_start(self, serialized, input_str, **kwargs): self.dispatcher.dispatch(AgentEvent( event_typeEventType.TOOL_START, agent_nameself.agent_name, trace_idself.trace_id, timestamptime.time(), payload{tool_name: serialized.get(name), tool_args: input_str} ))这样做的好处是钩子体系与 LangChain 版本解耦。LangChain 的 Callback 接口偶尔会变但我们只需要改 Bridge 这一层所有钩子实现不用动。这是典型的适配器模式应用。3. 通用能力钩子的实战实现3.1 日志钩子结构化日志与调用链追踪日志钩子是最基础也最常用的。但“打日志”这件事做得好和做得差差距很大。我见过太多项目日志就是print(ftool 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_asciiFalse)) 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用 INFOTOOL_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(fSLOW_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(fBlocked {direction}: {pattern.pattern}) elif self.action mask: event.payload[f{direction}_masked] pattern.sub(***, text) logger.warning(fSensitive 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(fRetry {self._attempts[key]}/{self.max_retries} after {delay}s) else: event.metadata[should_retry] False logger.error(fRetry exhausted for {key})指数退避delay backoff * 2^(attempt-1)第一次等 1 秒第二次 2 秒第三次 4 秒。这样避免在第三方服务已经过载时继续高频重试加重对方负担。重试要幂等不是所有工具都能重试。查询类工具搜索、读数据库可以重试写入类工具发消息、下单重试可能导致重复操作。所以重试钩子要能识别工具类型只对幂等工具重试。我一般给工具打标签idempotentTrue/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(modelgpt-4) tools [query_order, send_notification] agent create_openai_tools_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools) # 4. 挂载钩子桥接器 callback HookBridgeCallback(dispatcher, agent_nameorder_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_nameorder_agent) # Agent B executor_b build_agent(support_agent, tools_b) callback_b HookBridgeCallback(dispatcher, agent_namesupport_agent)事件里的agent_name字段区分了不同 Agent钩子内部可以按 Agent 维度做差异化处理。比如日志钩子可以给不同 Agent 打不同标签统计钩子可以按 Agent 分别聚合。5. 常见问题与排查技巧实录5.1 钩子不生效的排查思路钩子不生效是最常见的问题排查按这个顺序走排查点检查方法常见原因钩子是否注册打印 dispatcher._hooks配置未加载、enabledfalse事件是否触发在 dispatch 里打日志Callback 未挂载、事件类型不匹配优先级是否正确检查 priority 值优先级冲突导致被覆盖异常是否被吞检查钩子内部 try/except钩子抛异常被静默捕获我踩过最坑的一次是钩子注册了事件也触发了但钩子没执行。查了半天发现是enabled属性在基类里定义成了类属性子类实例化时没覆盖导致所有钩子共享同一个enabled值一个钩子被禁用全部都被禁用了。后来改成实例属性就好了。5.2 钩子性能拖慢主流程钩子多了之后主流程可能被拖慢。排查和优化思路先测量在dispatch里记录每个钩子的执行耗时找出最慢的钩子异步化IO 密集的钩子上报监控、写数据库改成异步执行采样日志钩子不需要记录每一次调用可以按 10% 采样批量统计类钩子不要每次事件都上报攒够一批再上报。我一般会加一个“钩子耗时监控”如果某个钩子平均耗时超过 10ms就考虑优化。超过 100ms 的钩子必须异步化。5.3 钩子之间的依赖与顺序问题有些钩子有依赖关系。比如“脱敏钩子”必须在“日志钩子”之前执行否则日志里会记录未脱敏的敏感数据。这种依赖通过优先级来保证脱敏钩子 priority10日志钩子 priority50但优先级只能保证顺序不能保证“前一个钩子成功执行”。如果脱敏钩子失败了日志钩子还是会记录原始数据。所以关键钩子要加“失败阻断”机制——脱敏失败时直接中断流程不继续执行后续钩子。5.4 生产环境钩子配置管理生产环境的钩子配置要支持动态调整不能改一次配置就重启服务。我的做法是配置存到配置中心如 Nacos、Apollo支持热更新钩子调度器监听配置变更动态增删钩子关键钩子如安全过滤不允许动态关闭防止误操作。另外钩子的启停要有审计日志——谁在什么时候开启了哪个钩子都要记录。这在合规场景下很重要。5.5 常见问题速查表问题现象可能原因解决方法钩子完全不执行未注册或 enabledfalse检查配置和注册逻辑钩子执行但无输出日志级别不对调整日志级别钩子执行顺序错乱优先级配置错误检查 priority 值主流程变慢钩子同步执行耗时高改异步或采样钩子异常导致主流程崩溃未做异常隔离dispatch 加 try/except敏感词漏过滤过滤位置不全输入/输出/工具结果都过滤Token 统计不准事件未覆盖所有 LLM 调用检查 Callback 挂载点6. 一些实操心得与扩展方向6.1 钩子粒度怎么把握钩子不是越细越好。我见过有人把钩子拆到“每次字符输出”级别结果性能直接崩了。我的经验是钩子粒度对齐业务语义。工具调用、LLM 调用、Agent 开始结束这三个粒度足够了。再细就是过度设计。6.2 钩子测试怎么做钩子本身也要测试。我一般写三类测试单元测试给定一个事件验证钩子的输出集成测试跑一个真实 Agent验证钩子被正确触发性能测试模拟高并发验证钩子不成为瓶颈。钩子的单元测试特别好写因为它是纯函数式的——输入事件输出副作用。构造一个AgentEvent对象调用handle断言结果即可。6.3 后续可以扩展的方向这套体系跑通之后可以往几个方向扩展钩子市场把通用钩子日志、统计、限流打包成独立包跨项目复用可视化把事件流可视化做成调用链追踪面板A/B 测试通过钩子动态切换提示词或模型做 A/B 测试熔断降级基于统计钩子的数据自动触发熔断。我个人在实际操作中的体会是切面钩子这套东西前期投入不大但后期收益极高。尤其是当 Agent 数量超过 3 个之后没有这套体系通用能力的维护成本会指数级上升。早点把这套基础设施搭好后面加 Agent 就是“复制粘贴改配置”的事。
阅读完成 · 觉得有帮助？
