【MLflow GenAI 教學】#02 Tracing 入門:一行程式碼開啟追蹤

測驗:MLflow Tracing 入門

共 5 題,點選答案後會立即顯示結果

1. 在生產環境中部署 MLflow Tracing 時,應該選擇哪種安裝方式?

  • A. pip install mlflow 完整版安裝
  • B. pip install mlflow-tracing 輕量版安裝
  • C. pip install mlflow-full 全功能版安裝
  • D. 同時安裝 mlflowmlflow-tracing

2. 在 MLflow 中,一個 Trace 和 Span 之間的關係是什麼?

  • A. Trace 是 Span 的子元素,一個 Span 包含多個 Trace
  • B. Trace 和 Span 是同義詞,可以互換使用
  • C. Trace 代表一次完整請求,由多個 Span 組成,Span 是追蹤的基本單位
  • D. Span 是用來追蹤網路請求,Trace 是用來追蹤函數呼叫

3. 以下哪個 SpanType 最適合用來標記向量資料庫的搜尋操作?

  • A. SpanType.LLM
  • B. SpanType.RETRIEVER
  • C. SpanType.TOOL
  • D. SpanType.CHAIN

4. 如何暫時停用 OpenAI 的自動追蹤功能?

  • A. mlflow.openai.autolog(disable=True)
  • B. mlflow.openai.disable()
  • C. mlflow.disable_tracing("openai")
  • D. mlflow.openai.autolog(enabled=False)

5. 當你需要追蹤任意程式碼區塊(而非函數)時,應該使用什麼方式?

@mlflow.trace def my_function(): # 只能追蹤整個函數 pass # 如何追蹤下面這段程式碼區塊? data = load_data() processed = process(data) save(processed)
  • A. 使用 @mlflow.trace 裝飾器包裝成函數
  • B. 使用 mlflow.start_trace() 開始追蹤
  • C. 使用 with mlflow.start_span() context manager
  • D. 使用 mlflow.trace_block() 函數

前言

在上一篇文章中,我們認識了 MLflow GenAI 的整體架構。現在,讓我們深入探討 MLflow 最強大的功能之一:Tracing(追蹤)

當你使用 AI Agent 或複雜的 LLM 應用程式時,常常會遇到這些問題:

  • 為什麼這次回應這麼慢?
  • Token 用量怎麼突然暴增?
  • 中間某個步驟到底發生了什麼事?

MLflow Tracing 就是解答這些問題的利器。它能記錄每個步驟的輸入、輸出、執行時間和 metadata,讓你完整掌握 AI 應用的執行流程。


學習目標

讀完本篇後,你將能夠:

  • 安裝並設定 MLflow Tracing 環境
  • 使用 autolog() 一行程式碼啟用自動追蹤
  • 理解 Trace 的結構:spans、attributes、context
  • 在 MLflow UI 檢視追蹤結果
  • 使用 @mlflow.trace 裝飾器手動追蹤自訂函數

1. 安裝 MLflow Tracing

MLflow 提供兩種安裝方式,根據你的使用場景選擇:

完整版安裝

pip install mlflow

完整版包含所有功能:Tracking Server、UI、Model Registry、Tracing 等。適合開發和實驗環境。

輕量版安裝(推薦用於生產環境)

pip install mlflow-tracing

mlflow-tracing 是專為生產環境設計的輕量套件,特點包括:

  • 體積比完整版小 95%
  • 依賴套件更少,部署更快
  • 適合 Docker 容器、Serverless 函數等場景
  • 支援所有追蹤功能和主流 AI 框架的自動追蹤

**注意**:mlflow-tracing 需要 Python >= 3.10,且不要與完整版 mlflow 同時安裝,以避免版本衝突。


2. 自動追蹤:一行程式碼搞定

MLflow 對主流 AI 框架提供開箱即用的自動追蹤功能。你只需要在程式開頭加一行 autolog(),就能自動記錄所有 API 呼叫。

OpenAI 自動追蹤

import mlflow
from openai import OpenAI

# 啟用 OpenAI 自動追蹤
mlflow.openai.autolog()

client = OpenAI()

# 這個呼叫會被自動追蹤
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "什麼是 MLflow?"}]
)

print(response.choices[0].message.content)
Code language: PHP (php)

Anthropic 自動追蹤

import mlflow
import anthropic

# 啟用 Anthropic 自動追蹤
mlflow.anthropic.autolog()

client = anthropic.Anthropic()

# 這個呼叫會被自動追蹤
message = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "什麼是 MLflow?"}]
)

print(message.content[0].text)
Code language: PHP (php)

支援的框架

MLflow 支援眾多 AI 框架的自動追蹤:

框架 啟用方式
OpenAI mlflow.openai.autolog()
Anthropic mlflow.anthropic.autolog()
LangChain mlflow.langchain.autolog()
LlamaIndex mlflow.llama_index.autolog()
DSPy mlflow.dspy.autolog()
AutoGen mlflow.autogen.autolog()

停用自動追蹤

如果需要暫時停用追蹤,可以傳入 disable=True

# 停用特定框架的追蹤
mlflow.openai.autolog(disable=True)

# 或停用所有自動追蹤
mlflow.autolog(disable=True)
Code language: PHP (php)

3. 理解 Trace 的結構

一個 Trace 代表一次完整的請求執行過程,由多個 Span 組成。Span 是追蹤的基本單位,記錄單一操作的詳細資訊。

Trace 與 Span 的關係

Trace (一次完整請求)
├── Span: Agent 主流程
│   ├── Span: LLM 呼叫
│   ├── Span: 工具執行
│   │   └── Span: API 呼叫
│   └── Span: LLM 呼叫(最終回應)

Span 的核心屬性

每個 Span 物件包含以下重要屬性:

屬性 說明
span_id Span 的唯一識別碼
trace_id 所屬 Trace 的識別碼
parent_id 父 Span 的識別碼(根 Span 為 None)
name Span 名稱
span_type Span 類型(如 LLM、TOOL、CHAIN)
start_time_ns 開始時間(奈秒)
end_time_ns 結束時間(奈秒)
inputs 輸入資料
outputs 輸出資料
status 狀態(OK、UNSET、ERROR)
attributes 額外的 metadata(key-value)
events 事件記錄(如例外)

SpanType 類型

MLflow 提供預定義的 Span 類型,幫助分類不同操作:

from mlflow.entities import SpanType

# 常用的 SpanType
SpanType.LLM          # LLM 呼叫
SpanType.CHAT_MODEL   # Chat 模型呼叫
SpanType.CHAIN        # 操作鏈
SpanType.AGENT        # 自主 Agent
SpanType.TOOL         # 工具執行
SpanType.RETRIEVER    # 資料檢索(向量搜尋)
SpanType.EMBEDDING    # 文字嵌入
SpanType.PARSER       # 解析器
SpanType.RERANKER     # 重排序器
SpanType.MEMORY       # 記憶體操作
SpanType.UNKNOWN      # 預設類型
Code language: CSS (css)

你也可以使用自訂字串作為 Span 類型:

@mlflow.trace(span_type="ROUTER")  # 自訂類型
def route_request(query):
    ...
Code language: PHP (php)

4. 啟動 MLflow UI 檢視追蹤結果

啟動 Tracking Server

在終端機執行:

mlflow ui

預設會在 http://localhost:5000 啟動。如果需要更換 port:

mlflow ui -p 8080

設定 Tracking URI

在程式碼中指定追蹤伺服器位置:

import mlflow

# 指向本機 Tracking Server
mlflow.set_tracking_uri("http://localhost:5000")
Code language: PHP (php)

在 UI 中檢視 Traces

  1. 開啟瀏覽器前往 http://localhost:5000
  2. 點選左側的實驗(Experiment)
  3. 切換到 Traces 標籤頁
  4. 點擊任一 Trace ID 查看詳細資訊

在 Traces 頁面,你可以:

  • 檢視每個 Span 的輸入/輸出
  • 查看執行時間和狀態
  • 使用搜尋列篩選 Traces
  • 批次刪除或編輯標籤

Jupyter Notebook 整合

如果你在 Jupyter Notebook 中開發,設定好 Tracking URI 後,Trace 會直接顯示在 notebook 輸出中:

import mlflow

mlflow.set_tracking_uri("http://localhost:5000")
mlflow.openai.autolog()

# 執行後,Trace 會直接在 notebook 中顯示
response = client.chat.completions.create(...)
Code language: PHP (php)

5. 手動追蹤:@mlflow.trace 裝飾器

當你需要追蹤自訂函數時,使用 @mlflow.trace 裝飾器:

基本用法

import mlflow

@mlflow.trace
def process_query(query: str) -> str:
    """處理使用者查詢"""
    # 你的邏輯
    result = query.upper()
    return result

# 呼叫函數時會自動建立 Trace
result = process_query("hello mlflow")
Code language: CSS (css)

自訂 Span 設定

from mlflow.entities import SpanType

@mlflow.trace(
    name="自訂名稱",           # 覆寫預設的函數名稱
    span_type=SpanType.TOOL,  # 指定 Span 類型
    attributes={"version": "1.0", "env": "dev"}  # 自訂 attributes
)
def my_tool(input_data: dict) -> dict:
    return {"processed": True, "data": input_data}
Code language: PHP (php)

巢狀追蹤

裝飾器會自動處理父子關係:

import mlflow

@mlflow.trace
def step_1(x):
    return x + 1

@mlflow.trace
def step_2(x):
    return x * 2

@mlflow.trace(name="Pipeline")
def run_pipeline(x):
    a = step_1(x)      # 子 Span
    b = step_2(a)      # 子 Span
    return b

# 執行後會產生巢狀的 Span 結構
result = run_pipeline(5)
Code language: PHP (php)

執行後的 Trace 結構:

Pipeline (root span)
├── step_1
└── step_2

動態更新 Span

在函數內部取得當前 Span 並更新:

@mlflow.trace(span_type=SpanType.LLM)
def call_llm(prompt: str):
    model_id = "gpt-4o-mini"

    # 取得當前活躍的 Span
    span = mlflow.get_current_active_span()
    span.set_attributes({"model": model_id, "temperature": 0.7})

    # 呼叫 LLM
    response = client.chat.completions.create(
        model=model_id,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.choices[0].message.content
Code language: PHP (php)

包裝外部函數

對於無法直接加裝飾器的外部函數:

import math
import mlflow

def calculate(x):
    # 包裝外部函數
    traced_sqrt = mlflow.trace(math.sqrt)
    traced_pow = mlflow.trace(math.pow)

    root = traced_sqrt(x)
    squared = traced_pow(root, 2)
    return squared
Code language: PHP (php)

6. 區塊追蹤:with mlflow.start_span()

當你需要追蹤任意程式碼區塊(不只是函數)時,使用 context manager:

基本用法

import mlflow

with mlflow.start_span(name="資料處理") as span:
    # 手動設定輸入
    span.set_inputs({"data_size": 1000})

    # 你的程式碼
    processed_data = [x * 2 for x in range(1000)]

    # 手動設定輸出
    span.set_outputs({"result_count": len(processed_data)})
Code language: PHP (php)

指定 Span 類型和屬性

from mlflow.entities import SpanType

with mlflow.start_span(
    name="向量搜尋",
    span_type=SpanType.RETRIEVER
) as span:
    span.set_inputs({"query": "MLflow 是什麼?"})
    span.set_attributes({"top_k": 5, "index": "documents"})

    # 執行向量搜尋
    results = vector_db.search(query, top_k=5)

    span.set_outputs({"documents": results})
Code language: PHP (php)

混合使用裝飾器和 Context Manager

import mlflow

@mlflow.trace
def step_a(x):
    return x + 1

@mlflow.trace
def step_b(x):
    return x * 2

def complex_workflow(data):
    with mlflow.start_span(name="Workflow") as span:
        span.set_inputs({"data": data})

        # 這些會變成子 Span
        result_a = step_a(data)
        result_b = step_b(result_a)

        span.set_outputs({"result": result_b})
        return result_b
Code language: PHP (php)

7. 進階功能

更新 Trace 標籤和預覽

@mlflow.trace
def summarize(document: str):
    # 設定 Trace 標籤
    mlflow.update_current_trace(tags={"task": "summarization"})

    # 自訂 UI 預覽文字(避免顯示過長的內容)
    mlflow.update_current_trace(
        request_preview=f"Document: {document[:50]}...",
        response_preview="Summary generated"
    )

    summary = f"Summary of: {document[:100]}..."
    return summary
Code language: PHP (php)

非同步函數追蹤

從 MLflow 2.16.0 開始,裝飾器完整支援 async/await:

import mlflow
import asyncio

@mlflow.trace
async def async_process(data):
    await asyncio.sleep(1)  # 模擬非同步操作
    return data * 2

async def main():
    result = await async_process(10)
    print(result)

asyncio.run(main())
Code language: PHP (php)

Generator 函數追蹤

從 MLflow 2.20.2 開始支援 generator:

@mlflow.trace
def stream_response():
    for i in range(5):
        yield f"chunk_{i}"

# 追蹤會記錄所有 yield 的值
for chunk in stream_response():
    print(chunk)
Code language: PHP (php)

多執行緒追蹤

多執行緒環境需要手動傳遞 context:

import mlflow
import contextvars
from concurrent.futures import ThreadPoolExecutor

@mlflow.trace
def process_item(item):
    return item * 2

@mlflow.trace
def parallel_process(items):
    # 複製當前 context
    ctx = contextvars.copy_context()

    with ThreadPoolExecutor(max_workers=4) as executor:
        # 在 context 中執行
        results = list(executor.map(
            lambda x: ctx.run(process_item, x),
            items
        ))

    return results
Code language: PHP (php)

8. 完整範例:RAG Pipeline 追蹤

讓我們用一個實際的 RAG(Retrieval-Augmented Generation)範例來展示如何組合使用這些追蹤功能:

import mlflow
from mlflow.entities import SpanType
from openai import OpenAI

# 啟用自動追蹤
mlflow.openai.autolog()
mlflow.set_tracking_uri("http://localhost:5000")

client = OpenAI()

# 模擬向量資料庫
documents = [
    "MLflow 是一個開源的機器學習平台。",
    "MLflow Tracing 提供 LLM 可觀測性。",
    "MLflow 支援實驗追蹤和模型管理。"
]

@mlflow.trace(span_type=SpanType.RETRIEVER)
def retrieve_documents(query: str, top_k: int = 2) -> list:
    """簡化的檢索函數"""
    # 實際應用中這裡會是向量搜尋
    span = mlflow.get_current_active_span()
    span.set_attributes({"top_k": top_k})
    return documents[:top_k]

@mlflow.trace(span_type=SpanType.CHAIN)
def build_prompt(query: str, context: list) -> str:
    """建構 prompt"""
    context_text = "\n".join(context)
    return f"""根據以下資訊回答問題:

資訊:
{context_text}

問題:{query}

請用繁體中文回答:"""

@mlflow.trace(name="RAG Pipeline", span_type=SpanType.AGENT)
def rag_query(query: str) -> str:
    """完整的 RAG Pipeline"""
    mlflow.update_current_trace(tags={"pipeline": "rag", "version": "1.0"})

    # 1. 檢索相關文件
    docs = retrieve_documents(query)

    # 2. 建構 prompt
    prompt = build_prompt(query, docs)

    # 3. 呼叫 LLM(會被 autolog 自動追蹤)
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )

    return response.choices[0].message.content

# 執行並追蹤
if __name__ == "__main__":
    result = rag_query("什麼是 MLflow?")
    print(result)
Code language: PHP (php)

執行後,在 MLflow UI 中你會看到完整的 Trace 結構:

RAG Pipeline (AGENT)
├── retrieve_documents (RETRIEVER)
├── build_prompt (CHAIN)
└── ChatCompletion (LLM) - 由 autolog 自動產生

重點回顧

概念 說明
安裝 pip install mlflowpip install mlflow-tracing(輕量版)
自動追蹤 mlflow.openai.autolog() / mlflow.anthropic.autolog()
手動追蹤 @mlflow.trace 裝飾器
區塊追蹤 with mlflow.start_span() context manager
UI 啟動 mlflow ui 後前往 http://localhost:5000
Span 類型 SpanType.LLMSpanType.TOOLSpanType.AGENT

下一步

現在你已經掌握了 MLflow Tracing 的基礎用法。在下一篇文章中,我們將深入探討如何使用 MLflow Evaluate 來評估你的 GenAI 應用品質,包括設定評估指標、建立測試資料集,以及解讀評估結果。


參考資源

進階測驗:MLflow Tracing 入門

測驗目標:驗證你是否能在實際情境中應用所學。
共 5 題,包含情境題與錯誤診斷題。

1. 你正在開發一個 RAG 應用程式,需要追蹤向量資料庫的搜尋操作。你想要在 Span 中記錄搜尋的 top_k 參數和使用的索引名稱。應該怎麼做? 情境題

@mlflow.trace(span_type=SpanType.RETRIEVER) def search_documents(query: str, top_k: int = 5): # 你的向量搜尋邏輯 results = vector_db.search(query, top_k=top_k) return results
  • A. 在裝飾器的 attributes 參數中直接設定 @mlflow.trace(attributes={"top_k": top_k})
  • B. 在函數內使用 mlflow.get_current_active_span().set_attributes({"top_k": top_k, "index": "documents"})
  • C. 使用 mlflow.log_params({"top_k": top_k}) 記錄參數
  • D. 在函數內使用 mlflow.update_current_trace(attributes={"top_k": top_k})

2. 你的團隊正在建立一個 AI Agent 系統,需要追蹤完整的執行流程。你有三個步驟:retrieve_context、build_prompt、call_llm,希望它們在 Trace 中顯示為巢狀結構。應該如何設計? 情境題

  • A. 每個函數都使用 @mlflow.trace,它們會自動平行顯示在同一層
  • B. 需要手動使用 parent_span 參數來設定父子關係
  • C. 在主函數上使用 @mlflow.trace,在主函數內呼叫其他有 @mlflow.trace 裝飾的函數,MLflow 會自動處理父子關係
  • D. 必須使用 with mlflow.start_span() 才能建立巢狀結構

3. 你需要在 AWS Lambda 函數中部署 MLflow Tracing。考慮到 Lambda 的冷啟動時間和套件大小限制,你應該如何安裝 MLflow? 情境題

  • A. 使用 pip install mlflow-tracing,因為它是專為生產環境設計的輕量套件,體積小 95%
  • B. 使用 pip install mlflow 完整版,確保所有功能都可用
  • C. 同時安裝 mlflowmlflow-tracing 以獲得最佳相容性
  • D. Lambda 不支援 MLflow Tracing,需要使用其他追蹤方案

4. 小明在多執行緒環境中使用 MLflow Tracing,但發現子執行緒的 Span 沒有正確關聯到父 Span。以下是他的程式碼,問題出在哪裡? 錯誤診斷

import mlflow from concurrent.futures import ThreadPoolExecutor @mlflow.trace def process_item(item): return item * 2 @mlflow.trace def parallel_process(items): with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(process_item, items)) return results # 執行後,process_item 的 Span 沒有正確顯示為 parallel_process 的子 Span
  • A. 應該在 process_item 函數上使用 @mlflow.trace(parent="parallel_process")
  • B. 多執行緒環境需要手動複製 context 並在 context 中執行,使用 contextvars.copy_context()
  • C. ThreadPoolExecutor 不支援 MLflow Tracing,應該改用 multiprocessing
  • D. 需要在每個子執行緒中重新呼叫 mlflow.set_tracking_uri()

5. 小華安裝了 MLflow 並嘗試使用自動追蹤,但執行後在 UI 中看不到任何 Trace。以下是她的程式碼,最可能的問題是什麼? 錯誤診斷

import mlflow from openai import OpenAI client = OpenAI() mlflow.openai.autolog() response = client.chat.completions.create( model=”gpt-4o-mini”, messages=[{“role”: “user”, “content”: “Hello!”}] ) print(response.choices[0].message.content)
  • A. mlflow.openai.autolog() 需要傳入 enabled=True 參數
  • B. 需要在 autolog() 之前先建立 OpenAI client
  • C. mlflow.openai.autolog() 應該在建立 OpenAI client 之前呼叫,這樣才能正確 patch
  • D. 需要使用 mlflow.start_run() 才能啟用追蹤

發佈留言

發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *