2.9 车间监控：回调系统

2.9 车间监控：回调系统

本节摘要：回调是横穿所有工位的监控轨道——链的启动、模型的每个 token、工具的每次调用都会以事件形式推送出来。本节讲事件全景、自定义监控探头的编写规范，以及日志、计费、流式输出三个实战用法。

没有监控的产线是黑盒

产线一旦组装完成，你很快会面对三个问题：这次调用到底花了多少 token？慢是慢在模型还是慢在检索？用户投诉回答诡异，当时的完整输入是什么？裸写代码只能在每处手工插打印语句，而 LangChain 把这些"插点"标准化成了事件流：任何工位的启动、结束、出错、逐 token 输出，都会通知到挂在上面的回调处理器。

事件全景先过一眼，写探头前知道有哪些钩子可用：

写第一个监控探头

自定义处理器只需要继承 BaseCallbackHandler、覆写你关心的事件：

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from time import time

class WorkshopMonitor(BaseCallbackHandler):
    """车间监控探头：记录每道工序的耗时与产出"""

    def on_chain_start(self, serialized, inputs, **kwargs):
        self.t0 = time()
        print(f"[产线启动] 输入：{inputs}")

    def on_chain_end(self, outputs, **kwargs):
        print(f"[产线完成] 耗时 {time()-self.t0:.2f} 秒")

    def on_llm_end(self, response, **kwargs):
        # 从模型返回里读出本次 token 用量
        usage = response.llm_output.get("token_usage") \
                if response.llm_output else {}
        print(f"[动力单元] token 用量：",
              usage.get("total_tokens", "未知"))

monitor = WorkshopMonitor()

chain = (ChatPromptTemplate.from_template("给{p}的公司起名，只输出名字")
         | ChatOpenAI(temperature=0)
         | StrOutputParser())

# 探头可以临时挂载，用完即摘
with monitor:
    print("产出：", chain.invoke({"p": "彩色袜子"}))
# 输出示例：
# [产线启动] 输入：{'p': '彩色袜子'}
# [动力单元] token 用量： 87
# 产出： 彩织坊
# [产线完成] 耗时 1.34 秒
​

也可以在调用时传入，只监控特定一次执行：

# 方式二：调用级挂载，config 里带上处理器
print(chain.invoke({"p": "彩色袜子"},
                   config={"callbacks": [monitor]}))
​

三个实战用法

用法一：流式渲染。 交互产品的逐字输出，本质就是把 on_llm_new_token 事件接到前端：

class StreamHandler(BaseCallbackHandler):
    def __init__(self):
        self.tokens = []

    def on_llm_new_token(self, token, **kwargs):
        # 每个 token 到达即转发（这里先攒起来演示）
        self.tokens.append(token)

streamer = StreamHandler()
chain2 = (ChatPromptTemplate.from_template("用50字介绍{p}的卖点")
          | ChatOpenAI(streaming=True)
          | StrOutputParser())

with streamer:
    chain2.invoke({"p": "彩色袜子"})
print("共收到 token 片段：", len(streamer.tokens))
# 输出示例：共收到 token 片段：38
​

用法二：计费台账。 on_llm_end 里的 token 用量是成本核算的原始凭证，攒起来就是一张实时台账：

class CostLedger(BaseCallbackHandler):
    def __init__(self, price_per_1k=0.002):
        self.price = price_per_1k
        self.total = 0

    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output or {}
        tokens = (usage.get("token_usage") or {}).get("total_tokens", 0)
        self.total += tokens * self.price / 1000

ledger = CostLedger()
with ledger:
    chain.invoke({"p": "彩色袜子"})
    chain.invoke({"p": "手绘围巾"})
print(f"本次两单成本约 {ledger.total:.6f} 美元")
# 输出示例：本次两单成本约 0.000374 美元
​

用法三：慢工序定位。 在链的每一段分别挂探头，延迟分布立刻可见——是模型慢、检索慢还是工具慢，数据说话：

# 分段挂探头：检索段与生成段各一个计时器
retrieval_timer = WorkshopMonitor()
with retrieval_timer:
    docs = ["虚拟检索结果"]  # 演示占位，实际调用检索器
print("检索段耗时已记录")
​

内置探头也值得一提：调试期的 verbose 开关（2.3 节见过）本质就是一个内置回调，把链的每步输入输出打到控制台。生产期则换成结构化日志后端，把同样的事件流写进日志系统。

⚠️ 回调里做重活会拖慢整条产线：事件在工序间隙同步触发，探头里再去调外部接口（发通知、写远端日志）要加超时与异步，否则监控本身成为最大延迟源。

💡 探头无侵入是回调设计的精髓：装与不装监控，产线代码一个字不用改。反过来，任何"必须改链代码才能记录"的需求，都该先想想是不是回调没用好。

本章收官

至此九个工位全部开过机：主带三件（模型、提示、链）承载每次调用，调度两件（代理、记忆）改变流向，供线三件（加载、向量库、检索）输送知识，监控一件（回调）横穿全场。零件齐了，下一章开始改装产线：评估是质检台、调试日志是仪表盘、安全是护栏、性能是涡轮——所有改装都建立在你刚建立的零件认知上。