1. Callback 功能介紹
Callback 是 LangChain 提供的回調機制，允許我們在 LLM 應用程序的各個階段使用 hook （鉤子）。鉤子的含義也非常簡單，我們把應用程序看成一個一個的處理邏輯，從開始到結束，鉤子就是在事件傳送到終點前截獲并監控事件的傳輸。
Callback 對于記錄日志、監控、流式傳輸等任務非常有用，簡單理解， Callback 就是記錄整個流程的運行情況的一個組件，在每個關鍵的節點記錄響應的信息以便跟蹤整個應用的運行情況。
例如：
- 在 Agent 模塊中調用了幾次 tool，每次的返回值是什么？
- 在 LLM 模塊的執行輸出是什么樣的，是否有報錯？
- 在 OutputParser 模塊的輸出解析是什么樣的，重試了幾次？
Callback 收集到的信息可以直接輸出到控制臺，也可以輸出到文件，更可以輸入到第三方應用，相當于獨立的日志管理系統，通過這些日志就可以分析應用的運行情況，統計異常率，運行的瓶頸模塊以便優化。在 LangChain 中，callback 模塊中具體實現包括兩大功能，對應 CallbackHandler 和CallbackManager 。 - CallbackHandler：對每個應用場景比如 Agent 或 Chain 或 Tool 的紀錄。
- CallbackManager：對所有 CallbackHandler 的封裝和管理，包括了單個場景的 Handle，也包括運行時整條鏈路的 Handle。不過在 LangChain 的底層，這些任務的執行邏輯由回調處理器（ CallbackHandler ）定義。
CallbackHandler 里的各個鉤子函數的觸發時間如下：
以下是 LangChain Callback 事件機制中常見的事件及其對應的觸發時機和方法名稱的完整表格，適用于實現自定義的CallbackHandler：
| 事件名稱 | 事件觸發時機 | 相關方法（Callback 方法名） |
|---|---|---|
| Chat Model Start | 當聊天模型（如 ChatOpenAI）開始執行時 | on_chat_model_start |
| LLM Start | 當大語言模型（如 OpenAI、Anthropic）開始執行時 | on_llm_start |
| LLM New Token | 當 LLM 生成新 token（流式輸出）時 | on_llm_new_token |
| LLM End | 當 LLM 執行結束時 | on_llm_end |
| LLM Error | 當 LLM 執行出錯時 | on_llm_error |
| Chain Start | 當整個鏈（Chain）開始運行時 | on_chain_start |
| Chain End | 當整個鏈運行結束時 | on_chain_end |
| Chain Error | 當鏈運行出錯時 | on_chain_error |
| Tool Start | 當工具（Tool）開始執行時 | on_tool_start |
| Tool End | 當工具執行結束時 | on_tool_end |
| Tool Error | 當工具執行出錯時 | on_tool_error |
| Agent Action | 當 Agent 執行某個動作（如調用工具）時 | on_agent_action |
| Agent Finish | 當 Agent 執行完畢（完成任務）時 | on_agent_finish |
| Retriever Start | 當 Retriever（檢索器）開始工作時 | on_retriever_start |
| Retriever End | 當 Retriever 檢索結束時 | on_retriever_end |
| Retriever Error | 當 Retriever 出錯時 | on_retriever_error |
| Text | 任意文本輸出事件（用于自定義鏈、Tool、Agent 的輸出） | on_text |
| Retry | 當某個組件（如 LLM、Tool）觸發重試機制時 | on_retry |
在 LangChain 中使用回調，使用 CallbackHandler 幾種方式：
- 在運行 invoke 時傳遞對應的 config 信息配置 callbacks（推薦）。
- 在 Chain 上調用 with_config 函數，傳遞對應的 config 并配置 callbacks（推薦）。
- 在構建大語言模型時，傳遞 callbacks 參數（不推薦）。
在 LangChain 中提供了兩個最基礎的 CallbackHandler，分別是： StdOutCallbackHandler 和
FileCallbackHandler 。
使用示例如下：
import dotenv
from langchain_core.callbacks import StdOutCallbackHandler
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
dotenv.load_dotenv()
# 1.編排prompt
prompt = ChatPromptTemplate.from_template("{query}")
# 2.創建大語言模型
llm = ChatOpenAI(model="gpt-3.5-turbo-16k")
# 3.構建鏈
chain = {"query": RunnablePassthrough()} | prompt | llm | StrOutputParser()
# 4.調用鏈并執行
content = chain.stream( "你好，你是？", config={"callbacks": [StdOutCallbackHandler()]}
)
for chunk in content: pass
自定義回調
在 LangChain 中，想創建自定義回調處理器，只需繼承 BaseCallbackHandler 并實現內部的部分接口即可，例如：
#!/usr/bin/env python
# -*- coding: utf-8 -*-import time
from typing import Dict, Any, List, Optional
from uuid import UUIDimport dotenv
from langchain_core.callbacks import StdOutCallbackHandler, BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.outputs import LLMResult
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAIdotenv.load_dotenv()class LLMOpsCallbackHandler(BaseCallbackHandler):"""自定義LLMOps回調處理器"""start_at: float = 0def on_chat_model_start(self,serialized: Dict[str, Any],messages: List[List[BaseMessage]],*,run_id: UUID,parent_run_id: Optional[UUID] = None,tags: Optional[List[str]] = None,metadata: Optional[Dict[str, Any]] = None,**kwargs: Any,) -> Any:print("聊天模型開始執行了")print("serialized:", serialized)print("messages:", messages)self.start_at = time.time()def on_llm_end(self,response: LLMResult,*,run_id: UUID,parent_run_id: Optional[UUID] = None,**kwargs: Any,) -> Any:end_at: float = time.time()print("完整輸出:", response)print("程序消耗:", end_at - self.start_at)# 1.編排prompt
prompt = ChatPromptTemplate.from_template("{query}")# 2.創建大語言模型
llm = ChatOpenAI(model="gpt-3.5-turbo-16k")# 3.構建鏈
chain = {"query": RunnablePassthrough()} | prompt | llm | StrOutputParser()# 4.調用鏈并執行
resp = chain.stream("你好，你是？",config={"callbacks": [StdOutCallbackHandler(), LLMOpsCallbackHandler()]}
)for chunk in resp:pass
)
