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 执行链路，本质上是一条"输入 → 模型思考 → 决定调工具 → 执行工具 → 结果回灌 → 再思考 → 输出"的循环。能挂钩子的位置主要有这么几个：

这六个位置基本覆盖了 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_factory=time.time) tool_calls: List[dict] = field(default_factory=list) token_usage: Dict[str, int] = field(default_factory=lambda: {"prompt": 0, "completion": 0}) extras: Dict[str, Any] = field(default_factory=dict) # 钩子间自由传值 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 # 默认中等优先级

鉴权钩子设priority=10，日志钩子设priority=50，统计钩子设priority=90。这样即使注册顺序乱了，执行顺序也是对的。

中断问题：鉴权失败、敏感词命中、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 status=ok", 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 = f"user {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 = f"rl:{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 = f"rate 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_tokens，Anthropic 是input_tokens/output_tokens，LangChain 新版本统一到了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 = f"sensitive 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, key=lambda 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塞进callbacks：

bridge = HookBridge(hooks=[auth_hook, log_hook, rate_hook, stats_hook], ctx=ctx) executor = AgentExecutor(agent=agent, tools=tools, 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: raise

4.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", default=True)

执行链里每次调用前检查is_enabled()，关掉的钩子直接跳过。这样改配置不用重启服务，运维体验好很多。

5. 踩过的坑与排查实录

5.1 钩子异常导致 Agent 崩溃

这是最常见的坑。日志钩子里json.dumps遇到不可序列化的对象抛异常，整个 Agent 就挂了。解决办法是在执行链里统一包一层 try/except：

def 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 场景下这个影响特别大，因为一次阻塞可能让并发的其他会话全部卡住。解决办法是钩子统一用 async，Redis 用redis.asyncio，HTTP 用httpx.AsyncClient。

5.4 常见问题速查表

5.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 都长得差不多"的迹象，别犹豫，把切面钩子这套东西搭起来。前期投入大概两三天，后面省下的时间是以月计的。