LangChain Callback：回调与链路追踪

LLM 应用在生产环境中有一个典型的可观测性（可观测性：能够通过日志、追踪等手段了解系统内部运行状态的能力）困境：传统服务出了问题，看日志、查 trace（调用链路记录），链路清清楚楚。LLM 应用中间多了好几个非确定性的步骤——System Prompt、RAG 检索、模型推理——出了问题不追踪链路，根本没法排查。

用户反馈某个问题回答得一塌糊涂。日志里只有请求进来、响应出去，中间发生了什么完全是黑盒。是 Prompt 里的 system message 描述不对？还是 RAG 检索到了不相关的文档？还是模型本身的问题？三个环节，每个都可能出问题，但没有任何证据，只能瞎猜。

LangChain 的 Callback 系统（回调系统：在程序执行到特定节点时，自动触发预先注册的函数）是解决这个问题的基础。

1.1 Callback 系统是怎么工作的

LangChain Callback事件流程图
LangChain Callback事件流程——on_chain_start到on_chain_end的完整事件链

Callback 是事件驱动的。LangChain 在链路执行的各个节点——LLM 被调用时、链开始运行时、工具被执行时、检索器返回结果时——都会触发对应的事件。注册一个 Callback Handler，这些事件发生时就会调用对应的代码。

核心事件：

事件	触发时机
`on_llm_start`	LLM 开始处理，能看到传入的 messages
`on_llm_end`	LLM 处理完成，能看到输出内容和 token 消耗
`on_llm_error`	LLM 调用出错
`on_chain_start`	链开始执行，能看到输入
`on_chain_end`	链执行完成，能看到输出
`on_tool_start`	工具开始执行，能看到工具名和参数
`on_tool_end`	工具执行完成，能看到返回值
`on_retriever_start`	检索开始，能看到查询
`on_retriever_end`	检索完成，能看到返回的文档
`on_llm_new_token`	流式输出时，每个新 token 触发

这些事件覆盖了 LLM 应用链路的所有关键节点。

1.2 自定义 Callback：继承 BaseCallbackHandler

自定义 Callback 很简单，继承 BaseCallbackHandler，覆写需要的事件方法：

python

import time
from typing import Any, Dict, List, Optional, Union
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.outputs import LLMResult

class TimingCallbackHandler(BaseCallbackHandler):
    """记录每次 LLM 调用的 token 消耗和延迟"""

    def __init__(self):
        self._start_time: Optional[float] = None
        self.call_records = []

    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        messages: List[List[BaseMessage]],
        **kwargs: Any,
    ) -> None:
        """LLM 开始调用时，记录起始时间"""
        self._start_time = time.time()
        # 打印发给模型的消息（调试用）
        for msg_list in messages:
            for msg in msg_list:
                print(f"[LLM Input] {msg.type}: {msg.content[:100]}...")

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """LLM 调用结束时，记录延迟和 token 消耗"""
        if self._start_time is None:
            return

        elapsed = time.time() - self._start_time

        # 提取 token 使用量
        token_usage = {}
        if response.llm_output and "token_usage" in response.llm_output:
            token_usage = response.llm_output["token_usage"]

        record = {
            "elapsed_ms": round(elapsed * 1000),
            "prompt_tokens": token_usage.get("prompt_tokens", 0),
            "completion_tokens": token_usage.get("completion_tokens", 0),
            "total_tokens": token_usage.get("total_tokens", 0),
        }
        self.call_records.append(record)

        print(f"[LLM Stats] 耗时 {record['elapsed_ms']}ms，"
              f"输入 {record['prompt_tokens']} tokens，"
              f"输出 {record['completion_tokens']} tokens")

    def on_llm_error(
        self,
        error: Union[Exception, KeyboardInterrupt],
        **kwargs: Any,
    ) -> None:
        """LLM 出错时记录"""
        print(f"[LLM Error] {type(error).__name__}: {str(error)}")

使用方式：

python

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

timing_callback = TimingCallbackHandler()

llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    callbacks=[timing_callback],  # 构造时传入，全局生效
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个简洁的助手，回答控制在 50 字以内。"),
    ("human", "{question}"),
])

chain = prompt | llm | StrOutputParser()

# 调用链
result = chain.invoke({"question": "什么是 RAG？"})
print(f"\n回答：{result}")

# 查看统计
print(f"\n本次调用记录：{timing_callback.call_records}")

1.3 追踪 RAG 链路的完整示例

光追踪 LLM 调用不够，RAG 应用还要追踪检索过程。下面是一个能追踪完整 RAG 链路的 Callback：

python

import time
import json
from typing import Any, Dict, List, Optional, Sequence, Union
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.documents import Document
from langchain_core.messages import BaseMessage
from langchain_core.outputs import LLMResult

class RAGTracingCallback(BaseCallbackHandler):
    """追踪 RAG 链路：记录检索到的文档、LLM 输入输出、耗时"""

    def __init__(self, session_id: str = "default"):
        self.session_id = session_id
        self.trace = []
        self._timers = {}

    def _log(self, event: str, data: Dict[str, Any]):
        self.trace.append({
            "event": event,
            "timestamp": time.time(),
            "data": data,
        })

    def on_retriever_start(self, serialized: Dict, query: str, **kwargs):
        self._timers["retriever"] = time.time()
        self._log("retriever_start", {"query": query})
        print(f"[Retriever] 开始检索：{query}")

    def on_retriever_end(self, documents: Sequence[Document], **kwargs):
        elapsed = time.time() - self._timers.get("retriever", time.time())
        doc_summaries = [
            {
                "content_preview": doc.page_content[:100],
                "metadata": doc.metadata,
            }
            for doc in documents
        ]
        self._log("retriever_end", {
            "doc_count": len(documents),
            "elapsed_ms": round(elapsed * 1000),
            "documents": doc_summaries,
        })
        print(f"[Retriever] 检索完成，{len(documents)} 个文档，耗时 {round(elapsed*1000)}ms")

    def on_llm_start(
        self,
        serialized: Dict,
        messages: List[List[BaseMessage]],
        **kwargs,
    ):
        self._timers["llm"] = time.time()
        # 记录完整的 prompt（生产环境注意脱敏）
        prompt_preview = ""
        if messages and messages[0]:
            prompt_preview = messages[0][-1].content[:200]
        self._log("llm_start", {"prompt_preview": prompt_preview})

    def on_llm_end(self, response: LLMResult, **kwargs):
        elapsed = time.time() - self._timers.get("llm", time.time())
        token_usage = {}
        if response.llm_output:
            token_usage = response.llm_output.get("token_usage", {})

        output = ""
        if response.generations and response.generations[0]:
            output = response.generations[0][0].text[:200]

        self._log("llm_end", {
            "elapsed_ms": round(elapsed * 1000),
            "token_usage": token_usage,
            "output_preview": output,
        })
        print(f"[LLM] 生成完成，耗时 {round(elapsed*1000)}ms，"
              f"共 {token_usage.get('total_tokens', '?')} tokens")

    def print_trace(self):
        """打印完整链路追踪"""
        print(f"\n===== 链路追踪（session: {self.session_id}) =====")
        for entry in self.trace:
            print(f"\n[{entry['event']}]")
            print(json.dumps(entry['data'], ensure_ascii=False, indent=2))
        print("=" * 50)

1.4 Callback 的两种传入方式

Callback 可以在两个地方传入，效果不同：

方式一：构造时传入（全局生效）

python

# 这个 llm 的所有调用都会触发 callback
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    callbacks=[TimingCallbackHandler()],
)

这个 llm 对象的每一次调用都会触发 Callback，不管是在哪条链里用的。适合全局监控。

方式二：调用时传入（单次生效）

python

llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    # 构造时不传 callback
)

# 只有这一次调用触发 callback
result = chain.invoke(
    {"question": "什么是向量数据库？"},
    config={"callbacks": [TimingCallbackHandler()]},
)

config 参数传的 Callback 只对这次调用生效。适合按需追踪，比如只追踪某类请求，或者按用户分组追踪。

实际项目里，全局监控（延迟、token 消耗、错误率）用构造时传入，细粒度调试用调用时传入。

1.5 流式输出的 Callback

流式输出时，on_llm_new_token 会在每个新 token 生成时触发：

python

class StreamingCallback(BaseCallbackHandler):
    """处理流式输出"""

    def __init__(self):
        self.tokens = []
        self.token_count = 0

    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        """每个新 token 触发，可以用来实时推送到前端"""
        self.tokens.append(token)
        self.token_count += 1
        # 实际项目里这里可以通过 WebSocket 推送到前端
        print(token, end="", flush=True)

# 使用
streaming_callback = StreamingCallback()

llm_streaming = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    streaming=True,
    callbacks=[streaming_callback],
)

print("流式输出：", end="")
llm_streaming.invoke("用三句话解释什么是 LangChain")
print(f"\n共生成 {streaming_callback.token_count} 个 token")

注意：streaming=True 时，on_llm_end 里的 token_usage 可能不准或为空，具体看模型 API 是否在流式模式下返回 token 统计。DeepSeek 和 OpenAI 都支持，但要在请求参数里额外开启。

1.6 接入 LangSmith

如果自己写 Callback 觉得麻烦，LangSmith 是官方的链路追踪平台，设置两个环境变量就能接入：

bash

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=your_langsmith_api_key
export LANGCHAIN_PROJECT=my-rag-app  # 可选，项目名

代码完全不用改。之后每次调用 LangChain 的组件，链路数据会自动上报到 LangSmith，在 Web 界面里能看到完整的调用树：哪次 LLM 调用传了什么 Prompt、返回了什么、用了多少 token、检索器返回了哪些文档。

python

import os
from langchain_openai import ChatOpenAI

# 只需要设置环境变量，不用改任何代码
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGSMITH_API_KEY", "")
os.environ["LANGCHAIN_PROJECT"] = "my-rag-app"

llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
)

# 这次调用的完整链路会自动出现在 LangSmith 界面
result = llm.invoke("什么是向量数据库？")
print(result.content)

LangSmith 免费版足够个人项目用。生产环境可以用付费版，或者自托管（LangSmith 有开源版本）。

1.7 Callback 是生产环境必需品

没有追踪的 LLM 应用，出了问题只能靠猜。用户反馈"回答不对"，不知道是 Prompt 的问题、检索的问题还是模型的问题；token 消耗每天在增长，不知道哪条链路在吃；延迟偶尔飙高，不知道瓶颈在哪里。

接好 Callback，这些问题都有了答案。接入成本很低，LangSmith 两个环境变量搞定，自定义 Callback 也就一两百行代码。