上一篇我们聊了 LangChain 的BaseCallbackHandler它像快递通知一样让你在 LLM 调用开始、结束、出错时收到消息。但 LangGraph 不只是“链”它是一张图。图有自己的生命周期节点流转、条件分支、循环、中断、恢复。这些事件通用的BaseCallbackHandler可管不着。所以 LangGraph 在BaseCallbackHandler的基础上专门扩展了一个GraphCallbackHandler。一、先想清楚图需要什么样的“通知”假设你搭了一个客户支持 Agent用 LangGraph 编排text用户输入 → 意图识别 → 查询知识库 → 生成回答 ↓ 如果置信度低 ↓ 中断等待人工审核 ↓ 恢复继续生成回答你想知道图在哪个检查点checkpoint被中断了中断时带的是什么载荷payload人工审核后图是从哪个检查点恢复的这些问题on_llm_end、on_chain_end回答不了。它们关注的是LLM 调用和链执行不是图的生命周期。于是 LangGraph 定义了两个专属事件事件触发时机GraphInterruptEvent图因中断而暂停执行GraphResumeEvent图从检查点恢复执行以及一个专属基类GraphCallbackHandler。二、GraphCallbackHandler 是什么一句话它是BaseCallbackHandler的子类专门用来观察 LangGraph 特有的图生命周期事件。python class GraphCallbackHandler(BaseCallbackHandler): def on_interrupt(self, event: GraphInterruptEvent) - Any: 图因中断暂停时触发 def on_resume(self, event: GraphResumeEvent) - Any: 图从检查点恢复时触发它继承自BaseCallbackHandler所以LangChain 的所有回调钩子你依然能用on_llm_start、on_chain_end、on_tool_error……。它只是在父类的基础上增加了图级别的两个钩子。关键规则只有一条只有继承自GraphCallbackHandler的处理器才会收到on_interrupt和on_resume事件。普通的BaseCallbackHandler收不到。三、事件里有什么——认识两个 Event 对象GraphInterruptEvent图被中断时LangGraph 会构造这个事件对象塞进这些字段python dataclass(frozenTrue) class GraphInterruptEvent: run_id: UUID | None # 本次图运行的唯一 ID status: GraphLifecycleStatus # 中断发生时的循环状态 checkpoint_id: str # 中断关联的检查点 ID checkpoint_ns: tuple[str, ...] # 检查点命名空间路径主图/子图 interrupts: tuple[Interrupt, ...] # 导致图暂停的中断载荷status的可能值有input、pending、done、interrupt_before、interrupt_after、out_of_steps。它告诉你中断发生在生命周期的哪个位置。interrupts是最有意思的字段。它是一个元组里面装的是触发这次暂停的Interrupt对象。你的节点里调用interrupt()时传了什么值这里就能读到什么。GraphResumeEvent图从检查点恢复时构造这个事件python dataclass(frozenTrue) class GraphResumeEvent: run_id: UUID | None status: GraphLifecycleStatus checkpoint_id: str # 从哪个检查点恢复的 checkpoint_ns: tuple[str, ...]注意GraphResumeEvent没有interrupts字段。恢复事件只关心“从哪恢复”不关心“为什么中断”。四、动手写一个监控图的中断与恢复假设你要做一个“人工审核”流程当 Agent 准备执行某个敏感操作时先中断等人批准后再恢复。节点里触发中断python from langgraph.types import interrupt def sensitive_action_node(state): user_input state[user_input] # 中断把待审核的信息传出去 decision interrupt({ action: delete_user_data, target: user_input[user_id], reason: 用户请求删除账户 }) # 恢复后decision 是人工审核的结果 if decision approve: return {result: 已执行删除} else: return {result: 操作被拒绝}写一个监控 Handlerpython from langgraph.callbacks import GraphCallbackHandler, GraphInterruptEvent, GraphResumeEvent class HumanReviewMonitor(GraphCallbackHandler): def on_interrupt(self, event: GraphInterruptEvent): print( * 40) print(图被中断了) print(f 检查点 ID: {event.checkpoint_id}) print(f 循环状态: {event.status}) for i, interrupt_payload in enumerate(event.interrupts): print(f 中断 {i1}: {interrupt_payload.value}) def on_resume(self, event: GraphResumeEvent): print( * 40) print(图恢复了) print(f 从检查点: {event.checkpoint_id}) print(f 循环状态: {event.status})挂上去跑起来python monitor HumanReviewMonitor() # 第一次调用图会在 interrupt 处暂停 config {configurable: {thread_id: review-001}} result graph.invoke( {user_input: {user_id: u_123}}, config{callbacks: [monitor], **config} )# 输出 # # 图被中断了 # 检查点 ID: ckpt_abc123 # 循环状态: interrupt_before # 中断 1: {action: delete_user_data, target: u_123, reason: 用户请求删除账户}人工审核后恢复执行python from langgraph.types import Command # 恢复图传入审核结果 graph.invoke( Command(resumeapprove), config{callbacks: [monitor], **config} )# 输出 # # 图恢复了 # 从检查点: ckpt_abc123 # 循环状态: pending整个过程你的节点代码里只有interrupt()和Command(resume...)监控逻辑完全由GraphCallbackHandler承担。五、和 LangChain 回调混着用你可能会问我既想统计 token又想监控中断怎么办答案很简单把两类 handler 一起放进config[callbacks]。python from langchain_core.callbacks import BaseCallbackHandler from langgraph.callbacks import GraphCallbackHandler class TokenTracker(BaseCallbackHandler): def on_llm_end(self, response, **kwargs): # 统计 token ... class GraphMonitor(GraphCallbackHandler): def on_interrupt(self, event): # 处理中断 ... graph.invoke( inputs, config{callbacks: [TokenTracker(), GraphMonitor()]} )运行时LangGraph 内部会做一次筛选从所有回调中挑出GraphCallbackHandler的子类实例构建一个专用的图回调管理器。TokenTracker照常收到 LLM 事件GraphMonitor收到中断/恢复事件互不干扰。六、它到底解决了什么问题回到开头那个客户支持 Agent 的场景。没有GraphCallbackHandler时你想知道“图为什么停了”“从哪恢复的”只能在节点里手动写日志、存状态、发通知或者事后翻检查点数据库猜测发生了什么。有了GraphCallbackHandler这些信息由 LangGraph主动推给你中断发生时你知道哪个检查点、什么载荷、循环处于什么状态恢复发生时你知道从哪个检查点、状态是什么。这对于Human-in-the-Loop场景尤其关键。人工审核界面需要知道“当前在等什么”审核系统需要知道“从哪继续”而GraphCallbackHandler提供了标准化的观测入口。七、总结GraphCallbackHandler是 LangGraph 在 LangChain 回调体系上的专项扩展。继承自BaseCallbackHandler所以 LangChain 的所有回调钩子你依然能用。新增on_interrupt和on_resume专门观察图的中断与恢复。事件对象携带检查点 ID、命名空间、中断载荷让“图停在哪、为什么停、从哪继续”变得可编程。通过config[callbacks]挂载和 LangChain 回调无缝混用。如果说BaseCallbackHandler是给你的 LLM 调用装了一个“通话记录器”那么GraphCallbackHandler就是给你的整张图装了一个“黑匣子”——它记录的不只是一次调用而是图的生命周期本身。图会中断会恢复会流转。黑匣子会告诉你它到底经历了什么。 GraphCallbackHandler：给你的 LangGraph 图引擎装一个“黑匣子”
上一篇我们聊了 LangChain 的BaseCallbackHandler它像快递通知一样让你在 LLM 调用开始、结束、出错时收到消息。但 LangGraph 不只是“链”它是一张图。图有自己的生命周期节点流转、条件分支、循环、中断、恢复。这些事件通用的BaseCallbackHandler可管不着。所以 LangGraph 在BaseCallbackHandler的基础上专门扩展了一个GraphCallbackHandler。一、先想清楚图需要什么样的“通知”假设你搭了一个客户支持 Agent用 LangGraph 编排text用户输入 → 意图识别 → 查询知识库 → 生成回答 ↓ 如果置信度低 ↓ 中断等待人工审核 ↓ 恢复继续生成回答你想知道图在哪个检查点checkpoint被中断了中断时带的是什么载荷payload人工审核后图是从哪个检查点恢复的这些问题on_llm_end、on_chain_end回答不了。它们关注的是LLM 调用和链执行不是图的生命周期。于是 LangGraph 定义了两个专属事件事件触发时机GraphInterruptEvent图因中断而暂停执行GraphResumeEvent图从检查点恢复执行以及一个专属基类GraphCallbackHandler。二、GraphCallbackHandler 是什么一句话它是BaseCallbackHandler的子类专门用来观察 LangGraph 特有的图生命周期事件。python class GraphCallbackHandler(BaseCallbackHandler): def on_interrupt(self, event: GraphInterruptEvent) - Any: 图因中断暂停时触发 def on_resume(self, event: GraphResumeEvent) - Any: 图从检查点恢复时触发它继承自BaseCallbackHandler所以LangChain 的所有回调钩子你依然能用on_llm_start、on_chain_end、on_tool_error……。它只是在父类的基础上增加了图级别的两个钩子。关键规则只有一条只有继承自GraphCallbackHandler的处理器才会收到on_interrupt和on_resume事件。普通的BaseCallbackHandler收不到。三、事件里有什么——认识两个 Event 对象GraphInterruptEvent图被中断时LangGraph 会构造这个事件对象塞进这些字段python dataclass(frozenTrue) class GraphInterruptEvent: run_id: UUID | None # 本次图运行的唯一 ID status: GraphLifecycleStatus # 中断发生时的循环状态 checkpoint_id: str # 中断关联的检查点 ID checkpoint_ns: tuple[str, ...] # 检查点命名空间路径主图/子图 interrupts: tuple[Interrupt, ...] # 导致图暂停的中断载荷status的可能值有input、pending、done、interrupt_before、interrupt_after、out_of_steps。它告诉你中断发生在生命周期的哪个位置。interrupts是最有意思的字段。它是一个元组里面装的是触发这次暂停的Interrupt对象。你的节点里调用interrupt()时传了什么值这里就能读到什么。GraphResumeEvent图从检查点恢复时构造这个事件python dataclass(frozenTrue) class GraphResumeEvent: run_id: UUID | None status: GraphLifecycleStatus checkpoint_id: str # 从哪个检查点恢复的 checkpoint_ns: tuple[str, ...]注意GraphResumeEvent没有interrupts字段。恢复事件只关心“从哪恢复”不关心“为什么中断”。四、动手写一个监控图的中断与恢复假设你要做一个“人工审核”流程当 Agent 准备执行某个敏感操作时先中断等人批准后再恢复。节点里触发中断python from langgraph.types import interrupt def sensitive_action_node(state): user_input state[user_input] # 中断把待审核的信息传出去 decision interrupt({ action: delete_user_data, target: user_input[user_id], reason: 用户请求删除账户 }) # 恢复后decision 是人工审核的结果 if decision approve: return {result: 已执行删除} else: return {result: 操作被拒绝}写一个监控 Handlerpython from langgraph.callbacks import GraphCallbackHandler, GraphInterruptEvent, GraphResumeEvent class HumanReviewMonitor(GraphCallbackHandler): def on_interrupt(self, event: GraphInterruptEvent): print( * 40) print(图被中断了) print(f 检查点 ID: {event.checkpoint_id}) print(f 循环状态: {event.status}) for i, interrupt_payload in enumerate(event.interrupts): print(f 中断 {i1}: {interrupt_payload.value}) def on_resume(self, event: GraphResumeEvent): print( * 40) print(图恢复了) print(f 从检查点: {event.checkpoint_id}) print(f 循环状态: {event.status})挂上去跑起来python monitor HumanReviewMonitor() # 第一次调用图会在 interrupt 处暂停 config {configurable: {thread_id: review-001}} result graph.invoke( {user_input: {user_id: u_123}}, config{callbacks: [monitor], **config} )# 输出 # # 图被中断了 # 检查点 ID: ckpt_abc123 # 循环状态: interrupt_before # 中断 1: {action: delete_user_data, target: u_123, reason: 用户请求删除账户}人工审核后恢复执行python from langgraph.types import Command # 恢复图传入审核结果 graph.invoke( Command(resumeapprove), config{callbacks: [monitor], **config} )# 输出 # # 图恢复了 # 从检查点: ckpt_abc123 # 循环状态: pending整个过程你的节点代码里只有interrupt()和Command(resume...)监控逻辑完全由GraphCallbackHandler承担。五、和 LangChain 回调混着用你可能会问我既想统计 token又想监控中断怎么办答案很简单把两类 handler 一起放进config[callbacks]。python from langchain_core.callbacks import BaseCallbackHandler from langgraph.callbacks import GraphCallbackHandler class TokenTracker(BaseCallbackHandler): def on_llm_end(self, response, **kwargs): # 统计 token ... class GraphMonitor(GraphCallbackHandler): def on_interrupt(self, event): # 处理中断 ... graph.invoke( inputs, config{callbacks: [TokenTracker(), GraphMonitor()]} )运行时LangGraph 内部会做一次筛选从所有回调中挑出GraphCallbackHandler的子类实例构建一个专用的图回调管理器。TokenTracker照常收到 LLM 事件GraphMonitor收到中断/恢复事件互不干扰。六、它到底解决了什么问题回到开头那个客户支持 Agent 的场景。没有GraphCallbackHandler时你想知道“图为什么停了”“从哪恢复的”只能在节点里手动写日志、存状态、发通知或者事后翻检查点数据库猜测发生了什么。有了GraphCallbackHandler这些信息由 LangGraph主动推给你中断发生时你知道哪个检查点、什么载荷、循环处于什么状态恢复发生时你知道从哪个检查点、状态是什么。这对于Human-in-the-Loop场景尤其关键。人工审核界面需要知道“当前在等什么”审核系统需要知道“从哪继续”而GraphCallbackHandler提供了标准化的观测入口。七、总结GraphCallbackHandler是 LangGraph 在 LangChain 回调体系上的专项扩展。继承自BaseCallbackHandler所以 LangChain 的所有回调钩子你依然能用。新增on_interrupt和on_resume专门观察图的中断与恢复。事件对象携带检查点 ID、命名空间、中断载荷让“图停在哪、为什么停、从哪继续”变得可编程。通过config[callbacks]挂载和 LangChain 回调无缝混用。如果说BaseCallbackHandler是给你的 LLM 调用装了一个“通话记录器”那么GraphCallbackHandler就是给你的整张图装了一个“黑匣子”——它记录的不只是一次调用而是图的生命周期本身。图会中断会恢复会流转。黑匣子会告诉你它到底经历了什么。