LangChain 的回调（Callback）功能

LangChain 的回调（Callback）功能是调试、监控链（Chain）应用的核心工具，它能在链执行的全生命周期（如开始/结束、模型调用、报错等）触发自定义逻辑，帮你精准定位问题、输出关键日志、统计执行耗时，是新手调试复杂链应用的“必备技能”。

一、核心概念（新手友好版）

可以把回调理解成“链执行的监听器”：

你定义一个/一组“回调函数”，告诉 LangChain：“当链执行到某个阶段（比如开始调用模型、模型返回结果）时，就执行我的这个函数”；
LangChain 会在对应的生命周期节点主动触发这些函数，你可以在函数里打印日志、记录耗时、保存数据，甚至中断执行；
核心价值：解决“黑盒调试”问题——不用改核心业务代码，就能监控链的每一步执行细节。

LangChain 内置了丰富的回调钩子（Hook），覆盖以下关键节点：

生命周期节点	作用
`on_chain_start`	链开始执行时触发
`on_chain_end`	链执行结束时触发
`on_chain_error`	链执行出错时触发
`on_llm_start`/`on_llm_end`	LLM模型开始/结束调用时触发
`on_prompt_start`	提示词模板渲染开始时触发

二、基础用法：用内置回调调试

LangChain 提供了开箱即用的 ConsoleCallbackHandler，能直接打印链执行的关键日志，无需自定义代码，是新手调试的首选。

示例：用 ConsoleCallbackHandler 调试链

from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import ConsoleCallbackHandler  # 内置控制台回调

import os
os.environ["OPENAI_API_KEY"] = "你的OpenAI API Key"

# 1. 初始化组件
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("请解释：{question}")

# 2. 自定义处理函数（模拟链的复杂逻辑）
def process_question(input_data):
    return {"question": input_data["question"].strip()}

# 3. 构建链
chain = (
    RunnableLambda(process_question)  # 步骤1：处理问题
    | RunnablePassthrough()          # 步骤2：透传
    | prompt                         # 步骤3：渲染提示词
    | llm                            # 步骤4：调用模型
)

# 4. 调用链并传入回调（关键：通过callbacks参数传入）
result = chain.invoke(
    {"question": "  RunnableParallel怎么用？  "},
    callbacks=[ConsoleCallbackHandler()]  # 传入内置控制台回调
)

print("\n最终结果：", result.content)

执行后控制台输出（关键日志）：

> Entering new RunnableLambda chain...  # 进入自定义函数链
> Finished chain.                      # 自定义函数执行完成
> Entering new RunnablePassthrough chain...
> Finished chain.
> Entering new ChatPromptTemplate chain...
> Finished chain.
> Entering new ChatOpenAI chain...
> Finished chain.

最终结果： RunnableParallel是LangChain中用于并行执行多个子组件的核心组件...

通过这些日志，你能清晰看到链的执行顺序和每个组件的执行状态，快速定位哪个环节出了问题（比如如果模型调用卡住，Entering new ChatOpenAI chain... 后不会出现 Finished chain.）。

三、进阶用法：自定义回调（精准调试）

内置回调只能打印基础日志，若需要更精细化的调试（比如记录耗时、提取模型输入输出、监控报错），可以自定义 BaseCallbackHandler 子类，重写对应钩子函数。

示例：自定义回调监控链的全生命周期

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import time
import os

os.environ["OPENAI_API_KEY"] = "你的OpenAI API Key"

# 1. 自定义回调类（继承BaseCallbackHandler）
class CustomDebugCallback(BaseCallbackHandler):
    def __init__(self):
        self.chain_start_time = {}  # 记录每个链的开始时间（处理嵌套链）
        self.llm_inputs = []        # 保存模型的输入提示词
        self.llm_outputs = []       # 保存模型的输出结果

    # 钩子1：链开始执行时触发
    def on_chain_start(self, serialized, inputs, **kwargs):
        chain_name = serialized.get("name", "未知链")
        self.chain_start_time[chain_name] = time.time()
        print(f"\n🔍 【链开始】{chain_name} | 输入：{inputs}")

    # 钩子2：链执行结束时触发
    def on_chain_end(self, serialized, outputs, **kwargs):
        chain_name = serialized.get("name", "未知链")
        cost_time = time.time() - self.chain_start_time[chain_name]
        print(f"✅ 【链结束】{chain_name} | 耗时：{cost_time:.2f}秒 | 输出：{outputs}")

    # 钩子3：LLM模型开始调用时触发
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"\n📢 【模型调用开始】| 提示词：{prompts[0]}")
        self.llm_inputs.append(prompts[0])  # 保存模型输入

    # 钩子4：LLM模型调用结束时触发
    def on_llm_end(self, response, **kwargs):
        output = response.generations[0][0].text
        print(f"📤 【模型调用结束】| 输出：{output[:50]}...")  # 只打印前50字
        self.llm_outputs.append(output)

    # 钩子5：链执行出错时触发（关键：调试报错）
    def on_chain_error(self, error, **kwargs):
        print(f"\n❌ 【链执行出错】| 错误信息：{str(error)[:100]}")

# 2. 初始化组件和链
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("请用100字解释：{question}")
chain = RunnablePassthrough() | prompt | llm

# 3. 初始化自定义回调
custom_callback = CustomDebugCallback()

# 4. 调用链（传入自定义回调）
try:
    result = chain.invoke(
        {"question": "LangChain的回调功能怎么用？"},
        callbacks=[custom_callback]  # 传入自定义回调
    )
except Exception as e:
    pass

# 5. 调试后可查看保存的模型输入输出
print("\n📋 调试总结：")
print(f"模型输入总数：{len(custom_callback.llm_inputs)}")
print(f"模型输出总数：{len(custom_callback.llm_outputs)}")

执行后输出（精准调试信息）：

🔍 【链开始】RunnablePassthrough | 输入：{'question': 'LangChain的回调功能怎么用？'}
✅ 【链结束】RunnablePassthrough | 耗时：0.00秒 | 输出：{'question': 'LangChain的回调功能怎么用？'}

🔍 【链开始】ChatPromptTemplate | 输入：{'question': 'LangChain的回调功能怎么用？'}
✅ 【链结束】ChatPromptTemplate | 耗时：0.00秒 | 输出：[HumanMessage(content='请用100字解释：LangChain的回调功能怎么用？')]

📢 【模型调用开始】| 提示词：请用100字解释：LangChain的回调功能怎么用？
📤 【模型调用结束】| 输出：LangChain的回调功能是监控链执行的监听器，可在链开始/结束、模型调用...

🔍 【链开始】ChatOpenAI | 输入：[HumanMessage(content='请用100字解释：LangChain的回调功能怎么用？')]
✅ 【链结束】ChatOpenAI | 耗时：0.85秒 | 输出：AIMessage(content='LangChain的回调功能是监控链执行的监听器，可在链开始/结束、模型调用等节点触发自定义逻辑，用于调试、日志打印、耗时统计。无需修改核心代码，通过callbacks参数传入，支持内置控制台回调和自定义回调，能精准定位链执行问题。')

📋 调试总结：
模型输入总数：1
模型输出总数：1

通过自定义回调，你能：

看到每个组件的执行耗时（比如模型调用耗时0.85秒）；
提取模型的原始输入提示词和输出结果；
捕获执行错误并打印关键信息；
保存调试数据（如输入输出列表），后续分析问题。

四、实战调试场景（新手高频问题）

场景1：调试链执行顺序/数据流转

问题：链的多个组件组合后，不知道数据是怎么传递的？
解决方案：在 on_chain_start/on_chain_end 中打印每个组件的输入/输出，清晰看到数据流转。

场景2：定位模型调用慢的问题

问题：链执行很慢，想知道是提示词渲染慢还是模型调用慢？
解决方案：在自定义回调中记录 ChatPromptTemplate 和 ChatOpenAI 的耗时，对比即可定位。

场景3：捕获链执行中的异常

问题：链偶尔报错，但不知道具体是哪个环节出错？
解决方案：重写 on_chain_error/on_llm_error，打印错误信息和出错环节的输入数据。

五、高级技巧：回调组合使用

可以同时传入多个回调（内置+自定义），满足不同调试需求：

# 同时使用内置控制台回调 + 自定义耗时统计回调
result = chain.invoke(
    {"question": "测试"},
    callbacks=[ConsoleCallbackHandler(), CustomDebugCallback()]
)

总结

核心作用：LangChain 回调是链应用的“调试监听器”，能在执行全生命周期触发自定义逻辑，解决黑盒调试问题；
基础用法：用 ConsoleCallbackHandler 快速打印执行日志，适合新手快速定位执行流程问题；
进阶用法：继承 BaseCallbackHandler 自定义回调，可监控耗时、提取模型输入输出、捕获异常，满足精准调试需求。

回调功能无需修改核心业务代码，是调试复杂链应用（如RAG、多模型链）的核心工具，掌握后能大幅提升排查问题的效率。

LangChain 的回调（Callback）功能