跳到主要內容

AG Day 9 觀測基礎：token、延遲與成本紀錄

AG Day 9 觀測基礎：token、延遲與成本紀錄

執行需求：CPU+API key。如果暫時沒有 API key，本篇文章提供完整的離線 mock 與 dry-run 模式，不需要外部連線也能在本機完整驗證觀測資料庫建立、遙測封裝、指標計算與日誌審查流程。

引言

在上一篇 AG Day 8（原文連結）中，我們成功為「研究助理（research-agent）」建立了基於 Pydantic v2 與 JSON Schema 的強型別資料契約，消除了自然語言輸出的語法模糊性。然而，當我們開始讓代理自主執行多輪搜尋、多次呼叫工具並反覆抽取論點時，另一個更嚴肅的維運挑戰隨之而來：整個執行流程如同一個黑盒子。

傳統軟體系統在發生故障時，通常伴隨著明確的例外堆疊追蹤（Stack Trace）與 HTTP 狀態碼；但 AI Agent 的失敗往往無聲無息。例如，代理可能陷入了無效搜尋的死迴圈，在幾分鐘內發送了數十次請求，默默消耗了數十萬個權杖（token）；或者某個外部工具的網路延遲飆升，導致整個分析任務嚴重逾時，而開發者卻無法得知時間究竟是耗費在模型思考、網路傳輸還是地端資料庫查詢；更嚴重的是，如果沒有精準的計費追蹤，營運成本可能會在不知不覺中失控超支。

要將原型概念轉化為可受控、可信賴的工程系統，我們必須在系統初期就植入可觀測性（Observability）機制。可觀測性的三大基石是權杖消耗（Tokens）、回應延遲（Latency）與財務成本（Cost）。在本文中，我們將在貫穿專案 research-agent 內部建立輕量、零外部服務依賴的本機觀測架構，利用 SQLite 資料庫（data/knowledge.db）的 runs 與 events 資料表，完整記錄代理每一步的生命週期事件，為後續邁向複雜的圖編排架構提供堅實的數據支撐。

為什麼需要觀測：代理黑盒子的維運挑戰

在傳統單次請求的對話介面中，觀測通常只需要記錄一次 HTTP 請求的狀態碼與往返時間。但在以 Agent 為核心的架構中，一次使用者任務的背後往往是由數個、甚至數十個連續的推理與工具互動所構成的狀態序列。這種分散、多步且具備自律分支特性的執行模式，帶來了三大關鍵痛點：

  1. 權杖消耗的隱蔽累積：在多輪對話中，每一次將歷史訊息重新送入模型時，輸入權杖量會隨著對話輪數呈線性甚至指數級增長。若文獻檢索節點帶入了龐大的網頁內容，一次呼叫就可能吃掉數萬個輸入權杖。如果沒有逐次回合的細緻度量，工程師很難發現究竟是哪一個步驟成為了「權杖吞噬怪獸」。
  2. 非均質延遲的定位困境：代理執行總耗時往往是由多個非同步環節疊加而成：包含模型首次權杖時間（Time to First Token, TTFT）、完整權杖解碼時間、網路傳輸耗時、以及地端執行搜尋、檔案讀寫或向量資料庫檢索的時間。若只記錄端到端總時間，一旦使用者反映「系統很慢」，維運團隊將完全無法區分是模型供應商的服務降級，還是地端 SQLite 查詢遭遇到鎖定阻塞。
  3. 非確定性推論的計費審計：不同模型（例如高階推理模型與輕量化模型）在計費價格上存在顯著差異，且快取命中權杖（Cached Tokens）與常規權杖的單價亦大不相同。建立即時的成本模型，不僅能提供清楚的營運帳單，更是實施「權杖配額熔斷（Circuit Breaker）」防範死迴圈無限花錢的前提條件。

三大觀測維度：Token、延遲與成本的計量原理

要建立專業的觀測架構，首先必須釐清這三個維度的底層定義與統計方式：

第一維度：權杖（Token）計量。主流大型語言模型 API 在其回應物件中均會附帶 usage 欄位，主要包含三項指標：輸入權杖數（prompt_tokens）、輸出權杖數（completion_tokens）以及總權杖數（total_tokens）。在 2025 年之後的主流商用 API 中，提示詞快取（Prompt Caching）已成為標準能力，usage 物件通常會額外提供 prompt_tokens_details.cached_tokens 欄位。精確區分快取命中數對於成本計算與效能評估至關重要。

第二維度：延遲（Latency）度量。在 Python 中測量精確的時間差，應始終採用 time.perf_counter()，而非容易受到系統時鐘調整影響的 time.time()。單次呼叫的延遲可細分為兩部分：連線建立至首個字元抵達的延遲（反映模型排隊與上下文載入速度），以及從首字到整體串流結束的解碼延遲（反映模型的生成吞吐率）。對於非串流呼叫，我們則精確度量從發起網路請求到完整物件反序列化完成的總毫秒數（ms）。

第三維度：成本（Cost）折算。成本計算並非寫死在程式碼中的常數，而是根據當前運作模型動態載入的計價設定檔。一般以每百萬權杖（Per Million Tokens, MTok）為單位進行換算。具體計價公式為：總成本 = (未快取輸入權杖 × 輸入單價) + (快取輸入權杖 × 快取單價) + (輸出權杖 × 輸出單價)。需要強調的是，各大模型供應商的定價策略隨市場動態調整，系統應提供外部設定注入介面，具體費率一律以各家官方文件為準。

資料儲存設計：runs 與 events 的階層架構

在我們的貫穿專案 research-agent 中，資料庫統一規劃在 data/knowledge.db（SQLite）。SPEC 規格中明確定義了兩張核心觀測資料表：runs 與 events。它們構成了經典的「父任務－子事件」階層關聯：

  • runs 資料表（執行總體層級）：記錄一次完整研究任務的生命週期。欄位包含全域唯一任務識別碼（run_id）、研究主題描述（topic）、執行狀態（status：如 RUNNING、COMPLETED、FAILED）、總輸入權杖、總輸出權杖、累計成本（美元）、總延遲時間（秒）、以及建立與完成時間戳記。
  • events 資料表（細粒度步階層級）：記錄代理在該次執行中每一次與模型互動或工具調度的原子事件。欄位包含事件識別碼（event_id）、所屬任務識別碼（run_id）、執行步數（step_number）、事件型態（event_type：如 LLM_CALL、TOOL_EXECUTION）、使用模型名稱（model）、該步消耗的輸入與輸出權杖、該步耗時（毫秒）、該步計算之成本、以及儲存詳細入參和出參的 JSON 欄位（metadata_json）。

這種階層式資料模型使我們既能在巨觀上審視某次任務的總開銷，又能在微觀上逐步重播代理的思維軌跡與每一步的效能開銷。

完整實作：打造輕量級觀測封裝與事件儲存庫

現在，我們進入具體實作。我們將在 src/research_agent/ 模組目錄下，建立資料庫結構宣告、觀測記錄器以及兼具 API 與離線 dry-run 模式的度量封裝函式。

第一步，我們在 src/research_agent/storage.py 中實作資料表初始化邏輯。我們啟用 SQLite 的 WAL（Write-Ahead Logging）模式，以確保在多步驟並行記錄時具備極佳的並行寫入效能：

"""src/research_agent/storage.py：觀測資料表初始化與資料庫操作。"""
import sqlite3
from pathlib import Path

DEFAULT_DB_PATH = Path("data/knowledge.db")

def get_db_connection(db_path: Path = DEFAULT_DB_PATH) -> sqlite3.Connection:
    """取得已設定 WAL 模式與字典回傳的資料庫連線。"""
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(db_path))
    # 啟用 WAL 模式以大幅改善多工寫入效能

    conn.execute("PRAGMA journal_mode = WAL;")
    conn.row_factory = sqlite3.Row
    return conn


def init_observability_tables(conn: sqlite3.Connection) -> None:
    """初始化 runs 與 events 觀測資料表結構。"""
    with conn:
        conn.execute("""
        CREATE TABLE IF NOT EXISTS runs (
            run_id TEXT PRIMARY KEY,
            topic TEXT NOT NULL,
            status TEXT NOT NULL,
            total_prompt_tokens INTEGER DEFAULT 0,
            total_completion_tokens INTEGER DEFAULT 0,
            total_cost REAL DEFAULT 0.0,
            total_latency_seconds REAL DEFAULT 0.0,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            finished_at TIMESTAMP
        );
        """)

        conn.execute("""
        CREATE TABLE IF NOT EXISTS events (
            event_id TEXT PRIMARY KEY,
            run_id TEXT NOT NULL,
            step_number INTEGER NOT NULL,
            event_type TEXT NOT NULL,
            model TEXT,
            prompt_tokens INTEGER DEFAULT 0,
            completion_tokens INTEGER DEFAULT 0,
            latency_ms REAL NOT NULL,
            cost REAL DEFAULT 0.0,
            metadata_json TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (run_id) REFERENCES runs(run_id)
        );
        """)
        # 針對關聯查詢建立索引

        conn.execute("CREATE INDEX IF NOT EXISTS idx_events_run_id ON events(run_id);")

第二步，我們在 src/research_agent/llm.py 中建立定價字典與計費計算器。請注意，此處數值僅供工程計算演示，實務上各模型最新單價一律以官方文件為準：

"""src/research_agent/llm.py：成本模型與費率計算器。"""
from dataclasses import dataclass
from typing import Dict, Tuple

# 模型定價資料結構（單位：美元 / 每百萬權杖 MTok）

# 注意：各家模型定價會隨時間調整，實務成本一律以官方文件為準

MODEL_PRICING: Dict[str, Tuple[float, float, float]] = {
    # 格式: 模型名稱: (輸入單價, 快取輸入單價, 輸出單價)

    "gpt-4o-mini": (0.15, 0.075, 0.60),
    "gpt-4o": (2.50, 1.25, 10.00),
    "claude-3-5-sonnet": (3.00, 0.30, 15.00),
}

@dataclass
class UsageMetrics:
    """單次執行的度量統計資料封裝。"""
    prompt_tokens: int = 0
    completion_tokens: int = 0
    cached_tokens: int = 0
    latency_ms: float = 0.0
    cost_usd: float = 0.0


def calculate_cost(
    model: str,
    prompt_tokens: int,
    completion_tokens: int,
    cached_tokens: int = 0,
) -> float:
    """依據模型與權杖量計算單次呼叫的估計成本（美元）。"""
    pricing = MODEL_PRICING.get(model, (0.50, 0.25, 2.00))
    input_rate, cache_rate, output_rate = pricing

    regular_prompt_tokens = max(0, prompt_tokens - cached_tokens)
    cost = (
        (regular_prompt_tokens / 1_000_000.0) * input_rate
        + (cached_tokens / 1_000_000.0) * cache_rate
        + (completion_tokens / 1_000_000.0) * output_rate
    )
    return round(cost, 6)

第三步，實作觀測裝飾器與事件記錄器 TelemetryRecorder。該類別負責封裝與 SQLite 資料庫的溝通，並能將單次呼叫的度量即時寫入 events，同時原子性地累積至 runs 摘要：

"""實作執行生命週期與事件追蹤記錄器。"""
import uuid
import json
import sqlite3
from typing import Optional, Dict, Any

class TelemetryRecorder:
    """負責將代理執行的每一步驟寫入本地 SQLite 觀測庫。"""
    def __init__(self, conn: sqlite3.Connection, run_id: str):
        self.conn = conn
        self.run_id = run_id
        self._step_counter = 0

    def start_run(self, topic: str) -> None:
        """開啟新任務記錄。"""
        with self.conn:
            self.conn.execute(
                "INSERT INTO runs (run_id, topic, status) VALUES (?, ?, ?);",
                (self.run_id, topic, "RUNNING"),
            )

    def record_event(
        self,
        event_type: str,
        latency_ms: float,
        model: Optional[str] = None,
        prompt_tokens: int = 0,
        completion_tokens: int = 0,
        cached_tokens: int = 0,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> UsageMetrics:
        """記錄單一步階事件，並累加總指標。"""
        self._step_counter += 1
        event_id = str(uuid.uuid4())
        cost = calculate_cost(model or "default", prompt_tokens, completion_tokens, cached_tokens)
        meta_str = json.dumps(metadata or {}, ensure_ascii=False)

        with self.conn:
            # 寫入原子事件

            self.conn.execute(
                """
                INSERT INTO events (
                    event_id, run_id, step_number, event_type, model,
                    prompt_tokens, completion_tokens, latency_ms, cost, metadata_json
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
                """,
                (
                    event_id, self.run_id, self._step_counter, event_type, model,
                    prompt_tokens, completion_tokens, latency_ms, cost, meta_str
                ),
            )
            # 原子更新母任務的統計總和

            self.conn.execute(
                """
                UPDATE runs SET
                    total_prompt_tokens = total_prompt_tokens + ?,
                    total_completion_tokens = total_completion_tokens + ?,
                    total_cost = total_cost + ?,
                    total_latency_seconds = total_latency_seconds + (? / 1000.0)
                WHERE run_id = ?;
                """,
                (prompt_tokens, completion_tokens, cost, latency_ms, self.run_id),
            )

        return UsageMetrics(
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            cached_tokens=cached_tokens,
            latency_ms=latency_ms,
            cost_usd=cost,
        )

    def finish_run(self, status: str = "COMPLETED") -> None:
        """完成任務並標註時間戳記。"""
        with self.conn:
            self.conn.execute(
                "UPDATE runs SET status = ?, finished_at = CURRENT_TIMESTAMP WHERE run_id = ?;",
                (status, self.run_id),
            )

第四步，實作具備觀測能力的 LLM 呼叫包裝函式 monitored_llm_call()。我們透過 Python 高精度的 time.perf_counter() 量測耗時，並支援 --dry-run 模式，讓沒有 API key 的讀者在離線環境下也能產出逼真的度量資料：

"""實作具備自動度量採集的高階模型呼叫包裝。"""
import os
import time
from typing import Tuple, Dict, Any

def monitored_llm_call(
    prompt: str,
    system_prompt: str,
    recorder: TelemetryRecorder,
    dry_run: bool = False,
) -> Tuple[str, UsageMetrics]:
    """執行模型呼叫，並自動將 token、延遲與費用記入觀測系統。"""
    model_name = os.getenv("RESEARCH_AGENT_MODEL", "gpt-4o-mini")
    api_key = os.getenv("OPENAI_API_KEY")

    # 檢查是否走入離線 dry-run 模式

    if dry_run or not api_key:
        print("[提示] 啟用觀測離線 Mock 模式，產生模擬遙測指標。")
        simulated_start = time.perf_counter()
        # 模擬少許計算耗時

        time.sleep(0.05)
        simulated_latency = (time.perf_counter() - simulated_start) * 1000.0
        
        simulated_text = "模擬回應：已成功分析研究主題，並提取相關技術架構論點。"
        metrics = recorder.record_event(
            event_type="LLM_CALL",
            latency_ms=simulated_latency,
            model=model_name,
            prompt_tokens=420,
            completion_tokens=180,
            cached_tokens=128,
            metadata={"mode": "dry-run", "prompt_snippet": prompt[:40]},
        )
        return simulated_text, metrics

    from openai import OpenAI
    client = OpenAI(api_key=api_key)

    start_time = time.perf_counter()
    response = client.chat.completions.create(
        model=model_name,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ],
    )
    latency_ms = (time.perf_counter() - start_time) * 1000.0

    output_text = response.choices[0].message.content or ""
    usage = response.usage
    prompt_tokens = usage.prompt_tokens if usage else 0
    completion_tokens = usage.completion_tokens if usage else 0
    cached_tokens = 0
    if usage and hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details:
        cached_tokens = getattr(usage.prompt_tokens_details, "cached_tokens", 0)

    metrics = recorder.record_event(
        event_type="LLM_CALL",
        latency_ms=latency_ms,
        model=model_name,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        cached_tokens=cached_tokens,
        metadata={"prompt_snippet": prompt[:40]},
    )
    return output_text, metrics

第五步，將整套流程整合為一個多步驟的研究代理任務模擬腳本，展示從建立任務、多步推理、記錄事件到最後查詢 SQLite 匯總報表的完整工作流：

"""執行多步代理工作流並查詢觀測資料庫。"""
import uuid
from src.research_agent.storage import get_db_connection, init_observability_tables
from src.research_agent.llm import TelemetryRecorder, monitored_llm_call

# 1. 初始化本地資料庫

conn = get_db_connection()
init_observability_tables(conn)

# 2. 建立新任務記錄

task_run_id = f"run_{uuid.uuid4().hex[:8]}"
recorder = TelemetryRecorder(conn, task_run_id)
recorder.start_run(topic="評估 LangGraph 狀態檢查點的持久化效能")

print(f"=== 開始執行研究任務 [ID: {task_run_id}] ===")

# 模擬步驟一：搜尋規劃

print("\n--- 步驟 1: 擬定檢索關鍵字 ---")
_, step1_metrics = monitored_llm_call(
    prompt="請針對 LangGraph 狀態檢查點擬定三組檢索詞。",
    system_prompt="你是一位研究助理規劃專家。",
    recorder=recorder,
    dry_run=True,
)
print(f"  耗時: {step1_metrics.latency_ms:.2f} ms | 權杖: {step1_metrics.prompt_tokens} in / {step1_metrics.completion_tokens} out")
print(f"  花費: ${step1_metrics.cost_usd:.6f}")

# 模擬步驟二：文獻摘要提煉

print("\n--- 步驟 2: 提煉文獻論點 ---")
_, step2_metrics = monitored_llm_call(
    prompt="請將檢索到的資料整合為結構化論點。",
    system_prompt="你是一位嚴謹的文獻萃取助理。",
    recorder=recorder,
    dry_run=True,
)
print(f"  耗時: {step2_metrics.latency_ms:.2f} ms | 權杖: {step2_metrics.prompt_tokens} in / {step2_metrics.completion_tokens} out")
print(f"  花費: ${step2_metrics.cost_usd:.6f}")

# 3. 標註任務結束

recorder.finish_run(status="COMPLETED")

# 4. 查詢母表統計匯總

run_summary = conn.execute(
    "SELECT * FROM runs WHERE run_id = ?;", (task_run_id,)
).fetchone()

print("\n" + "=" * 20 + " 任務執行觀測報表 " + "=" * 20)
print(f"任務狀態: {run_summary['status']}")
print(f"總耗時: {run_summary['total_latency_seconds']:.3f} 秒")
print(f"累計輸入權杖: {run_summary['total_prompt_tokens']}")
print(f"累計輸出權杖: {run_summary['total_completion_tokens']}")
print(f"總估計成本: ${run_summary['total_cost']:.6f} 美元")

# 輸出（範例輸出）：

# === 開始執行研究任務 [ID: run_a1b2c3d4] ===

#

# --- 步驟 1: 擬定檢索關鍵字 ---

#   耗時: 52.30 ms | 權杖: 420 in / 180 out

#   花費: $0.000161

#

# --- 步驟 2: 提煉文獻論點 ---

#   耗時: 51.80 ms | 權杖: 420 in / 180 out

#   花費: $0.000161

#

# ==================== 任務執行觀測報表 ====================

# 任務狀態: COMPLETED

# 總耗時: 0.104 秒

# 累計輸入權杖: 840

# 累計輸出權杖: 360

# 總估計成本: $0.000322 美元

第六步，我們為觀測模組編寫一組自動化單元測試 tests/test_observability.py，使用 pytest 驗證資料表寫入正確性、權杖加總邏輯與計費公式運算：

"""tests/test_observability.py：驗證觀測資料庫與指標計算之正確性。"""
import sqlite3
import pytest
from src.research_agent.storage import init_observability_tables
from src.research_agent.llm import calculate_cost, TelemetryRecorder


@pytest.fixture
def memory_db():
    """建立暫存記憶體資料庫供單元測試獨立使用。"""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    init_observability_tables(conn)
    yield conn
    conn.close()


def test_calculate_cost_with_cache():
    """測試在快取命中時的精確計費折算。"""
    # gpt-4o-mini 定價：未快取輸入 0.15/M, 快取 0.075/M, 輸出 0.60/M

    cost = calculate_cost(
        model="gpt-4o-mini",
        prompt_tokens=1_000_000,
        completion_tokens=1_000_000,
        cached_tokens=500_000,
    )
    # (500k * 0.15) + (500k * 0.075) + (1M * 0.60) = 0.075 + 0.0375 + 0.60 = 0.7125

    assert cost == 0.7125


def test_telemetry_recorder_lifecycle(memory_db):
    """測試任務建立、事件累計與狀態結案。"""
    recorder = TelemetryRecorder(memory_db, "test_run_1")
    recorder.start_run("測試自動化檢驗主題")

    # 記錄第一個事件

    recorder.record_event(
        event_type="TOOL",
        latency_ms=120.0,
        model="gpt-4o-mini",
        prompt_tokens=200,
        completion_tokens=50,
    )
    # 記錄第二個事件

    recorder.record_event(
        event_type="LLM",
        latency_ms=280.0,
        model="gpt-4o-mini",
        prompt_tokens=300,
        completion_tokens=150,
    )
    recorder.finish_run("COMPLETED")

    # 驗證 runs 表之累計結果

    run_row = memory_db.execute("SELECT * FROM runs WHERE run_id = 'test_run_1';").fetchone()
    assert run_row["status"] == "COMPLETED"
    assert run_row["total_prompt_tokens"] == 500
    assert run_row["total_completion_tokens"] == 200
    assert run_row["total_latency_seconds"] == pytest.approx(0.4, 0.001)

    # 驗證 events 表筆數

    events_count = memory_db.execute("SELECT COUNT(*) FROM events WHERE run_id = 'test_run_1';").fetchone()[0]
    assert events_count == 2

我們同樣可以透過命令列執行單元測試，確認所有資料庫交易與計算邏輯均符合預期：

# 執行觀測機制單元測試
uv run pytest tests/test_observability.py -v

常見錯誤與踩雷

在自建 AI Agent 的觀測與計量管線時，開發者經常遭遇以下幾個典型的架構陷阱：

  1. 串流輸出時遺失權杖用量資訊：在使用 OpenAI 原生相容的串流模式（stream=True）時，預設最後一個 chunk 不會附帶 usage 物件。如果直接讀取 response.usage 會得到 None，導致日誌中的 token 全部被記為 0。要修復此問題，必須在發起請求時明確傳入參數 stream_options={"include_usage": True}，模型才會在串流的最後一個資料塊中回傳完整的權杖統計。
  2. 忽略提示詞快取折扣：現代高階模型大多導入了自動提示詞快取機制（例如 Anthropic 的 Prompt Caching 或 OpenAI 的自動快取命中）。如果一律按照標準未快取價格計算成本，統計出來的費用可能會比實際帳單高出兩到三倍。實務上必須從 prompt_tokens_details 中提取 cached_tokens 並依各家官方文件的折扣比例進行扣減。
  3. SQLite 並行寫入鎖定（database is locked）：當代理系統進入平行處理（例如多個工具並行呼叫或背景非同步處理）時，若多個執行緒直接向預設設定的 SQLite 資料庫發起寫入，極易觸發 sqlite3.OperationalError: database is locked。解決關鍵在於初始化時宣告 PRAGMA journal_mode = WAL;，並為連線設定充足的逾時時間（例如 sqlite3.connect(path, timeout=30.0)）。
  4. 誤用時鐘函式測量延遲：使用 time.time() 測量程式碼執行時間是常見的壞習慣。因為 time.time() 採用的是作業系統的日曆時鐘（Wall-clock time），一旦伺服器執行 NTP 時間校準或閏秒調整，測量出來的時間差可能會變成負數或突增。測量微秒與毫秒級延遲，務必使用單調遞增的 time.perf_counter()。

效能與實務提醒

觀測系統的存在是為了輔助核心業務，切忌讓觀測本身的開銷「喧賓奪主」：

第一點是度量寫入對代理主迴圈的阻礙。在本地開發或小規模驗證階段，直接同步寫入 SQLite 是最簡單透明的方案。但當系統擴展至高頻率並行運作時，頻繁的資料庫 I/O 會拖慢代理的整體回應速度。實務上的最佳化策略是採用記憶體非同步佇列（asyncio.Queue），由獨立的背景背景工作者（Worker）批次將日誌寫入磁碟；或者在更大型的架構中，直接將追蹤資料推送到像 Langfuse 這類專業的遠端觀測伺服器（此架構會在 AG Day 35 中深入實作）。

第二點是建立「成本熔斷防護網（Circuit Breaker）」。軟體開發中最恐怖的事故之一，是代理因為提示詞邏輯漏洞陷入無限重試迴圈，通宵執行燒光整個專案的預算配額。在 TelemetryRecorder 或代理迴圈的核心調度器中，應設定強制的硬性防護上限，例如透過環境變數 RESEARCH_AGENT_MAX_STEPS（預設限制 15 步）與 RESEARCH_AGENT_MAX_COST（例如單次任務不得超過 0.5 美元）。一旦累計數值觸及臨界值，立即中斷執行並拋出熔斷異常。

第三點是日誌記錄的隱私與敏感資料脫敏。在 metadata_json 中記錄提示詞與模型回應時，必須審查文獻中是否夾帶使用者個人識別資料（PII）或企業機密。在金融或醫療等高合規性場景中，敏感欄位在進入本地觀測庫前應執行雜湊遮罩處理，避免觀測庫反而成為資安外洩的破口。

小結

今天我們深入探討了 AI Agent 系統的觀測基石。我們從代理黑盒子的維運痛點切入，釐清了權杖、延遲與成本三大指標的計量原理與換算公式。我們在貫穿專案 research-agent 中，利用本地輕量、無依賴的 SQLite 資料庫設計了 runs 與 events 階層資料表，並透過 WAL 模式確保了高效穩定的寫入。

我們實作了兼顧精確性與離線 dry-run 相容的度量封裝函式，讓無金鑰的開發者也能順暢演練觀測流程，並提供了完善的單元測試。透過這套本機觀測機制，我們的研究助理第一次擁有了透明的心跳監視器，每一次思考與工具呼叫的資源消耗都將無所遁形。

結語

回顧至此，從 Day 1 到 Day 9，我們已經靠著純 Python 程式碼、基礎迴圈、Pydantic v2 契約以及 SQLite 觀測庫，打造出了一個功能完整的單體代理原型。我們能呼叫模型、調度工具、檢查錯誤、抽取結構化資料，還能詳細記錄每一次執行的時間與金錢成本。

然而，當我們試圖讓研究助理承擔更複雜的任務時——例如同時展開多條獨立的搜尋分支、在遇到歧異時回溯先前的狀態重新推理、或是暫停流程等待人類審查並隨後復原狀態——手刻的 while 迴圈與巨大字典狀態很快就會陷入義大利麵式的程式碼泥淖中。手刻 Agent 的架構極限究竟在哪裡？我們為什麼必須引入專業的圖編排框架？

明天，我們會進入「AG Day 10 為什麼要框架：手刻 Agent 的極限」，透過真實的工程痛點深度剖析狀態膨脹、分支控制、記憶體恢復與並行處理的極限，正式揭開從純腳本邁向 LangGraph 狀態圖編排架構的序幕。

延伸資源

  • OpenAI API 官方文件（計費與使用量說明）：https://platform.openai.com/docs/api-reference/usage。掌握最新 token 統計與 prompt caching 規格。
  • SQLite WAL（Write-Ahead Logging）官方文件：https://www.sqlite.org/wal.html。深入理解並行寫入與交易日誌機制。
  • Python 官方 time 模組性能基準指引：https://docs.python.org/3/library/time.html。理解單調時鐘 perf_counter 與系統時鐘的差異。
  • OpenTelemetry 語意慣例（Semantic Conventions for Generative AI）：https://opentelemetry.io/docs/specs/semconv/gen-ai/。現代分散式系統觀測 LLM 的國際標準規格。
標籤：

留言

張貼留言

這個網誌中的熱門文章

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 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

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 等工具能處理和分析龐...
張貼留言
閱讀完整內容

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 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容