iT邦幫忙

2026 iThome 鐵人賽

DAY 25
0
Build on Google AI

打造專屬 AI 參謀群:以 Gemini Spark、Workspace 與 ADK 構建自動化決策雷達系列 第 25 篇

Day 25:可觀測性黑盒子:設計 Agent Trace 日誌與結構化思考過程追蹤

  • 分享至 

  • xImage
  •  

在 Day 21 到 Day 24 中,我們完成了情報從 Google Sheets 到 Obsidian Vault 以及 Notion 關聯資料庫的雙軌沉澱管線。

然而,當系統運作在 GitHub Actions 的無人值守環境時,一個嚴峻的工程挑戰浮現了:LLM 代理人管線的「思考黑盒子」問題。

當某篇論文被評估為 verified_score = 3 並遭到 ARCHIVED(封存)時,我們無從得知:

  • Reviewer(初審代理人)究竟基於哪條依據給出扣分?
  • Domain Expert 在萃取量化指標時,是否遭遇了幻覺或解析例外?
  • Synthesizer 推導二階效應的思考脈絡(Chain-of-Thought)是什麼?
  • 四大節點各自耗費了多少 Token 與執行毫秒?

在分散式與 Agent 系統中,日誌(Logs)與追蹤(Traces)是維持系統可信任度與快速定位 Bad Cases 的核心支柱。今天我們將為情報雷達設計專屬的 Agent Trace 日誌架構,實現免外部第三方平台依賴的結構化思考鏈追蹤。


一、可觀測性設計藍圖:Span 與 Trace Context

在 OpenTelemetry 的規範中,一個 Trace(呼叫鏈追蹤) 代表一筆情資從進站到出站的完整生命週期;而每個代理人的執行過程則是一個 Span(區間)。

我們在領域層為 IntelligenceItem 注入輕量級的追蹤上下文,避免引入龐大的 APM(Application Performance Monitoring)基礎設施:

Trace: entry_id="dfd9abe0"
│
├── Span 1: [Reviewer]
│   ├── Duration: 1,420ms | Tokens: 420
│   ├── Inputs: raw_content
│   ├── Decision: PASS (Score=5)
│   └── Reasoning: "確認包含完整實驗數據與多進程鎖定設計驗證..."
│
├── Span 2: [Domain Expert]
│   ├── Duration: 2,150ms | Tokens: 680
│   └── Extracted: metrics, limitations
│
├── Span 3: [Synthesizer]
│   ├── Duration: 1,890ms | Tokens: 530
│   └── Chain-of-Thought: "一階為並發控制 -> 二階推演出分散式狀態機在多代理協作之安全性..."
│
└── Span 4: [Editor]
    ├── Duration: 980ms | Tokens: 310
    └── Output: editorial_summary, tags


二、領域模型擴展:AgentSpan 與 ExecutionTrace

在 src/core/models.py 中引入追蹤模型,使每個代理人能以不可變且結構化的格式回傳執行細節:

# src/core/models.py (節錄擴展)
from datetime import datetime
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field

class AgentSpan(BaseModel):
    """單一代理人執行節點的追蹤區間 (Span)"""
    agent_name: str = Field(description="代理人名稱 (Reviewer, DomainExpert, Synthesizer, Editor)")
    started_at: datetime = Field(default_factory=datetime.utcnow)
    duration_ms: int = Field(description="執行耗時 (毫秒)")
    prompt_tokens: int = Field(default=0, description="輸入 Token 數")
    completion_tokens: int = Field(default=0, description="輸出 Token 數")
    decision: Optional[str] = Field(default=None, description="核心決策 (例如 PASS / ARCHIVED)")
    thought_process: Optional[str] = Field(default=None, description="思維鏈 (Chain-of-Thought) 摘要")
    error: Optional[str] = Field(default=None, description="例外訊息 (若執行失敗)")

class ExecutionTrace(BaseModel):
    """單筆情資處理的完整追蹤鏈 (Trace)"""
    trace_id: str
    entry_id: str
    total_duration_ms: int = 0
    spans: List[AgentSpan] = Field(default_factory=list)

並在 IntelligenceItem 的 metadata 中保留 execution_trace 鍵名,維持現有 Pydantic 模型定義向下相容。


三、裝飾器模式:Trace Context 與無侵入式攔截器

為了避免在每個代理人的業務邏輯中夾雜繁瑣的時間計算與日誌格式化代碼,我們採用 Python 裝飾器(Decorator) 模式,打造 @trace_agent_step:

# src/core/tracer.py
import functools
import logging
import time
from typing import Callable
from src.core.models import AgentSpan, IntelligenceItem

logger = logging.getLogger("agent.tracer")

def trace_agent_step(agent_name: str):
    """攔截代理人執行過程,自動記錄耗時、決策推導與思考鏈追蹤。"""
    def decorator(func: Callable):
        @functools.wraps(func)
        def wrapper(self, item: IntelligenceItem, *args, **kwargs):
            start_time = time.perf_counter()
            error_msg = None
            decision = None
            thought_process = None

            logger.info(f"[{item.entry_id}] 開始執行節點: {agent_name}")

            try:
                # 執行實際的代理人邏輯
                result_item, should_continue = func(self, item, *args, **kwargs)
                
                # 動態萃取關鍵代理人的決策與思考
                if agent_name == "Reviewer":
                    decision = f"SCORE_{result_item.verified_score}" if result_item.verified_score else "ARCHIVED"
                    thought_process = result_item.metadata.get("review_reasoning")
                elif agent_name == "Synthesizer":
                    decision = "SYNTHESIZED"
                    thought_process = result_item.cross_impact_notes
                
                return result_item, should_continue

            except Exception as exc:
                error_msg = str(exc)
                logger.error(f"[{item.entry_id}] 節點 {agent_name} 執行失敗: {exc}")
                raise exc

            finally:
                duration_ms = int((time.perf_counter() - start_time) * 1000)
                span = AgentSpan(
                    agent_name=agent_name,
                    duration_ms=duration_ms,
                    decision=decision,
                    thought_process=thought_process[:200] if thought_process else None,
                    error=error_msg
                )

                # 將 Span 寫入 IntelligenceItem 的追蹤歷史中
                if "traces" not in item.metadata:
                    item.metadata["traces"] = []
                item.metadata["traces"].append(span.model_dump(mode="json"))

                logger.info(
                    f"[{item.entry_id}] 節點 {agent_name} 完成 "
                    f"| 耗時: {duration_ms}ms "
                    f"| 決策: {decision or 'OK'} "
                    f"| 錯誤: {error_msg is not None}"
                )

        return wrapper
    return decorator


四、管線改造:在 Pipeline 注入結構化追蹤

在 src/services/pipeline.py 中,代理人方法只需掛上裝飾器,即可自動完成指標採集:

# src/services/pipeline.py (節錄)
from src.core.tracer import trace_agent_step

class ReviewerNode:
    @trace_agent_step("Reviewer")
    def evaluate(self, item: IntelligenceItem) -> tuple[IntelligenceItem, bool]:
        # 調用 LLM 評估邏輯
        ...
        return item, True

class DomainExpertNode:
    @trace_agent_step("DomainExpert")
    def extract_metrics(self, item: IntelligenceItem) -> tuple[IntelligenceItem, bool]:
        # 調用 LLM 指標萃取
        ...
        return item, True

當管線處理完成一筆情報時,item.metadata["traces"] 會完整保留每個節點的耗時、決策依據與思考鏈。


五、落盤與排查實戰:本地 JSONL 與 GitHub Actions 產出

追蹤資料收集完成後,我們提供兩種排查途徑:

1. 結構化日誌落盤(JSONL Trace Log)

在 pipeline.py 的批次結算尾端,將所有已處理情報的 Trace 聚合寫入 logs/traces_YYYY-MM-DD.jsonl:

import json
from pathlib import Path

def dump_traces(items: list[IntelligenceItem]):
    log_dir = Path("logs/traces")
    log_dir.mkdir(parents=True, exist_ok=True)
    today_str = datetime.utcnow().strftime("%Y-%m-%d")
    trace_file = log_dir / f"trace_{today_str}.jsonl"

    with trace_file.open("a", encoding="utf-8") as f:
        for item in items:
            record = {
                "entry_id": item.entry_id,
                "source_title": item.source_title,
                "status": item.status,
                "verified_score": item.verified_score,
                "traces": item.metadata.get("traces", [])
            }
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

2. GitHub Actions Artifacts 保存

在 .github/workflows/patrol.yml 中,加入保存日誌的步驟。即使 GitHub Runner 執行後銷毀,我們仍可在 GitHub 網頁上下載該次巡邏的所有 Agent 思考脈絡:

      - name: Upload Agent Execution Traces
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: agent-execution-traces
          path: logs/traces/*.jsonl
          retention-days: 7


六、實戰效果:一眼看清 Bad Case 原因

現在,當我們在 Google Sheets 或 Discord 看到某筆情資被標記為 ARCHIVED 或 verified_score = 3 時,只要開啟該筆 JSONL 紀錄:

{
  "entry_id": "4ad99832",
  "source_title": "State Locking Verification Paper",
  "status": "ARCHIVED",
  "verified_score": 3,
  "traces": [
    {
      "agent_name": "Reviewer",
      "duration_ms": 1340,
      "decision": "SCORE_3",
      "thought_process": "論文缺乏多節點並發壓測數據,僅在本地 Mock 驗證,創新性與工程實用價值不足以跨越 4 分門檻。",
      "error": null
    }
  ]
}

代理人的扣分理由一目了然,徹底告別「不知道 LLM 為何這樣判」的黑盒子窘境。


總結與下一步

今天我們建構了自主 Agent 系統的可觀測性基石:

  • 透過 AgentSpan 領域模型,規範了各代理人思考過程的標準欄位。
  • 透過 @trace_agent_step 裝飾器,以無侵入式的方式自動收集執行耗時、決策推導與思考鏈摘要。
  • 透過 JSONL 落盤與 GitHub Actions Artifacts,保證無人值守運作下的可回溯性。

然而,文字與 JSONL 檔案在快速排查時仍然偏向繁瑣。如果能在本機開啟一個輕量網頁,以時間軸與卡片形式直觀瀏覽每一次巡航的代理人決策與 Bad Cases,除錯效率將更上一層樓。

在 Day 26,我們將延伸今天的 Trace 成果:「本地可視化除錯:使用 Streamlit 快速打造 Agent Pipeline 戰情檢視器」,用不到 100 行代碼建構屬於自己的 Agent Inspector 儀表板!


上一篇
Day 24:跨平台延伸:實作 Notion API Adapter 將情報同步至 Notion 關聯資料庫
下一篇
Day 26:本地可視化除錯:使用 Streamlit 快速打造 Agent Pipeline 戰情檢視器
系列文
打造專屬 AI 參謀群:以 Gemini Spark、Workspace 與 ADK 構建自動化決策雷達 共 29 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言