LangChain 的回調(diào)（Callback）功能

LangChain 的回調(diào)（Callback）功能是調(diào)試、監(jiān)控鏈（Chain）應(yīng)用的核心工具，它能在鏈執(zhí)行的全生命周期（如開(kāi)始/結(jié)束、模型調(diào)用、報(bào)錯(cuò)等）觸發(fā)自定義邏輯，幫你精準(zhǔn)定位問(wèn)題、輸出關(guān)鍵日志、統(tǒng)計(jì)執(zhí)行耗時(shí)，是新手調(diào)試復(fù)雜鏈應(yīng)用的“必備技能”。

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

可以把回調(diào)理解成“鏈執(zhí)行的監(jiān)聽(tīng)器”：

  • 你定義一個(gè)/一組“回調(diào)函數(shù)”，告訴 LangChain：“當(dāng)鏈執(zhí)行到某個(gè)階段（比如開(kāi)始調(diào)用模型、模型返回結(jié)果）時(shí)，就執(zhí)行我的這個(gè)函數(shù)”；
  • LangChain 會(huì)在對(duì)應(yīng)的生命周期節(jié)點(diǎn)主動(dòng)觸發(fā)這些函數(shù)，你可以在函數(shù)里打印日志、記錄耗時(shí)、保存數(shù)據(jù)，甚至中斷執(zhí)行；
  • 核心價(jià)值：解決“黑盒調(diào)試”問(wèn)題——不用改核心業(yè)務(wù)代碼，就能監(jiān)控鏈的每一步執(zhí)行細(xì)節(jié)。

LangChain 內(nèi)置了豐富的回調(diào)鉤子（Hook），覆蓋以下關(guān)鍵節(jié)點(diǎn)：

生命周期節(jié)點(diǎn) 作用
on_chain_start 鏈開(kāi)始執(zhí)行時(shí)觸發(fā)
on_chain_end 鏈執(zhí)行結(jié)束時(shí)觸發(fā)
on_chain_error 鏈執(zhí)行出錯(cuò)時(shí)觸發(fā)
on_llm_start/on_llm_end LLM模型開(kāi)始/結(jié)束調(diào)用時(shí)觸發(fā)
on_prompt_start 提示詞模板渲染開(kāi)始時(shí)觸發(fā)

二、基礎(chǔ)用法：用內(nèi)置回調(diào)調(diào)試

LangChain 提供了開(kāi)箱即用的 ConsoleCallbackHandler，能直接打印鏈執(zhí)行的關(guān)鍵日志，無(wú)需自定義代碼，是新手調(diào)試的首選。

示例：用 ConsoleCallbackHandler 調(diào)試鏈

from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import ConsoleCallbackHandler  # 內(nèi)置控制臺(tái)回調(diào)

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

# 1. 初始化組件
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("請(qǐng)解釋?zhuān)簕question}")

# 2. 自定義處理函數(shù)（模擬鏈的復(fù)雜邏輯）
def process_question(input_data):
    return {"question": input_data["question"].strip()}

# 3. 構(gòu)建鏈
chain = (
    RunnableLambda(process_question)  # 步驟1：處理問(wèn)題
    | RunnablePassthrough()          # 步驟2：透?jìng)?    | prompt                         # 步驟3：渲染提示詞
    | llm                            # 步驟4：調(diào)用模型
)

# 4. 調(diào)用鏈并傳入回調(diào)（關(guān)鍵：通過(guò)callbacks參數(shù)傳入）
result = chain.invoke(
    {"question": "  RunnableParallel怎么用？  "},
    callbacks=[ConsoleCallbackHandler()]  # 傳入內(nèi)置控制臺(tái)回調(diào)
)

print("\n最終結(jié)果：", result.content)

執(zhí)行后控制臺(tái)輸出（關(guān)鍵日志）：

> Entering new RunnableLambda chain...  # 進(jìn)入自定義函數(shù)鏈
> Finished chain.                      # 自定義函數(shù)執(zhí)行完成
> Entering new RunnablePassthrough chain...
> Finished chain.
> Entering new ChatPromptTemplate chain...
> Finished chain.
> Entering new ChatOpenAI chain...
> Finished chain.

最終結(jié)果： RunnableParallel是LangChain中用于并行執(zhí)行多個(gè)子組件的核心組件...

通過(guò)這些日志，你能清晰看到鏈的執(zhí)行順序和每個(gè)組件的執(zhí)行狀態(tài)，快速定位哪個(gè)環(huán)節(jié)出了問(wèn)題（比如如果模型調(diào)用卡住，Entering new ChatOpenAI chain... 后不會(huì)出現(xiàn) Finished chain.）。

三、進(jìn)階用法：自定義回調(diào)（精準(zhǔn)調(diào)試）

內(nèi)置回調(diào)只能打印基礎(chǔ)日志，若需要更精細(xì)化的調(diào)試（比如記錄耗時(shí)、提取模型輸入輸出、監(jiān)控報(bào)錯(cuò)），可以自定義 BaseCallbackHandler 子類(lèi)，重寫(xiě)對(duì)應(yīng)鉤子函數(shù)。

示例：自定義回調(diào)監(jiān)控鏈的全生命周期

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. 自定義回調(diào)類(lèi)（繼承BaseCallbackHandler）
class CustomDebugCallback(BaseCallbackHandler):
    def __init__(self):
        self.chain_start_time = {}  # 記錄每個(gè)鏈的開(kāi)始時(shí)間（處理嵌套鏈）
        self.llm_inputs = []        # 保存模型的輸入提示詞
        self.llm_outputs = []       # 保存模型的輸出結(jié)果

    # 鉤子1：鏈開(kāi)始執(zhí)行時(shí)觸發(fā)
    def on_chain_start(self, serialized, inputs, **kwargs):
        chain_name = serialized.get("name", "未知鏈")
        self.chain_start_time[chain_name] = time.time()
        print(f"\n?? 【鏈開(kāi)始】{chain_name} | 輸入：{inputs}")

    # 鉤子2：鏈執(zhí)行結(jié)束時(shí)觸發(fā)
    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"? 【鏈結(jié)束】{chain_name} | 耗時(shí)：{cost_time:.2f}秒 | 輸出：{outputs}")

    # 鉤子3：LLM模型開(kāi)始調(diào)用時(shí)觸發(fā)
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"\n?? 【模型調(diào)用開(kāi)始】| 提示詞：{prompts[0]}")
        self.llm_inputs.append(prompts[0])  # 保存模型輸入

    # 鉤子4：LLM模型調(diào)用結(jié)束時(shí)觸發(fā)
    def on_llm_end(self, response, **kwargs):
        output = response.generations[0][0].text
        print(f"?? 【模型調(diào)用結(jié)束】| 輸出：{output[:50]}...")  # 只打印前50字
        self.llm_outputs.append(output)

    # 鉤子5：鏈執(zhí)行出錯(cuò)時(shí)觸發(fā)（關(guān)鍵：調(diào)試報(bào)錯(cuò)）
    def on_chain_error(self, error, **kwargs):
        print(f"\n? 【鏈執(zhí)行出錯(cuò)】| 錯(cuò)誤信息：{str(error)[:100]}")

# 2. 初始化組件和鏈
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("請(qǐng)用100字解釋?zhuān)簕question}")
chain = RunnablePassthrough() | prompt | llm

# 3. 初始化自定義回調(diào)
custom_callback = CustomDebugCallback()

# 4. 調(diào)用鏈（傳入自定義回調(diào)）
try:
    result = chain.invoke(
        {"question": "LangChain的回調(diào)功能怎么用？"},
        callbacks=[custom_callback]  # 傳入自定義回調(diào)
    )
except Exception as e:
    pass

# 5. 調(diào)試后可查看保存的模型輸入輸出
print("\n?? 調(diào)試總結(jié)：")
print(f"模型輸入總數(shù)：{len(custom_callback.llm_inputs)}")
print(f"模型輸出總數(shù)：{len(custom_callback.llm_outputs)}")

執(zhí)行后輸出（精準(zhǔn)調(diào)試信息）：

?? 【鏈開(kāi)始】RunnablePassthrough | 輸入：{'question': 'LangChain的回調(diào)功能怎么用？'}
? 【鏈結(jié)束】RunnablePassthrough | 耗時(shí)：0.00秒 | 輸出：{'question': 'LangChain的回調(diào)功能怎么用？'}

?? 【鏈開(kāi)始】ChatPromptTemplate | 輸入：{'question': 'LangChain的回調(diào)功能怎么用？'}
? 【鏈結(jié)束】ChatPromptTemplate | 耗時(shí)：0.00秒 | 輸出：[HumanMessage(content='請(qǐng)用100字解釋?zhuān)篖angChain的回調(diào)功能怎么用？')]

?? 【模型調(diào)用開(kāi)始】| 提示詞：請(qǐng)用100字解釋?zhuān)篖angChain的回調(diào)功能怎么用？
?? 【模型調(diào)用結(jié)束】| 輸出：LangChain的回調(diào)功能是監(jiān)控鏈執(zhí)行的監(jiān)聽(tīng)器，可在鏈開(kāi)始/結(jié)束、模型調(diào)用...

?? 【鏈開(kāi)始】ChatOpenAI | 輸入：[HumanMessage(content='請(qǐng)用100字解釋?zhuān)篖angChain的回調(diào)功能怎么用？')]
? 【鏈結(jié)束】ChatOpenAI | 耗時(shí)：0.85秒 | 輸出：AIMessage(content='LangChain的回調(diào)功能是監(jiān)控鏈執(zhí)行的監(jiān)聽(tīng)器，可在鏈開(kāi)始/結(jié)束、模型調(diào)用等節(jié)點(diǎn)觸發(fā)自定義邏輯，用于調(diào)試、日志打印、耗時(shí)統(tǒng)計(jì)。無(wú)需修改核心代碼，通過(guò)callbacks參數(shù)傳入，支持內(nèi)置控制臺(tái)回調(diào)和自定義回調(diào)，能精準(zhǔn)定位鏈執(zhí)行問(wèn)題。')

?? 調(diào)試總結(jié)：
模型輸入總數(shù)：1
模型輸出總數(shù)：1

通過(guò)自定義回調(diào)，你能：

  • 看到每個(gè)組件的執(zhí)行耗時(shí)（比如模型調(diào)用耗時(shí)0.85秒）；
  • 提取模型的原始輸入提示詞和輸出結(jié)果；
  • 捕獲執(zhí)行錯(cuò)誤并打印關(guān)鍵信息；
  • 保存調(diào)試數(shù)據(jù)（如輸入輸出列表），后續(xù)分析問(wèn)題。

四、實(shí)戰(zhàn)調(diào)試場(chǎng)景（新手高頻問(wèn)題）

場(chǎng)景1：調(diào)試鏈執(zhí)行順序/數(shù)據(jù)流轉(zhuǎn)

問(wèn)題：鏈的多個(gè)組件組合后，不知道數(shù)據(jù)是怎么傳遞的？
解決方案：在 on_chain_start/on_chain_end 中打印每個(gè)組件的輸入/輸出，清晰看到數(shù)據(jù)流轉(zhuǎn)。

場(chǎng)景2：定位模型調(diào)用慢的問(wèn)題

問(wèn)題：鏈執(zhí)行很慢，想知道是提示詞渲染慢還是模型調(diào)用慢？
解決方案：在自定義回調(diào)中記錄 ChatPromptTemplate 和 ChatOpenAI 的耗時(shí)，對(duì)比即可定位。

場(chǎng)景3：捕獲鏈執(zhí)行中的異常

問(wèn)題：鏈偶爾報(bào)錯(cuò)，但不知道具體是哪個(gè)環(huán)節(jié)出錯(cuò)？
解決方案：重寫(xiě) on_chain_error/on_llm_error，打印錯(cuò)誤信息和出錯(cuò)環(huán)節(jié)的輸入數(shù)據(jù)。

五、高級(jí)技巧：回調(diào)組合使用

可以同時(shí)傳入多個(gè)回調(diào)（內(nèi)置+自定義），滿足不同調(diào)試需求：

# 同時(shí)使用內(nèi)置控制臺(tái)回調(diào) + 自定義耗時(shí)統(tǒng)計(jì)回調(diào)
result = chain.invoke(
    {"question": "測(cè)試"},
    callbacks=[ConsoleCallbackHandler(), CustomDebugCallback()]
)

總結(jié)

  1. 核心作用：LangChain 回調(diào)是鏈應(yīng)用的“調(diào)試監(jiān)聽(tīng)器”，能在執(zhí)行全生命周期觸發(fā)自定義邏輯，解決黑盒調(diào)試問(wèn)題；
  2. 基礎(chǔ)用法：用 ConsoleCallbackHandler 快速打印執(zhí)行日志，適合新手快速定位執(zhí)行流程問(wèn)題；
  3. 進(jìn)階用法：繼承 BaseCallbackHandler 自定義回調(diào)，可監(jiān)控耗時(shí)、提取模型輸入輸出、捕獲異常，滿足精準(zhǔn)調(diào)試需求。

回調(diào)功能無(wú)需修改核心業(yè)務(wù)代碼，是調(diào)試復(fù)雜鏈應(yīng)用（如RAG、多模型鏈）的核心工具，掌握后能大幅提升排查問(wèn)題的效率。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容