LangChain 中的 Callback 回调机制是什么？有什么用？

标签：LangChain、Callback 回调

🏛 业务架构师

LangChain 的 Callback 机制是贯穿执行全链路的"事件钩子"——框架在关键节点（模型调用开始/结束、链启动/结束、工具执行、Agent 步骤、重试、错误）发出事件，注册的 CallbackHandler 接收处理，主流程不受影响。

典型用途（每条对应一个业务诉求）：

可观测性：记录每次 LLM 调用的输入输出、token 用量、延迟——自建日志或接 LangSmith/OpenTelemetry。价值：没有它，LLM 应用是黑盒，成本和效果都无从管理；
成本控制与计量：累计 token 消耗、按用户/租户计费、超预算熔断——SaaS 化 AI 产品的计费基础；
审计与合规：全量留痕"谁问了什么、AI 答了什么、调了什么工具"——金融、医疗场景的刚需；
流式处理：StreamingStdOutCallbackHandler 实现打字机输出——体验标配；
告警与降级：错误率、超时率超阈值时告警或自动切换备用模型。

用法：实现 BaseCallbackHandler 的相关方法（on_llm_start/on_llm_end 等），以构造参数或环境变量注入；新版本推荐 LangGraph 的配置式传参。

架构认知：Callback 是 AOP（切面）思想在 LLM 编排里的落地——业务逻辑与横切关注点（日志/计费/审计）分离。我的团队铁律：生产系统的 LLM 调用必须挂审计与成本 Callback，上线检查项之一——事后补监控的代价远大于开始就带上。

🏗 系统架构师

Callback = LangChain 的横切关注点机制：在链路执行的关键节点（LLM 调用前后、工具执行前后、链开始结束）触发用户注册的钩子函数——AOP（面向切面编程）在 LLM 框架的落地。

用途（生产价值清单）：

可观测：tracing 全链路（LangSmith/自研 OTel 上报——每步的输入输出、token 消耗、延迟）、指标打点（QPS、成本、错误率）；
审计合规：全量调用留痕（谁、何时、调了什么模型/工具、结果）——金融医疗刚需；
成本控制：token 计量与配额检查，超限熔断或告警；
行为干预：内容过滤（输入输出审查）、敏感信息脱敏、工具调用审批闸门（before_tool 钩子里拦高危操作）；
调试开发：verbose 日志、中间结果打印。

用法形态：

python

class MyHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kw): ...
    def on_llm_end(self, response, **kw): ...      # 记录 token/延迟
    def on_tool_error(self, error, **kw): ...       # 工具异常上报

chain.invoke(inputs, config={"callbacks": [MyHandler()]})

架构师视角的要点：

钩子是侵入式低、杠杆高的扩展点——观测、安全、计量不该散落在业务代码，统一挂 Callback 层；
性能纪律：同步 Callback 会累加延迟（重 I/O 的上报改异步批量）；tracing 全量上报成本不低，采样率要设计（1%~10% 全量 + 全量错误）；
边界：LangChain 1.x 后 LangSmith 集成与 OTel 原生化在淡化手写 Callback 的场景，但自定义治理逻辑（审批、脱敏）仍靠它——框架会演进，"横切关注点独立于业务逻辑"的架构原则不变。

💻 全栈开发专家

Callback 是 LangChain 的事件钩子机制：在 LLM 调用、链执行、工具调用的关键节点触发回调，用于横切关注点——不改业务代码就能挂上观测和治理。

事件点：on_llm_start / on_llm_new_token / on_llm_end / on_chain_start / on_chain_end / on_tool_start / on_tool_end / on_retry 等。

典型用途（生产四大件）：

观测追踪：记录每次 LLM 调用的 prompt/响应/延迟/token——自研接入或用 LangSmith/Langfuse 的 tracer（本质就是 callback）；
计费与配额：按 token 统计到租户/用户，超配额熔断；
审计合规：全量落库谁问了什么、AI 答了什么、调了什么工具；
调试：开发期打印中间步骤（set_debug(True) 全链路日志）。

python

from langchain_core.callbacks import BaseCallbackHandler

class TokenMeter(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs):
        u = response.llm_output["token_usage"]
        metrics.add(total=u["total_tokens"],
                    cost=u["prompt_tokens"]*0.001 + u["completion_tokens"]*0.002)

chain.invoke(q, config={"callbacks": [TokenMeter()]})

工程要点：

callback 抛异常会打断主流程——观测类回调内部必须 try/except 吞掉；
同步回调阻塞主链路，重的逻辑（落库、上报）丢异步队列；
每个 Runnable 都可以挂独立 callbacks（config 传入），也可以全局 set_default_callback——注意别重复计费；
流式场景用 on_llm_new_token 实时统计首 token 延迟。

一句话：Callback 是 LangChain 的 AOP——生产系统的可观测、成本、审计三座大山全靠它搭。

LangChain 中的 Callback 回调机制是什么？有什么用？ ​

🏛 业务架构师 ​

🏗 系统架构师 ​

💻 全栈开发专家 ​

LangChain 中的 Callback 回调机制是什么？有什么用？

🏛 业务架构师

🏗 系统架构师

💻 全栈开发专家