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 執行過程接上外部可觀測性平台。
留言
張貼留言