1. 为什么要在 LangChain Agent 里引入切面钩子1.1 从一个真实的痛点说起做过 Agent 项目的朋友大概率都遇到过这种场景一开始只是想做一个能查天气、能算数的小助手代码写得很清爽一个AgentExecutor加几个 Tool 就搞定了。可随着业务往上叠日志要记、Token 要统计、敏感词要过滤、超时要兜底、重试要控制、用户行为要埋点……这些需求和查天气本身没有半毛钱关系却硬生生塞进了每一个 Tool 函数里。我见过最夸张的一个项目一个search_tool函数里前前后后塞了 200 多行代码真正干活的只有 5 行剩下全是logger.info、try/except、if user_id in blacklist、redis.incr这类横切逻辑。结果就是加一个新 Tool得把老 Tool 的样板代码复制一遍改一次日志格式得改十几个文件某天产品说所有工具调用都要加审计整个团队加班三天。这就是典型的横切关注点Cross-Cutting Concerns污染业务代码。在传统后端开发里这个问题早就被 AOP面向切面编程解决了——Spring 的Around、Django 的 middleware、Gin 的 handler chain都是同一个思路。但到了 LangChain Agent 这一层很多人反而忘了这套成熟的方法论把代码写回了面条式。切面钩子Hook / Middleware要解决的就是这件事把日志、鉴权、限流、审计、Token 统计、异常兜底这些通用能力从业务 Tool 里抽出来挂到 Agent 执行生命周期的关键节点上做到业务只写业务通用能力一键复用。1.2 切面钩子到底挂在哪些位置LangChain 的 Agent 执行链路本质上是一条输入 → 模型思考 → 决定调工具 → 执行工具 → 结果回灌 → 再思考 → 输出的循环。能挂钩子的位置主要有这么几个钩子位置触发时机典型用途Agent 启动前用户输入进入 Agent 之前输入清洗、敏感词拦截、会话初始化模型调用前每次 LLM 请求发出前Prompt 注入审计、Token 预算检查、缓存命中判断模型调用后LLM 返回响应后Token 统计、响应脱敏、成本记账工具调用前每个 Tool 执行前权限校验、参数校验、限流、审计日志工具调用后Tool 返回结果后结果脱敏、结果缓存、异常包装Agent 结束时整个循环退出后会话归档、指标上报、清理资源这六个位置基本覆盖了 90% 的通用需求。你不需要每个都实现按业务需要挑几个挂上去就行。关键是钩子本身要可插拔——加一个钩子不改业务代码删一个钩子也不影响业务逻辑。1.3 为什么不用装饰器硬怼有人会说Python 装饰器不也能干这事吗log_tool、retry、rate_limit往 Tool 上一贴不也挺优雅能跑但有几个坑第一装饰器是静态的。一旦贴上运行时想动态开关就得在装饰器内部写if config.enable_log钩子逻辑和配置耦合在一起越写越乱。第二装饰器拿不到 Agent 级别的上下文。比如你想在工具调用前知道这是第几轮循环当前会话累计花了多少 Token装饰器里拿不到这些信息只能靠全局变量或者 contextvar 硬传很别扭。第三装饰器顺序敏感。log套retry和retry套log行为完全不一样团队里新人很容易搞错。切面钩子走的是运行时注册 生命周期回调的路子天然支持动态开关、上下文透传、顺序可控比装饰器更适合 Agent 这种多轮循环的场景。2. 切面钩子的核心设计接口、上下文与执行链2.1 钩子接口怎么定设计钩子接口核心就一句话让钩子能拿到它需要的一切同时不强迫它关心不关心的东西。我一般会定义一个基类把六个生命周期方法都留成空实现钩子按需覆写from abc import ABC from typing import Any, Optional class AgentHook(ABC): Agent 切面钩子基类按需覆写 def on_agent_start(self, ctx: HookContext) - None: pass def on_llm_start(self, ctx: HookContext, prompt: str) - Optional[str]: # 返回非 None 可替换 prompt return None def on_llm_end(self, ctx: HookContext, response: Any) - None: pass def on_tool_start(self, ctx: HookContext, tool_name: str, tool_input: dict) - Optional[dict]: # 返回非 None 可替换 tool_input return None def on_tool_end(self, ctx: HookContext, tool_name: str, result: Any) - Optional[Any]: # 返回非 None 可替换 result return None def on_agent_end(self, ctx: HookContext, final_output: Any) - None: pass注意几个设计细节返回值语义on_xxx_start返回None表示我不改你继续返回具体值表示我要替换。这样钩子既能做只读观察日志、统计也能做写操作改写 prompt、拦截参数。不抛异常原则钩子内部异常必须自己吞掉并记录绝不能因为一个日志钩子崩了导致整个 Agent 挂掉。这一点后面会专门讲。同步接口LangChain 的 Agent 有 sync 和 async 两套钩子最好也提供两套或者统一用 async 然后 sync 场景用asyncio.run包一层。我倾向于统一 async因为 Agent 场景里 IO 密集操作太多。2.2 HookContext 里该放什么HookContext是钩子之间传递信息的载体设计得好不好直接决定钩子好不好用。我的经验是放这几类from dataclasses import dataclass, field from typing import Any, Dict, List import time dataclass class HookContext: session_id: str user_id: str turn: int 0 # 当前是第几轮循环 started_at: float field(default_factorytime.time) tool_calls: List[dict] field(default_factorylist) token_usage: Dict[str, int] field(default_factorylambda: {prompt: 0, completion: 0}) extras: Dict[str, Any] field(default_factorydict) # 钩子间自由传值 aborted: bool False # 是否被某个钩子中断 abort_reason: str extras这个字段特别重要。比如限流钩子在on_tool_start里算出了当前 QPS 超了它可以把extras[rate_limited] True塞进去后面的审计钩子读到这个标记就知道该记一条限流日志。钩子之间通过 context 通信而不是互相 import这是解耦的关键。2.3 执行链的顺序与中断钩子注册进来是一个列表执行时按注册顺序依次调用。但有两个问题要处理顺序问题有些钩子必须在前面比如鉴权有些必须在后面比如统计。我的做法是给钩子加一个priority属性注册时按优先级排序数字小的先执行。class AgentHook(ABC): priority: int 100 # 默认中等优先级鉴权钩子设priority10日志钩子设priority50统计钩子设priority90。这样即使注册顺序乱了执行顺序也是对的。中断问题鉴权失败、敏感词命中、Token 超预算这些情况需要中断整个 Agent 执行。我的做法是钩子不直接抛异常而是设置ctx.aborted True和ctx.abort_reason执行链每调用一个钩子后检查一次发现 aborted 就停止后续钩子并向上抛一个统一的AgentAborted异常由 Agent 外层捕获后返回友好提示。这样做的好处是中断逻辑集中在一处不会散落在各个钩子里而且中断前已经执行过的钩子比如日志能正常收尾不会留下半截记录。3. 六个高频钩子的完整实现3.1 日志钩子把每次调用都记清楚日志钩子是最基础的但也是最容易写烂的。我见过太多项目日志里只有tool called四个字出问题时根本查不出东西。一个好的日志钩子应该记录谁user_id、在哪个会话session_id、第几轮turn、调了什么tool_name、参数是什么脱敏后、耗时多少、结果状态。import logging import time logger logging.getLogger(agent.audit) class LoggingHook(AgentHook): priority 50 def on_tool_start(self, ctx, tool_name, tool_input): ctx.extras[f_tool_start_{tool_name}] time.time() logger.info( tool_start session%s user%s turn%d tool%s input%s, ctx.session_id, ctx.user_id, ctx.turn, tool_name, self._mask(tool_input) ) return None def on_tool_end(self, ctx, tool_name, result): start ctx.extras.pop(f_tool_start_{tool_name}, time.time()) cost (time.time() - start) * 1000 logger.info( tool_end session%s tool%s cost_ms%.1f statusok, ctx.session_id, tool_name, cost ) return None def _mask(self, data: dict) - dict: # 手机号、身份证、密码等字段脱敏 sensitive {phone, id_card, password, token} return {k: (*** if k in sensitive else v) for k, v in data.items()}注意日志里千万不要直接打原始参数。我踩过一次坑某次工具参数里带了用户的完整手机号日志被同步到了第三方平台后来做合规审查时被揪出来整改了两周。脱敏函数一定要在日志钩子内部做不要指望调用方自觉。3.2 鉴权钩子把权限校验从业务里拔出来鉴权钩子的逻辑很直接在工具执行前检查当前用户有没有权限调这个工具。权限规则可以来自配置、数据库或者 Redis。class AuthHook(AgentHook): priority 10 # 必须最先执行 def __init__(self, permission_map: dict, checker): # permission_map: {search_tool: [vip, admin], ...} self.permission_map permission_map self.checker checker # 外部注入的权限查询函数 def on_tool_start(self, ctx, tool_name, tool_input): required self.permission_map.get(tool_name) if not required: return None user_roles self.checker(ctx.user_id) if not set(required) set(user_roles): ctx.aborted True ctx.abort_reason fuser {ctx.user_id} has no permission for {tool_name} logger.warning(auth_denied user%s tool%s, ctx.user_id, tool_name) return None这里有个设计取舍权限规则放配置还是放数据库。我的建议是规则结构放配置角色数据放数据库。因为规则结构哪个工具需要哪些角色变更频率低放配置文件方便 review而用户角色是动态的必须实时查。两者通过checker函数解耦测试时 mock 一个 checker 就行。3.3 限流钩子用 Redis 做滑动窗口限流是 Agent 场景里特别需要的——LLM 调用贵工具调用可能打爆下游。用 Redis 做滑动窗口是最常见的方案。import time import redis class RateLimitHook(AgentHook): priority 20 def __init__(self, redis_client: redis.Redis, limit: int 10, window: int 60): self.r redis_client self.limit limit self.window window def on_tool_start(self, ctx, tool_name, tool_input): key frl:{ctx.user_id}:{tool_name} now time.time() pipe self.r.pipeline() pipe.zremrangebyscore(key, 0, now - self.window) pipe.zadd(key, {str(now): now}) pipe.zcard(key) pipe.expire(key, self.window) _, _, count, _ pipe.execute() if count self.limit: ctx.aborted True ctx.abort_reason frate limit exceeded for {tool_name} ctx.extras[rate_limited] True return None提示zadd的 member 用时间戳字符串如果同一毫秒内有多次调用会覆盖导致计数偏少。生产环境建议用f{now}:{uuid4().hex[:6]}保证唯一。这个坑我在压测时才发现单机测试根本复现不出来。3.4 Token 统计钩子算清楚每一分钱Token 统计钩子挂在on_llm_end上从 LLM 响应里读 usage 字段累加到 context。class TokenStatsHook(AgentHook): priority 90 def __init__(self, price_per_1k_prompt: float, price_per_1k_completion: float): self.p_price price_per_1k_prompt self.c_price price_per_1k_completion def on_llm_end(self, ctx, response): usage getattr(response, usage_metadata, None) or {} p usage.get(input_tokens, 0) c usage.get(output_tokens, 0) ctx.token_usage[prompt] p ctx.token_usage[completion] c return None def on_agent_end(self, ctx, final_output): cost (ctx.token_usage[prompt] / 1000 * self.p_price ctx.token_usage[completion] / 1000 * self.c_price) logger.info( agent_cost session%s prompt%d completion%d cost%.4f, ctx.session_id, ctx.token_usage[prompt], ctx.token_usage[completion], cost )不同模型的 usage 字段名不一样OpenAI 是prompt_tokens/completion_tokensAnthropic 是input_tokens/output_tokensLangChain 新版本统一到了usage_metadata。写钩子时最好做一层兼容别硬编码某一个字段名。3.5 敏感词拦截钩子输入输出双向过滤敏感词拦截要挂在两个位置on_agent_start拦用户输入on_tool_end和on_llm_end拦模型输出。用 AC 自动机或者简单的关键词匹配都行量级不大的话直接any(kw in text for kw in keywords)就够。class SensitiveWordHook(AgentHook): priority 15 def __init__(self, keywords: list[str]): self.keywords keywords def on_agent_start(self, ctx): # 这里假设用户输入已经放在 ctx.extras[user_input] text ctx.extras.get(user_input, ) hit [kw for kw in self.keywords if kw in text] if hit: ctx.aborted True ctx.abort_reason fsensitive words detected: {hit} return None def on_llm_end(self, ctx, response): text getattr(response, content, ) or for kw in self.keywords: if kw in text: # 不中断只替换 response.content text.replace(kw, * * len(kw)) break return None输入拦截用中断输出拦截用替换这是两种不同的处理策略。输入是不允许进来输出是不允许出去语义不一样别搞混。3.6 异常兜底钩子让 Agent 优雅降级工具执行失败是常态——下游 API 挂了、超时了、返回格式不对。异常兜底钩子负责把异常包装成模型能理解的错误信息让 Agent 有机会重试或者换工具。class FallbackHook(AgentHook): priority 95 def on_tool_end(self, ctx, tool_name, result): if isinstance(result, Exception): logger.error(tool_error tool%s err%s, tool_name, result) return { error: True, message: f工具 {tool_name} 执行失败{type(result).__name__}, suggestion: 请尝试其他方式或告知用户稍后重试 } return None返回一个结构化的错误对象比直接抛异常好得多——模型看到这个 JSON 会自己决定下一步而不是整个 Agent 崩掉。4. 把钩子挂到 LangChain Agent 上4.1 用回调机制接入LangChain 本身有一套BaseCallbackHandler机制on_tool_start、on_tool_end、on_llm_start、on_llm_end这些回调点都有。最省事的做法是写一个CallbackHandler适配器把 LangChain 的回调转发到我们的钩子链上。from langchain_core.callbacks import BaseCallbackHandler class HookBridge(BaseCallbackHandler): def __init__(self, hooks: list[AgentHook], ctx: HookContext): self.hooks sorted(hooks, keylambda h: h.priority) self.ctx ctx def on_tool_start(self, serialized, input_str, **kwargs): tool_name serialized.get(name, unknown) tool_input kwargs.get(inputs, {}) for h in self.hooks: h.on_tool_start(self.ctx, tool_name, tool_input) if self.ctx.aborted: raise AgentAborted(self.ctx.abort_reason) def on_tool_end(self, output, **kwargs): tool_name kwargs.get(name, unknown) for h in self.hooks: new h.on_tool_end(self.ctx, tool_name, output) if new is not None: output new return output def on_llm_end(self, response, **kwargs): for h in self.hooks: h.on_llm_end(self.ctx, response)然后在构造 Agent 时把HookBridge塞进callbacksbridge HookBridge(hooks[auth_hook, log_hook, rate_hook, stats_hook], ctxctx) executor AgentExecutor(agentagent, toolstools, callbacks[bridge]) result executor.invoke({input: user_input})4.2 中断异常的处理AgentAborted是自定义异常需要在executor.invoke外层捕获class AgentAborted(Exception): pass try: result executor.invoke({input: user_input}) except AgentAborted as e: result {output: f请求被拦截{e}}注意on_tool_start里抛异常LangChain 会把它包一层再往上抛所以捕获时要判断异常链里有没有AgentAborted。稳妥的写法是except Exception as e: if AgentAborted in repr(e): result {output: 请求被拦截} else: raise4.3 钩子的动态开关生产环境经常需要临时关掉某个钩子比如日志量太大、限流误伤。我的做法是给每个钩子加一个enabled标志从配置中心或者 Redis 读class AgentHook(ABC): priority: int 100 name: str base def is_enabled(self) - bool: return True class LoggingHook(AgentHook): name logging def is_enabled(self): return feature_flag.is_on(agent.hook.logging, defaultTrue)执行链里每次调用前检查is_enabled()关掉的钩子直接跳过。这样改配置不用重启服务运维体验好很多。5. 踩过的坑与排查实录5.1 钩子异常导致 Agent 崩溃这是最常见的坑。日志钩子里json.dumps遇到不可序列化的对象抛异常整个 Agent 就挂了。解决办法是在执行链里统一包一层 try/exceptdef safe_call(hook, method, *args, **kwargs): try: return getattr(hook, method)(*args, **kwargs) except Exception as e: logger.exception(hook %s.%s failed: %s, hook.name, method, e) return None钩子永远不能影响主流程这是铁律。宁可少记一条日志也不能让用户请求失败。5.2 上下文丢失多轮对话场景下HookContext需要在多轮之间保持。我一开始每轮都 new 一个 context结果 Token 统计每轮都从零开始成本算出来只有实际的十分之一。后来改成把 context 存在会话级别的字典里用session_id做 key每轮复用。5.3 异步钩子阻塞事件循环钩子里如果做了同步的 Redis 调用或者 HTTP 请求会阻塞整个事件循环。Agent 场景下这个影响特别大因为一次阻塞可能让并发的其他会话全部卡住。解决办法是钩子统一用 asyncRedis 用redis.asyncioHTTP 用httpx.AsyncClient。5.4 常见问题速查表现象可能原因排查方向钩子完全不执行没挂到 callbacks 上 / priority 排序后位置不对打印 hooks 列表确认中断不生效钩子设了 aborted 但没检查执行链每步后检查 ctx.abortedToken 统计偏少context 每轮重建 / usage 字段名不匹配检查 context 生命周期和字段名限流误伤滑动窗口 member 重复 / 时间戳精度不够用 uuid 做 member日志里出现敏感信息脱敏函数漏字段用白名单而非黑名单钩子顺序错乱没按 priority 排序注册时 sorted5.5 几个实操心得第一钩子要能单独测试。每个钩子写单元测试mock 一个 context 传进去验证行为。别等到集成测试才发现问题那时候排查成本高十倍。第二钩子数量控制在 5-8 个。超过这个数执行链的调试难度会指数上升。如果发现钩子太多考虑合并——比如日志和审计可以合成一个。第三给钩子加耗时统计。钩子本身也是代码也会慢。我在safe_call里加了耗时记录发现某个审计钩子因为每次都查数据库单次耗时 200ms直接拖慢了整个 Agent。后来改成异步批量上报问题解决。第四中断原因要透传给用户。用户看到请求被拦截会一脸懵最好带上具体原因比如今日调用次数已达上限请明天再试。当然敏感原因比如鉴权失败要模糊化别暴露内部规则。6. 从切面钩子到通用能力中台6.1 钩子的复用边界切面钩子最大的价值不是少写代码而是把通用能力沉淀成可复用的资产。一个团队里鉴权钩子、限流钩子、日志钩子写一次所有 Agent 项目都能用。新项目立项时业务代码从零写通用能力直接从钩子库里挑上线周期能缩短一半。我现在的做法是维护一个内部钩子库按能力分类安全类鉴权、敏感词、输入清洗、输出脱敏稳定性类限流、熔断、重试、超时兜底可观测类日志、Token 统计、链路追踪、指标上报成本类预算控制、缓存命中、模型降级每个钩子独立成包通过pip install引入配置通过环境变量或者配置文件注入。这样钩子库和业务项目彻底解耦钩子库升级不影响业务代码。6.2 和 LangGraph 的配合如果项目用的是 LangGraph 而不是 AgentExecutor钩子的挂法略有不同。LangGraph 的节点执行可以通过pre_model_hook、post_model_hook这些参数挂或者直接在节点函数里手动调用钩子链。核心思路一样钩子链独立于图结构图只负责编排钩子负责横切。LangGraph 的好处是状态管理更清晰HookContext可以直接作为图状态的一部分流转不用自己维护会话字典。如果你的项目还在 AgentExecutor 阶段但已经有复杂的多轮状态需求可以考虑迁移到 LangGraph钩子代码基本不用改。6.3 一个真实的收益数据最后分享一个我们团队的实际数据。引入切面钩子之前一个中等复杂度的 Agent 项目业务代码里横切逻辑占比大约 40%新增一个 Tool 平均要写 80 行代码其中 60 行是样板。引入钩子之后新增 Tool 平均 25 行全是业务逻辑横切逻辑占比降到 8%。更关键的是变更成本以前改一次日志格式要动 15 个文件现在改 1 个钩子文件以前加一个全局限流要全组评审三天现在注册一个钩子十分钟搞定。这种改一处、全局生效的能力才是切面钩子真正的价值所在。如果你现在手上的 Agent 项目已经开始出现每个 Tool 都长得差不多的迹象别犹豫把切面钩子这套东西搭起来。前期投入大概两三天后面省下的时间是以月计的。