你有没有遇到过这种情况？
用 LangChain 搭了一个 AI 应用，跑起来挺顺，但一旦想搞清楚下面这些问题，就有点抓瞎：
这次请求到底花了多少 token？哪个环节最烧钱？
为什么这次回答这么慢？是模型慢，还是某个工具卡住了？
中间到底调了哪些链、哪些工具？模型看到的完整提示词是什么？
如果出错了，是哪个节点挂的？错误信息在哪？
你可能会想：那我就在每个节点里加日志、加计时、加 try-catch 呗。
可以，但代价是：业务代码会变得越来越臃肿。本来只该负责“生成 SQL”的节点，现在还要管打日志、算 token、上报监控。这就像让厨师一边炒菜，一边还要自己记流水账、拍视频、发朋友圈。
LangChain 的回调机制，就是为了解决这个问题。
一、回调机制是什么？先看一个生活比喻
你寄了一个快递，想知道包裹到哪了。
你不需要跟着快递员跑，也不需要给每个中转站打电话。你只需要在快递公司留个手机号，包裹每到一个节点，快递公司自动给你发一条通知：
已揽收
已到达分拨中心
正在派送
已签收
回调机制就是这个“自动通知”系统。
在 LangChain 里：
快递公司 = LangChain 运行时
包裹 = 一次 chain 调用
通知 = 回调事件
你的手机号 = 你写的
CallbackHandler
你只需要告诉 LangChain：“发生这些事的时候，通知我一下。”
然后 LangChain 就会在合适的时机自动调用你写好的方法。
这就叫：不要调用我们，我们会调用你。
二、核心角色只有两个
1.BaseCallbackHandler：你的“通知接收器”
这是一个抽象基类。你继承它，重写你关心的方法。
比如：
python from langchain_core.callbacks import BaseCallbackHandler class MyHandler(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): print("LLM 开始调用了") def on_llm_end(self, response, **kwargs): print("LLM 调用结束了") def on_llm_error(self, error, **kwargs): print(f"LLM 出错了：{error}")这些方法名不是随便起的，是 LangChain 规定好的“事件名”。
你重写了哪个，哪个事件发生时就会调用你。
2.CallbackManager：LangChain 内部的“调度中心”
你不需要直接操作它。
你只需要把 handler 挂到 LLM 或 chain 上，剩下的事 LangChain 自己会做。
python llm = ChatOpenAI(callbacks=[MyHandler()])或者：
python chain.invoke(..., config={"callbacks": [MyHandler()]})之后，每次 LLM 开始、结束、报错，MyHandler里对应的方法就会被自动调用。
三、第一个示例：打印每一步在干什么
LangChain 自带了一个StdOutCallbackHandler，可以直接把关键事件打印到控制台。
python from langchain_core.callbacks import StdOutCallbackHandler from langchain_openai import ChatOpenAI handler = StdOutCallbackHandler() llm = ChatOpenAI(callbacks=[handler]) llm.invoke("你好，介绍一下你自己")运行后，控制台会输出类似：
text [chain/start] Entering Chain run... [llm/start] Entering LLM run... [llm/end] Exiting LLM run... [chain/end] Exiting Chain run...这对调试非常有用。
你不需要改任何业务代码，就能看到 LangChain 内部到底跑了哪些步骤。
四、第二个示例：统计花了多少 token、多少时间
打印只是第一步。
真正生产环境里，我们更关心成本和性能。
下面写一个简单的TokenTracker：
python import time from langchain_core.callbacks import BaseCallbackHandler class TokenTracker(BaseCallbackHandler): def __init__(self): self.total_tokens = 0 self.start_time = None def on_llm_start(self, serialized, prompts, **kwargs): self.start_time = time.time() def on_llm_end(self, response, **kwargs): latency = time.time() - self.start_time # 不同版本字段可能略有差异，这里做示意 usage = {} if response.llm_output: usage = response.llm_output.get("token_usage", {}) total = usage.get("total_tokens", 0) self.total_tokens += total print(f"本次调用：{total} tokens，耗时 {latency:.2f} 秒") def on_llm_error(self, error, **kwargs): print(f"调用失败：{error}")挂上去：
python tracker = TokenTracker() llm = ChatOpenAI(callbacks=[tracker]) llm.invoke("帮我写一首关于春天的诗") print(f"总共用了 {tracker.total_tokens} tokens")这样，你既没有改业务代码，又拿到了 token 和延迟数据。
五、第三个示例：按业务阶段归因
如果你用 LangGraph 搭了一个多节点应用，比如：
意图识别
生成 SQL
执行查询
总结回答
你可能会想知道：到底哪个节点最烧 token？
这时候就可以利用on_chain_start和on_chain_end。
LangGraph 每个节点执行时，会表现为一个 chain run。
你可以在on_chain_start里记下run_id -> 节点名的映射，然后在on_llm_end里通过parent_run_id回溯，找到当前 LLM 调用属于哪个节点。
简化示意：
python class StageTracker(BaseCallbackHandler): def __init__(self): self.run_id_to_name = {} self.stage_tokens = {} def on_chain_start(self, serialized, inputs, *, run_id, **kwargs): name = serialized.get("name", "unknown") self.run_id_to_name[run_id] = name def on_chain_end(self, outputs, *, run_id, **kwargs): self.run_id_to_name.pop(run_id, None) def on_llm_end(self, response, *, parent_run_id, **kwargs): stage = self.run_id_to_name.get(parent_run_id, "unknown") usage = response.llm_output.get("token_usage", {}) if response.llm_output else {} total = usage.get("total_tokens", 0) self.stage_tokens[stage] = self.stage_tokens.get(stage, 0) + total最后你就能得到：
python { "intent_classify": 320, "generate_sql": 1100, "summarize": 570 }哪个环节最贵，一目了然。
六、异步回调：别让监控拖慢主流程
早期的回调是同步的。
如果你在on_llm_end里写文件、发 HTTP 请求，它会阻塞主流程，增加响应时间。
现在 LangChain 支持异步回调。你可以继承AsyncCallbackHandler，把方法写成async def：
python from langchain_core.callbacks import AsyncCallbackHandler class AsyncTokenTracker(AsyncCallbackHandler): async def on_llm_end(self, response, **kwargs): # 这里可以做异步 I/O，不会阻塞主流程 await save_to_database(response)这样，监控逻辑就不会拖慢用户请求。
七、回调机制能帮你做什么？
一旦你理解了这套机制，很多生产级需求都会变得很简单：
成本核算：按模型、按阶段统计 token，算出每次请求花了多少钱。
性能监控：记录每个 LLM 调用、每个工具调用的耗时。
调试排错：打印完整提示词、模型输出、错误堆栈。
安全审计：记录所有工具调用和输入输出，便于事后审查。
流式输出：通过
on_llm_new_token实时捕获每个 token，实现打字机效果。链路追踪：把事件上报到 LangSmith、Datadog、OpenTelemetry，形成完整调用链。
而且，所有这些都不需要改业务节点代码。
你只需要写一个 handler，挂上去，完事。
八、总结
LangChain 的回调机制，本质上是一套事件驱动的可观测性框架。
它用“控制反转”把监控逻辑和业务逻辑解耦：
协议是 LangChain 的：
BaseCallbackHandler、on_llm_start、on_llm_end等。逻辑是你自己的：记什么、怎么记、上报到哪里，完全由你决定。
它就像给你的 AI 应用装了一个“黑匣子”。
飞机正常飞的时候，黑匣子默默记录；一旦你想知道发生了什么，打开它，数据都在。
所以，下次你想在 LangChain 应用里加日志、算成本、做监控，先别急着改节点代码。
想想能不能写个CallbackHandler，挂上去。
业务代码只管飞，黑匣子的事，交给回调。
