LangChain 單元：Callbacks 與 Log — 深入了解 LangSmith

---

簡介

在建構 LLM（大型語言模型）應用時，觀測、除錯與效能追蹤是不可或缺的環節。
LangChain 作為串接 LLM、工具與記憶體的框架，提供了 Callback 機制與 Log 功能，讓開發者能在每一步驟中取得即時資訊。

而 LangSmith 則是 LangChain 官方推出的觀測平台，結合了 Callback、執行追蹤與可視化介面，協助團隊快速定位問題、優化流程，甚至建立自動化測試。
本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐，帶你一步步掌握 LangSmith 的使用方式，讓你的 LLM 應用從「能跑」提升到「好跑」！

---

核心概念

1. Callback 基礎

Callback 是一組在 LangChain 執行流程中 被自動呼叫的事件鉤子（hook），像是：

事件名稱	觸發時機	常見用途
on_chain_start	Chain 開始執行前	初始化記錄、設定上下文
on_chain_end	Chain 完成後	收斂結果、寫入 Log
on_tool_start	Tool（如搜尋、計算）被呼叫前	追蹤工具使用次數
on_tool_end	Tool 執行完畢	記錄回傳值、耗時
on_llm_start	LLM 被呼叫前	記錄 prompt、溫度等參數
on_llm_end	LLM 回傳結果後	捕捉 token 數、回應內容

LangSmith 內建了一套 LangChainTracer，會自動註冊上述事件，並把資料上傳至 LangSmith 伺服器，供後續分析與可視化。

2. LangSmith 的工作流程

1. 建立專案（Project）：在 LangSmith 網站先建立一個專案，用來分離不同應用或環境的資料。
2. 設定 API 金鑰：將 LANGCHAIN_API_KEY 與 LANGCHAIN_ENDPOINT（若使用自建伺服器）寫入環境變數。
3. 啟用 Tracer：在程式碼中加入 LangChainTracer，或使用 langchain.callbacks.get_openai_callback() 包裝 LLM 呼叫。
4. 執行 Chain：每一次執行都會產生一筆「Run」紀錄，包含 Prompt、LLM 回傳、使用的 Tool、耗時與 token 數等。
5. 觀測與分析：在 LangSmith UI 中可查看 Run 時間線、成本分析、失敗率，甚至對比不同模型或參數的表現。

3. Log 與自訂 Callback

雖然 LangSmith 已提供完整的追蹤功能，開發者仍可自行 擴充 Log，例如：

• 把重要的業務資訊寫入外部資料庫（MongoDB、PostgreSQL）。
• 在特定條件下發送 Slack/Teams 通知。
• 結合 A/B 測試框架，將實驗變數寫入 Run metadata。

以下示範如何 自訂 Callback，同時保留 LangSmith 的自動追蹤。

# custom_callback.py
from langchain.callbacks.base import BaseCallbackHandler
import datetime

class BusinessLogCallback(BaseCallbackHandler):
    """自訂 Callback，負責寫入業務日誌與觸發警報"""

    def __init__(self, db_client):
        self.db = db_client

    def on_chain_start(self, serialized, inputs, **kwargs):
        # 記錄 Chain 開始時間與輸入參數
        self.db.logs.insert_one({
            "type": "chain_start",
            "timestamp": datetime.datetime.utcnow(),
            "chain_id": serialized.get("id"),
            "inputs": inputs
        })

    def on_chain_end(self, outputs, **kwargs):
        # 記錄 Chain 完成與結果
        self.db.logs.insert_one({
            "type": "chain_end",
            "timestamp": datetime.datetime.utcnow(),
            "outputs": outputs
        })

    def on_llm_error(self, error, **kwargs):
        # LLM 發生錯誤時發送 Slack 通知（示意）
        import requests
        requests.post(
            "https://hooks.slack.com/services/XXX/YYY/ZZZ",
            json={"text": f"LLM 錯誤: {error}"}
        )

在主程式中同時掛載 LangSmith Tracer 與自訂 Callback：

# main.py
import os
from langchain.llms import OpenAI
from langchain.chains import LLMChain, SimpleSequentialChain
from langchain.prompts import PromptTemplate
from langchain.callbacks import get_openai_callback
from langchain.tracing import LangChainTracer
from pymongo import MongoClient
from custom_callback import BusinessLogCallback

# 設定 LangSmith 環境變數
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.langchain.com"

# 初始化 Tracer（自動送資料至 LangSmith）
tracer = LangChainTracer(project_name="客服聊天機器人")

# 建立自訂 Log Callback
mongo_client = MongoClient("mongodb://localhost:27017")
log_cb = BusinessLogCallback(db_client=mongo_client.mydb)

# 建立 LLM 與 Prompt
prompt = PromptTemplate(
    input_variables=["question"],
    template="以下是一個客戶問題，請用中文簡潔回答：{question}"
)
llm = OpenAI(model="gpt-3.5-turbo", temperature=0)

# 建立 Chain，並掛載兩個 Callback
chain = LLMChain(prompt=prompt, llm=llm, callbacks=[tracer, log_cb])

# 執行範例
with get_openai_callback() as cb:   # 取得 OpenAI 內建的成本統計
    result = chain.run({"question": "我的訂單什麼時候會到？"})
    print(result)
    print(f"使用 token 數: {cb.total_tokens}")

重點：只要把 LangChainTracer 加入 callbacks 列表，即可同時保留 LangSmith 的完整追蹤，同時加入自訂的 Log 需求。

4. 多層次 Trace：Nested Chains 與 Sub‑Runs

在大型應用中，常會把多個 Chain 串成 Pipeline，或在 Chain 內部呼叫其他 Chain、Tool。LangSmith 會自動產生 子 Run（sub‑run），形成樹狀結構，方便在 UI 中點擊展開查看每個子步驟的細節。

# nested_chain.py
from langchain.chains import SequentialChain, LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.tracing import LangChainTracer

tracer = LangChainTracer(project_name="多階段問答")

# 第一階段：分類問題類型
category_prompt = PromptTemplate(
    input_variables=["question"],
    template="請判斷以下問題屬於哪一類（訂單、付款、退貨）：{question}"
)
category_chain = LLMChain(llm=OpenAI(), prompt=category_prompt, callbacks=[tracer])

# 第二階段：根據類型產生答案
answer_prompt = PromptTemplate(
    input_variables=["category", "question"],
    template="根據類別 {category}，請用中文回答：{question}"
)
answer_chain = LLMChain(llm=OpenAI(), prompt=answer_prompt, callbacks=[tracer])

# 組合成 SequentialChain
pipeline = SequentialChain(
    chains=[category_chain, answer_chain],
    input_variables=["question"],
    output_variables=["category", "answer"],
    callbacks=[tracer]
)

result = pipeline.run({"question": "我想退貨，怎麼辦？"})
print(result)

在 LangSmith UI 中，你會看到：

Run (pipeline) ──> Sub‑run (category_chain) ──> Sub‑run (answer_chain)

每個子 Run 都保留自己的 Prompt、LLM 回傳與成本，讓你可以快速比較不同階段的表現。

---

常見陷阱與最佳實踐

陷阱	說明	最佳實踐
忘記設定 API 金鑰	程式執行時會拋出 AuthenticationError，且無法上傳 Trace。	使用 .env 檔或 CI/CD secret，確保 LANGCHAIN_API_KEY 正確載入。
過度產生子 Run	若每個 Tool 都建立獨立 Chain，會導致 Trace 數量爆炸，影響 UI 讀取速度。	合併相似工具，或在不需要細粒度追蹤時使用 disable_tracing。
Log 敏感資訊外洩	Prompt 可能包含個人資料或商業機密，直接寫入雲端會有資安風險。	在 Tracer 初始化時設定 redact_inputs=True，或自行過濾敏感欄位。
忘記關閉 OpenAI Callback	使用 get_openai_callback() 時若未使用 with 會導致資源未釋放，成本統計不完整。	總是使用 with 或手動呼叫 callback.close()。
大量同步寫入 DB	自訂 Callback 若同步寫入資料庫，可能成為瓶頸。	使用 非同步 Queue（如 Celery）或批次寫入。

進階最佳實踐

1. 分層專案管理

   • 為不同環境（dev / staging / prod）建立獨立的 LangSmith Project，避免測試資料污染正式報表。

2. 成本標籤（Cost Tags）

   • 在 Run metadata 中加入 team, feature, experiment_id 等欄位，方便在 UI 中使用篩選功能，快速定位高成本區塊。

3. 自動化測試結合 Trace

   • 使用 pytest 撰寫 LLM Unit Test，搭配 LangChainTracer 捕捉每一次測試的輸入/輸出，並在 CI 中生成測試報告。

# test_chain.py
import pytest
from langchain.tracing import LangChainTracer
from my_app.chain import pipeline

tracer = LangChainTracer(project_name="測試環境", tags={"suite": "unit_test"})

def test_order_status():
    result = pipeline.run({"question": "我的訂單編號 12345 什麼時候會到？"})
    assert "預計" in result["answer"]
    # 追蹤資料會自動送至 LangSmith，可在 UI 中檢視

4. 定期清理過期 Run
   • LangSmith 提供 Retention Policy，設定保留天數，減少儲存成本。

---

實際應用場景

場景	需求	LangSmith 如何協助
客服聊天機器人	需要即時監控錯誤、分析高頻問題、控制 token 成本	透過 Trace 觀測每筆對話的 Prompt、LLM 回覆與 token 數，結合 Dashboard 追蹤「失敗率」與「每日成本」。
金融報告生成	合規要求必須留下審計痕跡	每一次 LLM 產出都會在 LangSmith 留下完整 Run，包含模型版本、溫度、使用的工具，方便事後稽核。
內容審查系統	多階段檢測（關鍵字、情感、事實）需追蹤每一步	使用 Nested Chains，LangSmith 會自動產生子 Run，讓開發者快速定位是哪一階段導致審查失敗。
A/B 測試新提示詞	比較不同 Prompt 的效果與成本	在 Run metadata 中加入 prompt_version 標籤，利用 LangSmith UI 的分組功能直接對照。
自動化報表	每日跑 10,000 筆 LLM 請求，需監控成本與效能	設定 Retention Policy 與 Cost Alerts，當每日 token 超過門檻時自動發送 Slack 通知。

---

總結

• Callback 為 LangChain 提供了細粒度的觀測點，配合 LangSmith Tracer 能夠把每一次 LLM 呼叫、Tool 使用與 Chain 流程完整記錄下來。
• LangSmith 不只是 Log 平台，更是一個 成本分析、錯誤追蹤、A/B 測試與合規審計 的一站式解決方案。
• 透過 自訂 Callback，開發者可以在保留完整 Trace 的同時，加入業務日誌、警報或資料庫寫入等額外需求。
• 在實作時要注意 金鑰設定、敏感資訊遮蔽、子 Run 數量控制，以及 非同步寫入 的最佳化，以免影響系統效能。

掌握了 LangSmith 的觀測能力，你的 LLM 應用將不再是「黑盒」運作，而是透明、可控且具成本可視化的 智慧服務。快把本文的範例套用到自己的專案，立即體驗從測試到上線的全程可觀測！祝開發順利 🚀