跳到主要內容

AG Day 35 追蹤平台:Langfuse 觀測實戰

AG Day 35 追蹤平台:Langfuse 觀測實戰

執行需求:需外部服務帳號。AG Day 34(原文連結)把 LLM-as-judge 接上了評估流程,research-agent 現在有規則式基準評估與模型評分兩層品質檢查。但這些結果目前都只是印在終端機、跑完就消失,AG Day 9 建立的觀測紀錄也只是寫進本機的日誌檔案,沒有任何地方能把「這次執行走了哪些節點、呼叫了幾次模型、花了多少 token、評分結果如何」整合成一個可以視覺化查詢、可以跨次執行比較的畫面。今天我們要把 research-agent 接上 Langfuse——一個專門給大型語言模型應用使用的可觀測性平台,讓每一次多代理執行都留下結構化、可回溯查詢的追蹤紀錄(trace)。這篇需要一個 Langfuse 帳號(雲端版或自行架設的自架版皆可)才能看到完整效果;沒有帳號的讀者,程式一樣可以在 LANGFUSE_ENABLED=false 的情況下正常運作,只是追蹤資料不會真的被送出,僅印出本機除錯訊息供對照。

引言

回顧一下 research-agent 目前分散在各處的觀測資料:AG Day 9 建立的日誌記錄了每一步的思考與工具呼叫;AG Day 30 的交接封包記錄了每個 worker 的任務結果;AG Day 32-34 的評估流程算出了基準通過率與 LLM-as-judge 評分。這些資料各自有各自的用途,但彼此之間沒有連結——你沒辦法在一個畫面裡看到「這次執行的第三步呼叫了哪個模型、花了多少 token,而這次執行的最終報告在評估集裡得到幾分」。這正是追蹤平台要解決的問題:把一次完整的代理執行,從最外層的使用者請求,到裡面每一個節點、每一次模型呼叫、每一次工具呼叫,全部串成一棵有父子關係的樹(trace tree),並且可以把評分結果也掛在同一棵樹上,讓分析執行狀況跟分析評估結果不再是兩件互不相干的事。

今天的實作大致分成四個部分:先介紹 Langfuse 的核心概念(trace、span、generation、score);接著把 RESEARCH_AGENT_MODEL 環境變數體系擴充成正式支援 LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY;再把 AG Day 29 的多代理圖用 Langfuse 的裝飾器包起來,讓每一次 supervisor 決策與 worker 執行都自動被記錄成一個 span;最後把 AG Day 34 的評分結果回寫成 Langfuse 的 score,讓執行紀錄與評估結果在同一個畫面上呈現。

原理/觀念

Langfuse 的核心資料模型:trace、span、generation、score

Langfuse 用四種概念組織可觀測性資料。trace 是一次完整的端到端執行,例如使用者送出一個研究問題到拿到最終報告的整個過程;span 是 trace 裡的一段有開始與結束時間的子步驟,例如 supervisor 做一次路由決策,或某個 worker 執行一次任務;generation 是專門用來記錄一次模型呼叫的特殊 span,會額外記錄輸入輸出的 token 數量、使用的模型名稱;score 則是掛在某個 trace 上的評分紀錄,可以是規則式檢查的結果,也可以是 LLM-as-judge 的判斷。把這四種概念對應到 research-agent:一次 run_id 對應一個 trace,AG Day 29 每個節點的執行對應一個 span,正式模式下呼叫 llm.chat 的地方對應一個 generation,AG Day 32-34 的評估結果對應 score。

為什麼觀測要跟評估掛在一起看

如果追蹤紀錄跟評估結果分開存放,你很難回答「表現差的那幾次執行,是不是都在同一個節點卡住了」這種問題。把評分結果用 score 掛回對應的 trace 之後,Langfuse 的介面通常可以直接篩選「分數低於某個門檻的執行」,再往下鑽進那些執行的完整 span 樹,看看是哪一步出了問題。這種「先看整體分數分佈、再往下鑽進個別案例」的分析方式,是可觀測性平台相對於單純看日誌檔案最大的價值所在,日誌檔案要做同樣的分析通常得手動寫腳本去對照時間戳記,效率差很多。

離線模擬模式下的觀測策略

跟系列裡其他需要外部服務的章節一樣,我們必須讓沒有 Langfuse 帳號的讀者也能驗證程式邏輯。今天的做法是把所有 Langfuse 相關呼叫包在一層開關後面:讀取 LANGFUSE_ENABLED 這個衍生設定(由是否同時存在 LANGFUSE_PUBLIC_KEY 與 LANGFUSE_SECRET_KEY 決定),關閉時所有裝飾器與記錄函式改為印出本機除錯訊息,不對外發送任何網路請求,這樣今天所有程式碼範例都能在完全離線的情況下被驗證邏輯是否正確,只是看不到 Langfuse 網頁介面上的視覺化效果。

完整實作

需先 uv pip install langfuse,並在 Langfuse 網站(雲端版)或自架的執行個體上建立專案,取得 LANGFUSE_PUBLIC_KEY 與 LANGFUSE_SECRET_KEY:

cd research-agent
uv pip install langfuse

第一步:擴充 config.py,新增 Langfuse 相關的環境變數與衍生的 langfuse_enabled 判斷邏輯,沿用 AG Day 2 建立的「沒金鑰就自動降級」設計原則:

# research-agent/src/research_agent/config.py(擴充:Langfuse 設定)
import os
from dataclasses import dataclass


@dataclass(frozen=True)
class Settings:
    model_name: str
    max_steps: int
    is_dry_run: bool
    langfuse_public_key: str | None
    langfuse_secret_key: str | None
    langfuse_enabled: bool


def load_settings(force_dry_run: bool = False) -> "Settings":
    model = os.getenv("RESEARCH_AGENT_MODEL", "mock-model")
    max_steps = int(os.getenv("RESEARCH_AGENT_MAX_STEPS", "10"))
    has_llm_key = bool(os.getenv("OPENAI_API_KEY") or os.getenv("ANTHROPIC_API_KEY"))

    lf_public = os.getenv("LANGFUSE_PUBLIC_KEY")
    lf_secret = os.getenv("LANGFUSE_SECRET_KEY")
    langfuse_enabled = bool(lf_public and lf_secret)

    return Settings(
        model_name=model, max_steps=max_steps,
        is_dry_run=force_dry_run or (not has_llm_key),
        langfuse_public_key=lf_public, langfuse_secret_key=lf_secret,
        langfuse_enabled=langfuse_enabled,
    )

第二步:新增 observability.py,把 Langfuse 客戶端的初始化與「有沒有啟用」的判斷邏輯集中在一處,其他模組只需要呼叫這裡提供的函式,不需要各自判斷要不要送資料:

# research-agent/src/research_agent/observability.py
from research_agent.config import load_settings

_client = None


def get_langfuse_client():
    """惰性初始化 Langfuse 客戶端;未啟用時回傳 None。"""
    global _client
    settings = load_settings()
    if not settings.langfuse_enabled:
        return None
    if _client is None:
        from langfuse import Langfuse
        _client = Langfuse(public_key=settings.langfuse_public_key, secret_key=settings.langfuse_secret_key)
    return _client


def record_score(trace_id: str, name: str, value: float, comment: str = "") -> None:
    """把評估結果以 score 的形式掛回指定的 trace;未啟用 Langfuse 時只印出除錯訊息。"""
    client = get_langfuse_client()
    if client is None:
        print(f"[觀測停用] score={name} value={value} comment={comment}")
        return
    client.score(trace_id=trace_id, name=name, value=value, comment=comment)

第三步:把 AG Day 29 的 supervisor 與 worker 節點用 Langfuse 提供的裝飾器包起來,讓每個節點的執行自動變成一個 span。這裡用一個自訂的包裝函式,未啟用 Langfuse 時直接透傳,不改變任何行為:

# research-agent/src/research_agent/observability.py(追加:節點包裝器)
import functools


def traced_node(name: str):
    """把 LangGraph 節點函式包成一個 Langfuse span;未啟用時原樣執行。"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(state, *args, **kwargs):
            client = get_langfuse_client()
            if client is None:
                return func(state, *args, **kwargs)
            with client.start_as_current_span(name=name) as span:
                result = func(state, *args, **kwargs)
                span.update(input={"topic": state.get("topic")}, output=result)
                return result
        return wrapper
    return decorator

把這個裝飾器套用到既有的節點上只需要加一行:

# research-agent/src/research_agent/agents/workers.py(套用裝飾器)
from research_agent.observability import traced_node

@traced_node("search_worker")
def search_worker(state):
    ...  # 函式主體沿用 AG Day 29-30 的實作,不需要更動

第四步:正式模式下呼叫模型的地方(llm.py 的 chat 函式)也應該被記錄成 generation,而不是普通的 span,這樣 Langfuse 才能額外顯示 token 用量與模型名稱。這裡示範一個包裝函式,把既有的 chat 呼叫包一層記錄邏輯:

# research-agent/src/research_agent/observability.py(追加:模型呼叫包裝器)
def traced_generation(model_name: str, messages: list[dict], response_text: str, usage: dict) -> None:
    """把一次模型呼叫記錄成 Langfuse 的 generation,包含輸入輸出與 token 用量。"""
    client = get_langfuse_client()
    if client is None:
        print(f"[觀測停用] generation model={model_name} tokens={usage}")
        return
    client.start_generation(
        name="research-agent-chat",
        model=model_name,
        input=messages,
        output=response_text,
        usage_details=usage,
    ).end()

第五步:程式結束前務必呼叫一次 flush,確保 Langfuse SDK 背景佇列裡還沒送出的資料,在程式退出前確實送達伺服器,否則短時間執行完就結束的腳本很容易遺失最後幾筆紀錄:

# research-agent/src/research_agent/observability.py(追加:結束前送出佇列)
def flush_pending_traces() -> None:
    """在程式即將結束前呼叫,確保背景佇列裡的追蹤資料確實送出。"""
    client = get_langfuse_client()
    if client is None:
        return
    client.flush()

第六步:寫一個整合示範腳本,跑一次多代理流程,並把 AG Day 34 算出的 LLM-as-judge 分數回寫成 Langfuse 的 score:

# research-agent/scripts/run_with_tracing.py
import uuid
from research_agent.agents.graph import build_team_graph
from research_agent.evals import EvalCase, run_baseline_eval
from research_agent.observability import record_score

def main():
    team = build_team_graph()
    trace_id = str(uuid.uuid4())
    initial_state = {
        "messages": [], "topic": "Langfuse 觀測實戰",
        "task_queue": [], "completed_tasks": [], "final_report": None, "step_count": 0,
    }
    result = team.invoke(initial_state, config={"configurable": {"thread_id": trace_id}})

    case = EvalCase(case_id="trace-demo", question="Langfuse 觀測實戰",
                     required_keywords=["觀測", "追蹤"], min_citations=1)
    baseline = run_baseline_eval(case, result["final_report"], ["示範資料"], step_count=result["step_count"])
    record_score(trace_id, name="baseline_pass_rate", value=1.0 if baseline.passed_baseline else 0.0,
                 comment=f"關鍵字覆蓋率 {baseline.keyword_coverage:.2f}")

    print(f"trace_id={trace_id},基準評估:{'通過' if baseline.passed_baseline else '未通過'}")

    from research_agent.observability import flush_pending_traces
    flush_pending_traces()

if __name__ == "__main__":
    main()

執行示範(未設定 Langfuse 金鑰時的離線輸出):

cd research-agent
uv run python scripts/run_with_tracing.py

示範輸出:

[觀測停用] score=baseline_pass_rate value=1.0 comment=關鍵字覆蓋率 1.00
trace_id=1a2b3c4d-...,基準評估:通過

若設定了 LANGFUSE_PUBLIC_KEY 與 LANGFUSE_SECRET_KEY,登入 Langfuse 網頁介面就能看到這次執行的完整 span 樹(supervisor 路由了幾次、指派給哪些 worker)以及掛在上面的分數,可以依分數高低排序、篩選出表現較差的執行進一步分析,這個排序與篩選能力正是單純翻閱本機日誌檔案很難做到的事。

常見錯誤與踩雷

第一個常見錯誤,也是最容易犯的錯誤,是把 LANGFUSE_SECRET_KEY 寫進程式碼或提交進版本控制,這跟 AG Day 2 提過的金鑰外洩風險一模一樣,Langfuse 的金鑰同樣只能透過環境變數或 .env 檔案讀取,並確保 .env 已列入 .gitignore。

第二個常見錯誤是在高頻率呼叫的節點裡忘記處理 Langfuse 客戶端初始化失敗的情況(例如網路暫時不通、金鑰打錯)。今天的 get_langfuse_client 只在第一次呼叫時初始化並快取,如果初始化失敗會直接拋出例外,導致整個多代理流程被觀測層的問題拖垮。正式環境建議在初始化外面包一層 try/except,失敗時記錄警告並退回未啟用狀態,讓觀測性問題不會反過來影響核心功能的可用性——可觀測性工具本身不應該成為系統的單一失敗點。

第三個常見錯誤是把大量原始資料(例如完整的搜尋結果、超長的對話歷史)整段塞進 span 的 input/output 欄位。這樣做會讓 Langfuse 的用量與費用快速增加,也讓追蹤介面難以閱讀。比較好的做法是只記錄摘要或關鍵欄位(例如任務數量、狀態),完整資料的參照留在 knowledge.db,這跟 AG Day 31 討論檢查點狀態大小時的原則是一致的。

第四個常見錯誤是把 trace_id 跟 AG Day 31 的 thread_id 混為兩套互不相干的識別碼,事後想比對「這個檢查點對應的追蹤紀錄在哪裡」時發現完全對不起來。今天的示範刻意讓兩者共用同一個識別碼字串(run_id/trace_id/thread_id 都指向同一個值),這樣任何一個環節出問題,都能用同一個字串在三個不同的資料來源(檢查點資料庫、runs 資料表、Langfuse 介面)裡交叉查證,省去額外維護對照表的麻煩。

效能與實務提醒

Langfuse 的 SDK 通常採用非同步批次上傳的方式送出追蹤資料,不會讓每一次 span 記錄都同步等待網路回應,這對多代理系統這種本來就有不少節點呼叫的場景很重要——如果每個節點執行完都要同步等待一次網路往返才能繼續,整體延遲會明顯增加。實務上仍建議在程式結束前呼叫一次 flush(如果 SDK 提供),確保背景佇列裡的資料在程式退出前確實送出,否則短時間執行完就退出的腳本可能會遺失最後幾筆追蹤紀錄。

另外,追蹤資料量會隨著使用量快速累積,正式環境需要規劃資料保留策略(例如只保留最近三個月的詳細 trace,更早的只留彙整統計),這通常是 Langfuse 平台方案本身要考慮的用量與費用問題,需要視實際流量規模評估,這裡不做具體數字的假設。

最後,把追蹤平台導入專案的最大價值,不是「多一個看起來很專業的儀表板」,而是能把 AG Day 32-34 累積的評估工作跟真實執行資料結合起來,形成一個持續的回饋迴路:評分低的執行可以被快速定位、追查原因,改動之後再看評分是否真的提升,這正是後面 AG Day 36 討論成本與延遲最佳化時,最需要依賴的資料基礎。

值得一提的是,generation 記錄下來的 token 用量資料,其實同時服務了兩個目的:一個是今天強調的品質追蹤,另一個是接下來要談的成本控管。如果沒有從一開始就系統性地記錄每一次模型呼叫花了多少 token、用了哪個模型,事後想回頭估算「這個月大概燒了多少錢」會非常困難,只能憑印象猜測。把這件事交給 Langfuse 這種專門的可觀測性平台處理,比自己維護一份計費用的試算表要可靠得多,也更容易隨著呼叫量成長而擴展,不需要自己重新發明一套記帳邏輯。

小結

今天我們把 research-agent 接上 Langfuse 可觀測性平台:擴充 config.py 支援 LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY 兩個環境變數並提供未設定金鑰時的離線降級;新增 observability.py 集中管理客戶端初始化、節點包裝與分數回寫;把 AG Day 29 的多代理節點用裝飾器包成 span,並把 AG Day 34 的評估結果掛回對應的 trace,讓執行過程與品質評估第一次出現在同一個可查詢的畫面裡。

新增的術語:trace(一次完整端到端執行的追蹤紀錄)、span(trace 裡有時間範圍的子步驟)、generation(記錄一次模型呼叫的特殊 span)、score(掛在 trace 上的評分紀錄)、flush(把背景佇列裡尚未送出的追蹤資料強制送出的動作)。這些概念雖然是 Langfuse 的用語,但背後的思路對任何可觀測性平台都是通用的,換一家廠商也大致適用。

結語

有了追蹤平台,我們終於能把「代理跑得好不好」跟「代理跑得快不快、花費貴不貴」放在同一份資料裡一起交叉分析,不必再分頭查看好幾個不同的地方。目前 research-agent 每次執行都預設用同一個模型、沒有任何快取機制,成本與延遲完全沒有被刻意最佳化過,這在流量小的時候感覺不明顯,但只要使用量一上升,帳單金額跟使用者等待的時間都會直接反映出來,無法再假裝視而不見。

明天,我們會進入「AG Day 36 成本與延遲優化:快取與模型分級」,利用今天建立的追蹤資料,找出真正拖慢速度、燒錢的環節在哪裡,並導入回應快取與依任務難度分級選用模型的策略,在不明顯犧牲整體品質的前提下把花費壓下來。

延伸資源

  • Langfuse 官方文件:https://langfuse.com/docs。trace、span、generation、score 的完整資料模型與 Python SDK 用法請以官方文件當次版本為準。
  • Langfuse 官方文件的 Python 整合指南:說明裝飾器、上下文管理器與批次上傳(flush)機制的正確用法。
  • LangGraph 官方文件的 Observability/整合第三方追蹤工具章節:https://langchain-ai.github.io/langgraph/。討論如何把 LangGraph 執行過程接上外部可觀測性平台。

留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中,變數與資料型別是處理資料的基礎。變數是存放資料的容器,資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別,是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數,以及常見的資料型別,包括整數、浮點數、字串、布林值,還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議,以及如何用 type() 檢查資料型別。 什麼是變數?如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數,我們可以給一筆資料一個名字,並在程式的其他地方用這個名字取用該筆資料。在 Python 中,變數不需要事先宣告型別,因為 Python 是動態型別語言,變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單,只要用賦值符號 = 把值指定給變數即可。例如: x = 5 # 定義變數 x,並把整數 5 賦值給它 name = "Alice" # 定義變數 name,並把字串 "Alice" 賦值給它 在這裡,x 是一個變數,被賦予整數 5;name 是另一個變數,被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改,也就是說,我們可以在程式的不同地方給同一個變數新的值。例如: x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求,在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同,定義變數時不需要宣告型別。賦值時,Python 會根據值自動判斷變數的型別。例如: x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值,這是 Python 的彈性之一。 常見資料型別 在 Python 中,資料型別決定我們可以對變數進行哪些操作...

Python 從入門到 PyTorch 深度學習:開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習:開啟 AI 世界的大門 隨著人工智慧(AI)與深度學習(Deep Learning)快速發展,越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者,還是已經有程式基礎的開發者,學好 Python 與深度學習框架(例如 PyTorch),都能為你打開更多可能。 為什麼選擇 Python? Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手,而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化,還是建立機器學習與深度學習模型,Python 都能勝任。對想進入 AI 或資料科學領域的人來說,它幾乎是必備工具。 PyTorch 是什麼? PyTorch 是由 Meta(原 Facebook)AI 研究團隊開發的開源深度學習框架,以易用、靈活和動態計算圖著稱,是許多 AI 研究人員與開發者的首選。相較於其他框架,PyTorch 的寫法更貼近原生 Python,對初學者相對友善。無論是簡單的實驗,還是複雜的深度學習模型,PyTorch 都能提供強大的支援。 這個系列能帶給你什麼? 這個系列會從 Python 的基礎開始,帶你一步一步學習,最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式,也能跟著文章的節奏累積技能,理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎:從變數、條件判斷到函式與模組。 資料處理工具:用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化:用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎:線性代數、微積分與機率。 PyTorch 入門:理解張量、模型建構與 GPU 加速。 基礎深度學習模型:CNN 與 RNN 的實作應用。 深度學習專案實戰:從資料前處理到模型部署的端到端流程。 誰適合這個系列? 程式初學者 :如果你對 AI 充滿好奇,卻還沒寫過程式,系列的第一部分會帶你快速上手 Python,並幫助你理解深度學習的基本觀念。 資料科學愛好者 :如果你已經熟悉一些資料處理方法,進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 :想更深入了...

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡,程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者,或是想踏入人工智慧(AI)領域,學會寫程式都能明顯提升你的競爭力。在眾多程式語言中,Python 因為語法簡單、功能強大、應用範圍廣泛,成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢,並一步步教你在不同系統上安裝與設定 Python 開發環境,最後寫出第一支 Python 程式。 為什麼選擇 Python? Python 是一種高階程式語言,由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性,並用縮排來定義程式區塊,這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇;就算是經驗豐富的開發者,也能用它完成複雜的專案。 Python 的優勢如下: 簡單易學 :Python 的語法清楚、結構簡潔,初學者很快就能上手。和其他語言相比,學習曲線相對平緩,不需要先弄懂一堆複雜觀念,就能開始寫程式。 應用範圍廣泛 :從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲,Python 都有大量開源函式庫與工具支援,而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 :Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas;開發網站有 Django、Flask;做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件,讓開發更有效率。 跨平台支援 :Python 支援 Windows、macOS、Linux 等作業系統,程式通常不需要太多修改就能跨平台執行,讓開發與部署更有彈性。 活躍的社群 :Python 擁有龐大的開發者社群。學習或開發上遇到問題,幾乎都能在社群與論壇(例如 Stack Overflow)找到答案,對初學者來說是很強的後盾,也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能,讓許多領域都開始大量使用它。以下是幾個常見的應用方向: 資料科學 :隨著大數據與人工智慧興起,資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...