LangChain Callbacks 与事件驱动架构

一、概念速查

核心概念

概念 定义
Callback LangChain 执行过程中触发的生命周期事件的回调函数
BaseCallbackHandler 所有回调处理器的抽象基类，定义各事件的接口
CallbackManager 事件分发中心，管理一组 CallbackHandler 的注册与调用
事件驱动架构 执行流程中异步/同步地产生事件，由注册的处理器消费的模式

BaseCallbackHandler 核心接口

事件方法 触发时机 典型场景
on_llm_start LLM 调用开始时 记录请求 prompt、计费
on_llm_end LLM 调用完成时 记录响应 token 数、延迟
on_llm_error LLM 调用抛出异常时 告警、重试逻辑
on_chat_model_start ChatModel 调用开始时 同上，仅 ChatModel 路径
on_chain_start RunnableChain 开始时 追踪链路入口
on_chain_end RunnableChain 完成时 记录链路输出
on_chain_error RunnableChain 异常时 链路级告警
on_tool_start Tool 调用开始时 审计工具参数
on_tool_end Tool 调用完成时 记录工具输出
on_tool_error Tool 抛出异常时 工具级告警
on_retry Retry 机制触发重试时 监控重试频率
on_text 输出文本时 日志记录
on_agent_action Agent 决定执行一个 Action 时 Agent 步骤追踪
on_agent_finish Agent 完成全部步骤时 Agent 最终输出

内置处理器

处理器 作用 依赖
StdOutCallbackHandler 将事件输出到 stdout langchain-core
FileCallbackHandler 将事件写入文件 langchain-core
ConsoleCallbackHandler LangChain CLI 控制台输出 langchain-cli

二、底层原理

2.1 事件分发机制

CallbackManager 内部维护一个 handlers: list[BaseCallbackHandler] 列表。当一个事件被触发时（例如 LLM 开始调用），CallbackManager.on_llm_start() 会遍历所有注册的 handler，依次调用其 on_llm_start() 方法：

RunnableChain.invoke()

CallbackManager.on_chain_start()

Handler1.on_chain_start()

Handler2.on_chain_start()

Handler3.on_chain_start()

LLM.invoke()

CallbackManager.on_llm_start()

Handler1.on_llm_start()

Handler2.on_llm_start()

Handler3.on_llm_start()

CallbackManager.on_llm_end()

Handler1.on_llm_end()

Handler2.on_llm_end()

Handler3.on_llm_end()

CallbackManager.on_chain_end()

Handler1.on_chain_end()

Handler2.on_chain_end()

Handler3.on_chain_end()

每个 Runnable（Chain、LLM、Tool、Retriever 等）都持有自己的 CallbackManager 实例。当外层 Runnable 调用内层 Runnable 时，callback 继承链会自动传递——子 Runnable 会继承父 Runnable 的 callback handlers，同时可以追加自己的 handler。

2.2 继承链机制

LangChain 通过 RunnableConfig 中的 callbacks 字段传递 handler。当一个 chain 包含一个 LLM 调用时：

# langchain>=0.3.0
from langchain_core.runnables import RunnableConfig

config = RunnableConfig(callbacks=[my_handler])
chain.invoke({"input": "hello"}, config=config)
# 内部 LLM 自动继承 config 中的 callbacks

继承链的实现路径：

  1. Runnable 的 invoke() 从 RunnableConfig 提取 callbacks
  2. 如果 RunnableConfig 没有显式传 callbacks，从 parent run 继承
  3. CallbackManager 新建时合并 parent 的 handlers 与当前配置的 handlers
  4. 事件冒泡：内层事件先通知本地 handler，再通知 parent handler

parent_config.callbacks

Chain CallbackManager

LLM CallbackManager

LLM.on_llm_start

本地Handler

继承自Chain的Handler

Chain.on_chain_start

继承自...的Handler

2.3 同步与异步双通道

BaseCallbackHandler 同时定义同步和异步接口：

# langchain>=0.3.0
from langchain_core.callbacks import BaseCallbackHandler

class MyHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        # 同步路径
        pass

    async def on_llm_start(self, serialized, prompts, **kwargs):
        # 异步路径
        pass

CallbackManager 会在同步执行链（invoke）中只调用同步方法，在异步执行链（ainvoke）中只调用异步方法。如果处理器只实现了同步方法，异步路径会自动降级为同步调用。

2.4 构造函数传参 vs 运行时传参

传参方式有两种：

# langchain>=0.3.0
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import StdOutCallbackHandler

# 方式一：构造时传入（生命周期绑定）
model = ChatOpenAI(
    model="gpt-4o-mini",
    callbacks=[StdOutCallbackHandler()],
)

# 方式二：运行时传入（每次调用独立）
model.invoke("你好", config={"callbacks": [StdOutCallbackHandler()]})

方式一适用于固定 logger，方式二适用于请求级别的追踪 ID 注入。

三、架构设计原则

3.1 处理器应当无状态或幂等

Callback handler 可能在多线程环境下被共享。如果 handler 持有状态（如累加 token 计数器），需要确保线程安全或使用 run_inline 隔离。

推荐模式：每个请求创建一个新的 handler 实例，在 on_chain_start 中注入 run_id：

# langchain>=0.3.0
import json
from uuid import UUID
from pathlib import Path
from langchain_core.callbacks import BaseCallbackHandler

class AuditHandler(BaseCallbackHandler):
    """审计日志处理器：记录每次 LLM 调用的入参和出参"""

    def __init__(self, log_dir: str = "audit_logs"):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(exist_ok=True)

    def on_llm_start(self, serialized, prompts, **kwargs):
        run_id = kwargs.get("run_id")
        entry = {"event": "llm_start", "run_id": str(run_id), "prompts": prompts}
        (self.log_dir / f"{run_id}.jsonl").write_text(
            json.dumps(entry, ensure_ascii=False) + "\n", encoding="utf-8"
        )

    def on_llm_end(self, response, **kwargs):
        run_id = kwargs.get("run_id")
        entry = {"event": "llm_end", "run_id": str(run_id), "llm_output": response.llm_output}
        with open(self.log_dir / f"{run_id}.jsonl", "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    def on_llm_error(self, error, **kwargs):
        run_id = kwargs.get("run_id")
        entry = {"event": "llm_error", "run_id": str(run_id), "error": repr(error)}
        with open(self.log_dir / f"{run_id}.jsonl", "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

3.2 不要阻塞主流程

Callback handler 与主执行流程是同步关系——handler 的执行时间直接累加到 invoke() 的总耗时。如果需要异步持久化（写入数据库、发送 HTTP 请求），使用队列解耦：

# langchain>=0.3.0
import threading
from queue import Queue
from langchain_core.callbacks import BaseCallbackHandler

class AsyncLogHandler(BaseCallbackHandler):
    """写日志到远端，通过队列解耦不阻塞主流程"""

    def __init__(self):
        self.queue = Queue()
        self._worker = threading.Thread(target=self._consumer, daemon=True)
        self._worker.start()

    def on_llm_start(self, serialized, prompts, **kwargs):
        self.queue.put(("llm_start", serialized, prompts, kwargs.get("run_id")))

    def on_llm_end(self, response, **kwargs):
        self.queue.put(("llm_end", response.llm_output, kwargs.get("run_id")))

    def _consumer(self):
        while True:
            event = self.queue.get()
            # 实际写入远端存储
            ...

3.3 使用 tags 细粒度过滤

在生产环境中可能注册多个 handler，每个 handler 只关心特定链路：

# langchain>=0.3.0
chain = prompt | model | parser
chain.invoke(
    {"input": "hello"},
    config={
        "tags": ["critical", "user-facing"],
        "callbacks": [alert_handler],
    },
)

在 handler 内检查 tags：

# langchain>=0.3.0
class AlertHandler(BaseCallbackHandler):
    def on_chain_error(self, error, **kwargs):
        tags = kwargs.get("tags", [])
        if "critical" in tags:
            send_alert(error)  # 仅关键链路触发告警

3.4 典型场景组合

场景 推荐处理器 关键事件
调试开发 StdOutCallbackHandler 全部事件
生产日志 自定义 JsonLogHandler on_llm_start/end/error
合规审计 自定义 AuditHandler on_tool_start/end（记录工具参数+输出）
性能监控 自定义 MetricHandler on_llm_end（计算耗时 + token 数）
异常告警 自定义 AlertHandler on_llm_error, on_chain_error, on_tool_error
调用追踪 LangSmithTracer（远端 handler） 全部事件

3.5 与 LangSmith 的关系

LangSmith 的追踪功能本质上是注册了一个远端 CallbackHandler。当设置 LANGCHAIN_TRACING_V2=true 时，LangChain 自动将 LangSmithTracer（继承自 BaseCallbackHandler）注入到 CallbackManager 中：

LangChain 应用

CallbackManager

StdOutCallbackHandler

自定义AuditHandler

LangSmithTracer
(远端CallbackHandler)

序列化 + POST

LangSmith API
(云端存储/可视化)

LangSmithTracer 在各事件回调中收集运行数据（prompt、response、token 计数、延迟、错误），序列化为 protobuf 或 JSON，异步 POST 到 LangSmith 后端。用户可在 LangSmith Web UI 中查看 trace 树、对比实验、分析成本。本质上，LangSmith 就是一个运行在远端的、自带存储和可视化能力的超级 CallbackHandler。

四、快速参考

完整导入路径

from langchain_core.callbacks import BaseCallbackHandler  # langchain>=0.3.0
from langchain_core.callbacks import CallbackManager      # langchain>=0.3.0
from langchain_core.callbacks import StdOutCallbackHandler # langchain>=0.3.0
from langchain_core.callbacks import FileCallbackHandler   # langchain>=0.3.0
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.runnables import RunnableConfig

最小自定义 handler 模板

# langchain>=0.3.0
from langchain_core.callbacks import BaseCallbackHandler

class MyHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        # serialized: dict, LLM 配置的序列化表示
        # prompts: list[str], 传入的 prompt 列表
        # kwargs: run_id, parent_run_id, tags, metadata 等
        pass

    def on_llm_end(self, response, **kwargs):
        # response: LLMResult, 包含 generations 和 llm_output
        # llm_output 中通常包含 token_usage
        pass

    def on_tool_start(self, serialized, input_str, **kwargs):
        # input_str: str, 工具的输入（已转为字符串）
        pass

    def on_tool_end(self, output, **kwargs):
        # output: str, 工具的输出
        pass

注意事项

  • Handler 中的异常不会中断主流程（LangChain 默认 raise_error=False），但可以通过配置 CallbackManager(raise_error=True) 改变
  • 同一个 handler 实例不要在多线程 invoke 中共享可变状态
  • on_retry 事件在重试发生前触发，可用来记录重试原因和次数
  • run_id 是 UUID 类型，可作为链路追踪的全局标识符
# 架构 # java
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

企业级Agent怎么搭建，腾讯云给出了标准答案

avatar AI Agent技术社区
cover

两个开源项目，带你系统学习 AI Agent

avatar AI Agent技术社区
cover

Android 工程师的 AI 搭档：Android CLI + Skills 实战笔记

avatar AI Agent技术社区