LangChain 單元：Callbacks 與 Log – LCEL Callbacks 完全指南

---

簡介

在使用 LangChain（尤其是 LCEL：LangChain Expression Language）建構大型語言模型（LLM）應用時，Callbacks 是不可或缺的觀測與除錯機制。
它讓開發者可以在每個鏈（Chain）或工具（Tool）執行的關鍵節點上「掛鉤」自訂程式碼，從而取得執行時間、輸入輸出、錯誤資訊，甚至即時寫入日誌（Log）或觸發外部服務。

沒有良好的 Callback 監控，開發者往往只能靠 print 或手動追蹤，當系統規模擴大、併發請求激增時，問題排查會變得相當痛苦。
本篇文章將以 LangChain Python（同樣概念適用於 langchainjs）為例，深入說明 LCEL 中的 Callback 機制、實作方式與最佳實踐，協助你在真實專案中快速定位瓶頸、記錄關鍵資訊，並提升系統的可觀測性與維護性。

---

核心概念

1️⃣ Callback 基礎

LangChain 中的 Callback 其實是一組 事件（event） 的集合，主要包括：

事件名稱	觸發時機	常見用途
on_chain_start	鏈開始執行前	記錄開始時間、輸入參數
on_chain_end	鏈執行結束後	記錄結束時間、輸出結果、耗時
on_tool_start	工具（Tool）執行前	監控外部 API 呼叫
on_tool_end	工具執行後	捕捉回傳值、錯誤
on_llm_start	LLM 呼叫前	記錄 prompt、溫度等設定
on_llm_end	LLM 回傳後	取得模型回應、token 數量
on_error	任意階段發生例外	錯誤上報、重試機制

重點：所有事件都會收到一個 run_id，用來在日誌或追蹤系統中串聯同一次請求的多個階段。

2️⃣ LCEL 與 Callback 的結合

LCEL 允許開發者以類似 DSL 的方式組合 LLM、工具與記憶體（Memory），而每個組件背後都會自動觸發上述 Callback。
只要在建立 Runnable（可執行物件）時傳入自訂的 CallbackManager，所有子元件都會共用同一套 Callback 設定。

from langchain.callbacks.base import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 建立一個只輸出到標準輸出的 Callback Manager
callback_manager = CallbackManager([StreamingStdOutCallbackHandler()])

3️⃣ 實作自訂 Callback

LangChain 提供了 BaseCallbackHandler 抽象類別，開發者只需要繼承它並實作感興趣的事件即可。以下是一個簡易的 JSON Log 實作範例：

# file: my_callback.py
import json
import time
from langchain.callbacks.base import BaseCallbackHandler

class JsonLogCallbackHandler(BaseCallbackHandler):
    """將每個事件以 JSON 格式寫入檔案，適合集中式日誌系統"""

    def __init__(self, logfile: str = "lc_callbacks.log"):
        self.logfile = logfile

    def _write(self, event: str, payload: dict):
        entry = {
            "timestamp": time.time(),
            "event": event,
            **payload,
        }
        with open(self.logfile, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    def on_chain_start(self, serialized, inputs, **kwargs):
        self._write("chain_start", {"inputs": inputs, "run_id": kwargs.get("run_id")})

    def on_chain_end(self, outputs, **kwargs):
        self._write("chain_end", {"outputs": outputs, "run_id": kwargs.get("run_id")})

    def on_llm_start(self, serialized, prompts, **kwargs):
        self._write("llm_start", {"prompts": prompts, "run_id": kwargs.get("run_id")})

    def on_llm_end(self, response, **kwargs):
        self._write("llm_end", {"response": response, "run_id": kwargs.get("run_id")})

    def on_error(self, error, **kwargs):
        self._write("error", {"error": str(error), "run_id": kwargs.get("run_id")})

小技巧：如果你使用雲端日誌服務（如 Datadog、Logflare），只要把 _write 改成 HTTP POST，即可即時上報。

4️⃣ 在 LCEL 中使用 Callback

假設我們要建立一個簡單的 問答 流程：
Prompt → LLM → Tool（搜尋） → LLM，整個流程使用 LCEL 寫成 Runnable，並套用剛剛的 JsonLogCallbackHandler。

from langchain.llms import OpenAI
from langchain.tools import DuckDuckGoSearchRun
from langchain.prompts import PromptTemplate
from langchain.schema.runnable import RunnablePassthrough, RunnableLambda
from my_callback import JsonLogCallbackHandler
from langchain.callbacks.base import CallbackManager

# 1. 建立 LLM 與工具
llm = OpenAI(model="gpt-3.5-turbo", temperature=0.7)
search_tool = DuckDuckGoSearchRun()

# 2. 建立 Prompt
prompt = PromptTemplate.from_template(
    "請根據以下搜尋結果回答問題：\n\n{search_results}\n\n問題：{question}"
)

# 3. 建立 Callback Manager
callback_manager = CallbackManager([JsonLogCallbackHandler()])

# 4. 用 LCEL 組合 Runnable
#   - 輸入: {"question": "..."}
#   - 輸出: LLM 最終回應
workflow = (
    RunnablePassthrough()
    .assign(
        # 先呼叫搜尋工具，將結果存入 search_results
        search_results=lambda x: search_tool.run(x["question"])
    )
    .assign(
        # 把搜尋結果塞進 Prompt，交給 LLM
        answer=RunnableLambda(
            lambda x: llm(prompt.format(**x))
        )
    )
    .pick("answer")
)

# 5. 執行時帶入 Callback Manager
result = workflow.invoke(
    {"question": "台北 2024 年的天氣趨勢如何？"},
    config={"callbacks": callback_manager}
)

print("最終答案:", result)

說明：

1. RunnablePassthrough() 直接把原始輸入傳下去。
2. .assign() 兩次分別呼叫 搜尋工具 與 LLM，每一次都會觸發對應的 on_tool_*、on_llm_* 事件。
3. config={"callbacks": callback_manager} 把自訂的 Callback 注入整條鏈，所有子元件自動共用。

5️⃣ 多層 Callback 組合

在大型專案中，往往需要 不同層級 的 Callback：

• 全局層：負責寫入集中式日誌、追蹤每個請求的 run_id。
• 局部層：僅在特定工具或鏈上記錄更細節的資訊（例如 API 回傳的 HTTP 狀態碼）。

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 全局：寫入檔案
global_cb = CallbackManager([JsonLogCallbackHandler("global.log")])

# 局部：即時印出 LLM 回傳的文字
local_cb = CallbackManager([StreamingStdOutCallbackHandler()])

# 在特定工具上使用局部 Callback
search_tool = DuckDuckGoSearchRun(callbacks=local_cb)

# 其餘部分仍使用全局 Callback
workflow = (
    RunnablePassthrough()
    .assign(search_results=lambda x: search_tool.run(x["question"]))
    .assign(answer=RunnableLambda(lambda x: llm(prompt.format(**x))))
    .pick("answer")
)

result = workflow.invoke(
    {"question": "什麼是量子糾纏？"},
    config={"callbacks": global_cb}
)

技巧：CallbackManager 允許在 Runnable 建構時直接傳入 callbacks= 參數，這樣即使在全局配置了 global_cb，局部的 local_cb 仍會覆寫或補足。

---

常見陷阱與最佳實踐

陷阱	可能的後果	解決方案
忘記傳入 config={"callbacks": …}	事件不會被觸發，日誌缺失	務必在 invoke 或 stream 時傳入 Callback Manager
在高併發環境下使用同步檔案寫入	I/O 阻塞，吞吐量下降	使用非同步寫入或把日誌推送至外部服務（如 Kafka、Pub/Sub）
Callback 內部拋出例外	整條鏈中斷，難以追蹤根因	在自訂 Handler 中使用 try/except 包裹所有邏輯，並在 on_error 中記錄
過度記錄大量 Prompt	日誌檔案爆炸、隱私風險	只在開發/除錯階段啟用，或在生產環境使用 遮蔽（masking） 方式
忘記釋放資源（如檔案句柄、網路連線）	記憶體泄漏、連線枯竭	在 CallbackManager 結束時呼叫 handler.close()（若有）或使用 with 句法管理

最佳實踐

1. 統一 run_id：在入口（API、CLI）層面產生唯一 run_id，並透過 config={"run_id": …} 傳入，確保所有日誌能夠串聯。
2. 分層日誌：全局日誌負責 概覽（開始、結束、錯誤），局部日誌負責 細節（API 回傳、token 數）。
3. 使用結構化日誌：JSON、Protobuf 或 Cloud Logging 支援的格式，方便後續搜尋與分析。
4. 遮蔽敏感資訊：在 on_llm_start 時將 prompt 中的個人資料或 API 金鑰做遮蔽（例如 ***），避免資料外洩。
5. 整合 APM（Application Performance Monitoring）：將 on_chain_start / on_chain_end 的時間差上報給 New Relic、Datadog 等，以即時監控耗時瓶頸。

---

實際應用場景

場景	為何需要 Callback	典型實作
客服聊天機器人	需要追蹤每一次客戶詢問、LLM 回覆與外部資料查詢時間，判斷是否需要升級人工客服。	在 on_chain_end 記錄總耗時，若超過 3 秒則觸發警示。
金融報告自動生成	合規要求必須保存所有 Prompt 與模型回應的完整紀錄，以備審計。	使用 JsonLogCallbackHandler 把 on_llm_end 資料寫入加密的 S3 Bucket。
搜尋加強的 RAG 系統	需要觀測向量搜尋、重排與 LLM 整合的每一步，以優化召回率。	在 on_tool_start/on_tool_end 捕捉向量檢索的 top_k、相似度分數，寫入 ElasticSearch。
批次資料標註	大量資料跑同一條 LCEL 流程，需統計成功率與失敗原因。	on_error 把失敗的 input、error 寫入錯誤表格（如 Google Sheet），方便後續手動修正。
多模型路由	系統根據不同需求選擇不同 LLM（如 GPT‑4、Claude），必須記錄哪個模型被使用。	在 on_llm_start 中加入 model_name 欄位，並在日誌中標註。

---

總結

• Callbacks 是 LangChain（特別是 LCEL）提供的強大觀測機制，讓開發者能在每個執行階段取得完整的輸入、輸出與耗時資訊。
• 只要 建立 CallbackManager，並在 invoke/stream 時傳入，即可自動為所有子元件掛鉤事件。
• 透過 自訂 BaseCallbackHandler，你可以把日誌寫入檔案、資料庫、或外部觀測平台，並在 on_error 中捕捉例外、執行重試。
• 最佳實踐 包括統一 run_id、分層結構化日誌、遮蔽敏感資訊，以及避免同步 I/O 阻塞。
• 在 客服、金融、RAG、批次標註 等實務場景中，合理運用 Callback 能顯著提升除錯效率、合規性與系統可觀測性。

掌握了 LCEL 的 Callback 機制，你就能在開發 LLM 應用時，像在傳統軟體開發中使用 log 與 監控 那樣，隨時掌握系統內部的每一次「呼喊」與「回應」。祝你在 LangChain 的旅程中，寫出更可靠、更易維護的智能應用！ 🚀