Hook 模式——在 Agent 生命周期里注入自定义逻辑——从零开始学 LangGraph（十三·番外篇）

Hook 模式——在 Agent 生命周期里注入自定义逻辑写在前面这是系列的第 13 期番外篇。前 12 期我们覆盖了 LangGraph 的所有核心功能但有一个模式一直没有以独立概念来讲——Hook钩子。Hook 是在某个执行点自动触发的自定义逻辑。它不一定暂停执行不一定等待用户输入——它可以是记个日志、“改个状态”、“发个通知”。LangGraph 里能找到三种 Hook 机制Hook 类型时机是否暂停典型用途编译期 Hook(Breakpoint)Node 执行前/后✅ 暂停调试、人工确认、状态修改运行时 Hook(Interrupt)Node 执行中✅ 暂停条件性审批回调 Hook(Callback)任意事件❌ 不暂停日志、监控、统计这一期把它们整合成一个模式来讲并给出几个真的能用在生产环境里的 Hook 组合。1. 编译期 Hookinterrupt_before / interrupt_after这是最容易被忽视的能力。compile()时配置在 Node 执行前后的钩子graph builder.compile( checkpointersaver, interrupt_before[critical_node], # 进 critical_node 前暂停 interrupt_after[llm_node], # llm_node 跑完后暂停 )1.1 hook 住之后能干什么# 第 1 步正常调用图会在 hook 点自动暂停 config {configurable: {thread_id: hook-demo}} stream graph.stream_events(input_data, config, versionv3) _ stream.output # 第 2 步检查当前状态hook 住时的快照 state graph.get_state(config) print(当前 State:, state.values) print(下一个 Node:, state.next) # 第 3 步可以修改状态 graph.update_state(config, {messages: [(ai, 我帮你改了回答)]}) # 第 4 步继续执行 stream2 graph.stream_events(Command(resumeNone), config, versionv3)1.2 实战在所有工具调用前自动校验from langgraph.graph import StateGraph, MessagesState, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command from typing import Annotated, Literal from langchain.tools import tool from langchain.chat_models import init_chat_model from langchain_core.messages import ToolMessage tools [] tools_by_name {} tool def transfer_money(to_account: str, amount: float) - str: 转账到指定账户 print(f 真实转账: {to_account} 金额 {amount}) return f已向 {to_account} 转账 {amount} 元 tool def send_email(to: str, content: str) - str: 发送邮件 return f已发送邮件至 {to} # 注册工具 tools [transfer_money, send_email] tools_by_name {t.name: t for t in tools} model init_chat_model(gpt-4o-mini, temperature0) model_with_tools model.bind_tools(tools) # ─── Agent ─── def llm_node(state: MessagesState) - dict: response model_with_tools.invoke([ (system, 你是银行助手可转账和发邮件。), *state[messages] ]) return {messages: [response]} def tool_node(state: MessagesState) - dict: last state[messages][-1] results [] for tc in last.tool_calls: t tools_by_name[tc[name]] results.append(ToolMessage(contentstr(t.invoke(tc[args])), tool_call_idtc[id])) return {messages: results} def should_continue(state: MessagesState) - Literal[tools, END]: last state[messages][-1] return tools if last.tool_calls else END builder StateGraph(MessagesState) builder.add_node(llm, llm_node) builder.add_node(tools, tool_node) builder.add_edge(START, llm) builder.add_conditional_edges(llm, should_continue, {tools: tools, END: END}) builder.add_edge(tools, llm) saver InMemorySaver() # ⭐️ 关键在 tools 节点前挂一个 Hook agent builder.compile( checkpointersaver, interrupt_before[tools], # ← 每次调工具前先暂停 ) def ask_with_audit(query: str): 调 Agent但每次调工具前都做审计 config {configurable: {thread_id: audit-1}} stream agent.stream_events( {messages: [(user, query)]}, config, versionv3, ) try: _ stream.output except: pass # ⭐️ Hook 点检查 LLM 准备调什么工具 if stream.interrupted: state agent.get_state(config) last_msg state.values[messages][-1] for tc in last_msg.tool_calls: print(f\n ⛔ Hook 住LLM 想调: {tc[name]}({tc[args]})) # 审计规则 if tc[name] transfer_money: amount tc[args].get(amount, 0) if amount 10000: print(f ❌ 金额 {amount} 超过限额拒绝执行) # 修改状态阻止这次调用 graph.update_state(config, { messages: [(ai, f转账金额 {amount} 元需要人工审批请联系客服。)] }) break else: print(f ✅ 金额 {amount} 在限额内放行) elif tc[name] send_email: print(f ✅ 发送邮件放行) # 继续执行 stream2 agent.stream_events(Command(resumeNone), config, versionv3) result stream2.output return result return stream.output if __name__ __main__: ask_with_audit(给张三转账 5000 元) # 通过 ask_with_audit(给李四转账 50000 元) # 被 hook 拦截关键洞察interrupt_before[tools]是一个全局 hook——每次Agent 准备调工具时都会触发。你不用在每个 Node 里写审计逻辑hook 一层搞定。2. 运行时 Hookinterrupt()第 7 期已经详细讲过interrupt()了这里换个视角——把它看作条件性运行时 hookdef review_node(state): # ... 前面的逻辑正常运行 ... # 只有满足条件时才触发 hook if state[amount] 10000: approved interrupt({ # ← 运行时 hook type: approval, amount: state[amount], }) if not approved: return {status: rejected} # hook 通过继续执行 return {status: approved}编译期 Hook vs 运行时 Hook维度编译期 Hook (interrupt_before/after)运行时 Hook (interrupt)配置位置compile()参数Node 函数内部是否条件性❌ 固定位置✅ 可以条件触发修改 State✅ 在外部调用update_state✅ 在 Node 内直接改典型场景全量审计、调试、流量录制条件审批、数据采集粒度Node 级Node 内任意位置3. 回调 HookBaseCallbackHandler前两种 hook 都会暂停执行。如果你只需要顺便做点事而不想暂停用回调 hook。LangChain 提供了BaseCallbackHandler可以在 LLM 调用、工具执行、链开始/结束时触发from langchain.callbacks.base import BaseCallbackHandler from langchain.chat_models import init_chat_model class LoggingHook(BaseCallbackHandler): 打日志的 Hook——不暂停不阻塞 def on_llm_start(self, serialized, prompts, **kwargs): print(f [HOOK] LLM 开始调用prompt 长度: {len(str(prompts))}) def on_llm_end(self, response, **kwargs): tokens response.llm_output or {} print(f [HOOK] LLM 返回token: {tokens.get(token_usage, {})}) def on_tool_start(self, serialized, input_str, **kwargs): print(f [HOOK] 工具开始: {serialized.get(name, unknown)}) def on_tool_end(self, output, **kwargs): print(f [HOOK] 工具结束输出长度: {len(str(output))}) def on_chain_error(self, error, **kwargs): print(f ❌ [HOOK] 链出错: {error}) # 在调用时传入 handler LoggingHook() model init_chat_model(gpt-4o-mini, temperature0) # 方式 1单次调用 model.invoke(你好, callbacks[handler]) # 方式 2在 Agent 中使用 result agent.invoke( {messages: [(user, 你好)]}, {configurable: {thread_id: 1}}, callbacks[handler], )3.1 统计 Hook记 token 消耗import json import time from datetime import datetime class TokenTracker(BaseCallbackHandler): 统计每次 Agent 调用的 token 消耗 def __init__(self): self.runs [] self._current_run {} def on_llm_start(self, serialized, prompts, **kwargs): self._current_run { start_time: time.time(), model: serialized.get(name, unknown), } def on_llm_end(self, response, **kwargs): self._current_run[end_time] time.time() self._current_run[duration] self._current_run[end_time] - self._current_run[start_time] usage response.llm_output or {} if hasattr(response, usage_metadata): usage { prompt_tokens: response.usage_metadata.get(input_tokens, 0), completion_tokens: response.usage_metadata.get(output_tokens, 0), total_tokens: response.usage_metadata.get(total_tokens, 0), } self._current_run[usage] usage self._current_run[timestamp] datetime.now().isoformat() self.runs.append(self._current_run) def print_summary(self): total_tokens sum(r.get(usage, {}).get(total_tokens, 0) for r in self.runs) total_cost total_tokens * 0.00015 / 1000 # gpt-4o-mini 的价格 print(f\n Token 统计:) print(f 总调用次数: {len(self.runs)}) print(f 总 Token: {total_tokens}) print(f 估算费用: ${total_cost:.6f}) print(f 平均耗时: {sum(r[duration] for r in self.runs) / len(self.runs):.2f}s if self.runs else 无调用) def save_to_file(self, pathtoken_logs.json): with open(path, w) as f: json.dump(self.runs, f, ensure_asciiFalse, indent2) print(f 日志已保存到 {path})4. 组合模式三层 Hook 一起用把三种 Hook 组合在一个 Agent 里实现全链路可观测 关键点暂停from langgraph.graph import StateGraph, MessagesState, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command from typing import Annotated, Literal # ─── 回调 Hook全局监控 ─── class MonitorHook(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): print(f [监控] LLM 调用开始) def on_tool_start(self, serialized, input_str, **kwargs): print(f [监控] 工具: {serialized.get(name)} 参数: {input_str[:80]}) def on_tool_end(self, output, **kwargs): print(f [监控] 工具返回: {str(output)[:80]}) # ─── 构建带三层 Hook 的 Agent ─── builder StateGraph(MessagesState) builder.add_node(llm, llm_node) builder.add_node(tools, tool_node) builder.add_edge(START, llm) builder.add_conditional_edges(llm, should_continue, {tools: tools, END: END}) builder.add_edge(tools, llm) # 第一层 Hook编译期——所有工具调用前暂停 agent builder.compile( checkpointerInMemorySaver(), interrupt_before[tools], ) def run_with_all_hooks(query: str): config {configurable: {thread_id: all-hooks}} monitor MonitorHook() # 回调 Hook 通过 callbacks 参数传入 stream agent.stream_events( {messages: [(user, query)]}, config, versionv3, callbacks[monitor], # ← 第三层 Hook回调 ) try: _ stream.output except: pass # 编译期 Hook 触发 if stream.interrupted: state agent.get_state(config) # 检查状态、可以做修改 last state.values[messages][-1] for tc in last.tool_calls: print(f [编译期Hook] 即将调用 {tc[name]}允许放行) # 放行 stream2 agent.stream_events(Command(resumeNone), config, versionv3) return stream2.output return stream.output三种 Hook 的执行顺序用户输入 ↓ [回调 Hook] on_chain_start → on_llm_start ↓ LLM 节点执行 ↓ [回调 Hook] on_llm_end ↓ 条件路由 → 决定调工具 ↓ [编译期 Hook] interrupt_before[tools] ← 暂停 ├── 检查、修改 └── Command(resumeNone) ← 放行 ↓ [回调 Hook] on_tool_start ↓ Tool 节点执行 ↓ [回调 Hook] on_tool_end ↓ 回到 LLM...5. 实战 Hook 模式模板5.1 调试 Hook打印每一步的 Stateclass DebugHook(BaseCallbackHandler): 开发调试用——打印每个事件的详细信息 def on_chain_start(self, serialized, inputs, **kwargs): name serialized.get(name, unnamed) print(f\n{─*40}) print(f▶️ 开始: {name}) def on_chain_end(self, outputs, **kwargs): print(f✅ 结束) print(f{─*40})5.2 拦截 Hook在 tool_node 前注入缓存# 在 interrupt_before[tools] hook 点 def inject_cache_hook(agent, config): state agent.get_state(config) last state.values[messages][-1] for tc in last.tool_calls: # 生成缓存 key cache_key f{tc[name]}:{json.dumps(tc[args], sort_keysTrue)} # 检查是否有缓存 cached cache.get(cache_key) if cached: print(f 命中缓存: {cache_key}) # 注入缓存结果跳过真实调用 agent.update_state(config, { messages: [ToolMessage(contentcached, tool_call_idtc[id])] })5.3 限流 Hook控制 Agent 调用频率import time from collections import deque class RateLimitHook: 限流 Hook——控制每分钟的 LLM 调用次数 def __init__(self, max_calls_per_minute10): self.timestamps deque() self.max_calls max_calls_per_minute def check(self) - bool: now time.time() # 清理超过 1 分钟的记录 while self.timestamps and now - self.timestamps[0] 60: self.timestamps.popleft() if len(self.timestamps) self.max_calls: return False # 限流 self.timestamps.append(now) return True # 在 interrupt_before[llm] 使用 rate_limiter RateLimitHook(max_calls_per_minute5) # 在 hook 点检查 def rate_limit_hook(agent, config): if not rate_limiter.check(): # 超过限制注入错误消息 agent.update_state(config, { messages: [(ai, 请求过于频繁请稍后再试。)] }) return False return True6. Hook 与 Functional APIFunctional API 里没有interrupt_before但可以通过回调 hook 和interrupt()达到类似效果from langgraph.func import entrypoint, task from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import interrupt, Command task def risky_operation(data: str) - str: return f处理结果: {data.upper()} entrypoint(checkpointerInMemorySaver()) def safe_workflow(input_data: str) - dict: 在 entrypoint 里做 hook 检查 # 运行时 hook执行前检查 if 危险 in input_data: approved interrupt({ type: safety_check, input: input_data, message: 检测到敏感输入是否继续, }) if not approved: return {status: blocked, reason: 安全策略拦截} result risky_operation(input_data).result() return {status: completed, result: result} # 使用回调 hook 做日志 from langchain.callbacks.base import BaseCallbackHandler class FuncHook(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f[FuncHook] task 开始) config {configurable: {thread_id: func-hook}} s1 safe_workflow.stream_events(正常数据, config, versionv3, callbacks[FuncHook()]) _ s1.output s2 safe_workflow.stream_events(危险操作, config, versionv3) _ s2.output if s2.interrupted: print(f⛔ Hook 拦截: {s2.interrupts[0].value}) safe_workflow.stream_events(Command(resumeFalse), config, versionv3)7. Hook 设计原则要做的# ✅ Hook 是正交的——不侵入业务逻辑 # 业务 Node 不需要知道自己被 hook 了 def business_node(state): return {result: process(state[data])} # ✅ Hook 应该可插拔——编译时控制 graph builder.compile( interrupt_before[tools] if is_production else None, ) # ✅ 回调 Hook 应该轻量——不能阻塞主流程 def on_llm_end(self, response, **kwargs): self.count 1 # 轻量 # ❌ 不要在这里调 API、写文件除非异步不要做的# ❌ 在回调里修改 State class BadHook(BaseCallbackHandler): def on_tool_end(self, output, **kwargs): # 回调不应该修改 graph 的核心状态 # 修改 State 应该在 compile-time hook 里做 pass # ❌ Hook 里有耗时操作阻塞主流程 def on_llm_start(self, ...): time.sleep(2) # 不要回调应该毫秒级返回 # ❌ 过度使用 interrupt_before # 每个 Node 都上 hook 会让执行变慢 # 只在需要的位置加8. 本期核心总结Hook 类型配置方式是否暂停典型用途编译期 Hookcompile(interrupt_before/after...)✅审计、拦截、注入、调试运行时 HookNode 内interrupt()✅ 条件性审批、数据采集、安全检查回调 HookBaseCallbackHandlercallbacks❌日志、监控、Token 统计、报警组合模式三种一起用-全链路可观测 关键点控制什么时候用哪种场景用哪个“每次调工具前我都想看一眼”编译期 Hookinterrupt_before[tools]“只有金额大于 1 万才要审批”运行时 Hookinterrupt()加条件判断“我想统计每次 Agent 调用花了多少钱”回调 HookBaseCallbackHandler“开发时我想看每一步的 State 变化”编译期 Hookget_state()“我想在 LLM 响应里自动注入公司免责声明”编译期 Hookinterrupt_after[llm]update_state()参考资料LangGraph Breakpointshttps://docs.langchain.com/oss/python/langgraph/breakpointsLangGraph InterruptsInterrupts - Docs by LangChainLangChain Callbackshttps://docs.langchain.com/oss/python/langchain/callbacksHook 模式是 LangGraph 最容易被低估的能力之一。它可以让你在不修改 Agent 业务逻辑的前提下注入审计、监控、限流、缓存等各种横切关注点。如果说 Node/Edge 是 Agent 的骨架State 是血肉那 Hook 就是神经系统——不直接干活但决定什么时候干什么活。