Callbacks与LangSmith

链路上有 prompt、model、parser 多级，线上出问题很难定位是哪一步慢或哪一步输入异常。Callbacks 在 Runnable 生命周期挂钩子；LangSmith 收集 trace，在 UI 里看每次 invoke 的树状 span 与 token 用量。

段末注释：Callbacks（回调）= Runnable/LLM 执行事件钩子；LangSmith = LangChain 官方观测与评测平台。

---

1. 一句话定位

维度	内容
角色	能力层 可观测性：日志、指标、trace
输入 → 输出	不改变链 IO；产生 side-effect 日志
典型调用入口	config={"callbacks": [handler]}、环境变量 LangSmith
与 LangGraph	图 invoke 同样传 config；LangSmith 显示 node 级 span

---

2. 实现逻辑

1. class MyHandler(BaseCallbackHandler): on_llm_end(...)
2. chain.invoke(x, config={"callbacks": [MyHandler()]})
3. 链执行：on_chain_start → on_llm_start → on_llm_end → on_chain_end
4. 开 LangSmith：export LANGCHAIN_TRACING_V2=true LANGCHAIN_API_KEY=...
5. 同上 invoke 自动上传 trace（无需手写 handler）
6. tags/metadata 在 config 里附加到 run

字段级变形：业务输入不变；LangSmith run 记录 inputs/outputs、latency、child runs。

---

3. 原理说明

3.1 BaseCallbackHandler

BaseCallbackHandler（类，langchain_core.callbacks.base）
功能：钩子观测 LLM / Chain / Tool。子 Runnable 继承父 config 的 callbacks。
默认值：所有 on_* 为空操作。最小实现：覆盖 ≥1 个钩子。

钩子	典型入参	最小调用次数
on_llm_start	prompts / messages	每次模型调用 1 次
on_llm_end	LLMResult	与 start 成对
on_chain_start/end	serialized, inputs / outputs	每个 Runnable 步骤
on_tool_start/end	serialized, input_str / output	每次工具

class TokenLogHandler(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs):
        print(type(response).__name__)
chain.invoke({"q": "首都"}, config={"callbacks": [TokenLogHandler()]})

LLMResult（数据类）
generations: list[list[ChatGeneration]]，最小形状 (1, 1)：一批 1 条、1 个候选。llm_output 默认可为 None。

3.2 LangSmith 环境变量

变量	默认值	最小维度
LANGCHAIN_TRACING_V2	未设置=关	开时必须为 "true"
LANGCHAIN_API_KEY	无	非空 Key
LANGCHAIN_PROJECT	默认项目名	非空 str

三者同时有效才上传 trace；只设 callbacks 仍可本地打印。

3.3 tags 与 metadata

config["tags"]：list[str]，默认 []。过滤时 ≥1 个标签。
config["metadata"]：dict，默认 {}。值须可 JSON 序列化。

chain.invoke(x, config={"tags": ["prod"], "metadata": {"tenant": "x"}})

3.4 与生产日志

Callback 内避免阻塞；重活异步写。勿在 callback 里再 invoke 同链（易递归）。最小安全 handler：只读、只 print/队列。

---

4. 最小可运行示例

pip install -U langchain-openai

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

class TokenLogHandler(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs):
        gen = response.generations[0][0]
        usage = getattr(gen.message, "usage_metadata", None)
        print("llm_end", type(response).__name__, usage)

model = ChatOpenAI(
    model="qwen3.5:9b",
    api_key="ollama",
    base_url="http://localhost:11434/v1",
    temperature=0,
)
chain = ChatPromptTemplate.from_template("只答一个词：{q}") | model | StrOutputParser()
print(chain.invoke({"q": "中国的首都"}, config={
    "callbacks": [TokenLogHandler()],
    "tags": ["demo"],
    "metadata": {"scene": "callback-tutorial"},
}))
# 预期：打印 llm_end LLMResult 与 usage（本地 Ollama 可能为 None）；答案形态含北京

LangSmith（需账号与 Key）：

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=lsv2_...
export LANGCHAIN_PROJECT=langchain-tutorial
# 再 invoke，在 smith.langchain.com 查看 trace

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点	配置指导
callbacks	list[Handler]，invoke config	给这一次 run 挂本地生命周期钩子	过长阻塞会拖垮链；异常应在 handler 内 catch	自定义 Handler	勿同步打外部 HTTP
LANGCHAIN_TRACING_V2	env，未设=关	进程级打开 LangSmith 上传	无 Key 时静默不上或 warn	true/false	生产按环境开
LANGCHAIN_PROJECT	env	trace 归入哪个 LangSmith 项目	不设则混进 default 项目	每应用一名	避免多服务共用 default
tags	list[str]，config	过滤 trace 的标签	与 metadata 互补；子步骤并集	env、版本	环境/版本分开打
metadata	dict，config	自定义可检索维度	合规审查；勿放 PII	user_id（非 PII）	只放计费/租户键
run_name	str，可选	UI 上本步显示名	只改可读性，不参与过滤	场景名	调试友好短名

---

5. 易踩坑

1. 开了 TRACING 但未设 API_KEY：静默不上传或 warn，以为没 trace。
2. callback 抛异常中断链：handler 内应 catch。
3. 本地 print 与 LangSmith 重复：生产统一走 trace，减少双份逻辑。

---

小结

• BaseCallbackHandler 挂 on_llm_ / on_chain_ / on_tool_***。
• 通过 RunnableConfig.callbacks 传入。
• LangSmith 用环境变量 + 同一套 invoke 自动追踪。
• tags/metadata 便于检索与成本分摊。

---

参考链接

• Callbacks
• Trace with LangChain
• BaseCallbackHandler