Day 25：透明化監控！設計 Agent Trace 日誌與結構化思考過程追蹤

2026 iThome 鐵人賽

DAY 25

Build on Google AI

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

18th鐵人賽

mervyn67

2026-10-09 23:53:39

61 瀏覽

分享至

在多代理人（Multi-Agent）架構中，最讓人不安的往往不是「系統拋出異常」，而是「系統靜悄悄地跑了很久，卻不知道卡在哪個環節，或者跑完後資料沒有如期回填」。

當 Reviewer、Domain Expert、Synthesizer 與 Editor 四大 Agent 連續協同運作時，我們需要具備細粒度的可觀測性（Observability）：

每個 Agent 到底花了多少時間？
哪一個階段遭遇了 429 或 503 重試？
如何將這些思考與執行軌跡（Trace）無縫記錄到本地日誌與 Google Sheets 中？

今天我們不改動任何單一 Agent 的業務邏輯代碼，直接在中樞服務實作「無侵入式 Trace 包覆器」，並解決資料回寫時的邊界陷阱！

一、中樞切入點：無侵入式 Trace 設計

在 src/services/pipeline.py 中，我們原本就有負責防禦性重試的 _execute_stage_with_retry。

如果直接把重試邏輯替換掉，會失去網路抖動時的保護；因此最優雅的做法是裝飾器/包覆器模式（Decorator Pattern）：

內層 (_execute_stage_with_retry)：專門掛載 @retry，負責指數退避重試。
外層 (_execute_stage_with_trace)：負責在呼叫前後計算耗時、捕捉例外，並即時將 Trace 落盤。

process_item(item)
    └── _execute_stage_with_trace(stage_name, stage_fn, item)  <-- 外層：計時、捕獲異常、即時寫入 JSONL
            └── _execute_stage_with_retry(stage_fn, item)      <-- 內層：tenacity 指數退避重試
                    └── stage_fn(item)                         <-- Agent 核心執行

二、核心程式碼升級

1. 修改 `src/services/pipeline.py`

我們在 pipeline.py 中引入 _record_trace 與 _execute_stage_with_trace，即時將每一次的 Span 寫入 logs/traces/：

import json
import logging
import os
import time
from datetime import datetime, timezone

from google import genai
from tenacity import retry, stop_after_attempt, wait_exponential

from src.agents.domain_expert import DomainExpertAgent
from src.agents.editor import EditorAgent
from src.agents.reviewer import ReviewerAgent
from src.agents.synthesizer import SynthesizerAgent
from src.core.models import IntelligenceItem, ItemStatus

# 確保 logs/traces 目錄存在
os.makedirs("logs/traces", exist_ok=True)

logger = logging.getLogger(__name__)


class IntelligencePipeline:
    def __init__(self, client: genai.Client):
        self.reviewer = ReviewerAgent(client)
        self.expert = DomainExpertAgent(client)
        self.synthesizer = SynthesizerAgent(client)
        self.editor = EditorAgent(client)

    @retry(
        stop_after_attempt(3),
        wait_exponential(multiplier=2, min=3, max=15),
        reraise=True,
    )
    def _execute_stage_with_retry(self, stage_fn, item):
        """防禦性單一階段執行器（負責指數退避重試）"""
        return stage_fn(item)

    def _record_trace(self, item, stage_name: str, duration: float, error=None):
        """將單一 Agent 執行軌跡追加到 metadata 與本地日誌中"""
        span = {
            "stage": stage_name,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "duration_sec": round(duration, 2),
            "status": "ERROR" if error else "SUCCESS",
            "error": str(error) if error else None,
        }
        if "execution_trace" not in item.metadata:
            item.metadata["execution_trace"] = []
        item.metadata["execution_trace"].append(span)

        # 即時串流寫入本地 JSONL 檔案供 Streamlit Inspector 讀取
        today_str = datetime.now().strftime("%Y-%m-%d")
        trace_file = f"logs/traces/trace_{today_str}.jsonl"
        with open(trace_file, "a", encoding="utf-8") as f:
            log_entry = {
                "entry_id": item.entry_id,
                "source_title": item.source_title,
                "span": span,
            }
            f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")

    def _execute_stage_with_trace(self, stage_name: str, stage_fn, item):
        """無侵入式 Trace 包覆器"""
        start_time = time.time()
        err = None
        try:
            # 呼叫具備 tenacity 重試機制的內層執行器
            return self._execute_stage_with_retry(stage_fn, item)
        except Exception as e:
            err = e
            raise e
        finally:
            duration = time.time() - start_time
            self._record_trace(item, stage_name, duration, error=err)

    def process_item(self, item: IntelligenceItem) -> IntelligenceItem:
        """執行單一條目的完整多代理人串接管線"""
        logger.info(f"開始處理情資 [{item.entry_id}]: {item.source_title}")

        try:
            # 1. Reviewer 初審 (帶 Trace + Retry)
            item, should_continue = self._execute_stage_with_trace(
                "Reviewer", self.reviewer.evaluate, item
            )
            if not should_continue:
                logger.warning(f"條目 [{item.entry_id}] 未達門檻 (得分: {item.raw_score})，熔斷歸檔。")
                item.status = ItemStatus.ARCHIVED
                return item
            time.sleep(12)  # 均勻節奏：確保每分鐘不超過 5 次 (60s / 5 = 12s)

            # 2. Domain Expert 專家提煉 (帶 Trace + Retry)
            logger.info(f"條目 [{item.entry_id}] 通過初審，進入 Domain Expert 深入萃取。")
            item = self._execute_stage_with_trace("DomainExpert", self.expert.analyze, item)
            time.sleep(12)

            # 3. Synthesizer 跨領域推演 (帶 Trace + Retry)
            logger.info(f"條目 [{item.entry_id}] 進入 Synthesizer 進行跨領域影響推演。")
            item = self._execute_stage_with_trace("Synthesizer", self.synthesizer.synthesize, item)
            time.sleep(12)

            # 4. Editor 總編潤飾 (帶 Trace + Retry)
            logger.info(f"條目 [{item.entry_id}] 進入 Editor 進行最終摘要定稿。")
            item = self._execute_stage_with_trace("Editor", self.editor.edit, item)

            # 5. 標記處理完成
            item.status = ItemStatus.PROCESSED
            logger.info(f"條目 [{item.entry_id}] 流水線全流程執行完畢，狀態切換為 PROCESSED。")
            return item

        except Exception as e:
            err_str = str(e)
            logger.error(f"處理情資 [{item.entry_id}] 遭遇未捕獲例外: {err_str}", exc_info=True)
            item.status = ItemStatus.ERROR
            item.metadata["error_reason"] = err_str
            return item

2. 升級 `src/adapters/sheets_adapter.py`（回寫 O 欄與防禦性查詢）

為了將包含 execution_trace 的完整 metadata 回填至 Google Sheets 的第 O 欄，並徹底避免字串被 Google Sheets 自動轉型，我們對 sheets_adapter.py 進行重要補強：

    def _safe_get_all_records(self) -> list[dict]:
        """
        安全取得工作表所有紀錄
        防止像 82e93867 這類英數字串被 gspread 誤判為科學記號 (82 * 10^93867) 溢位成 inf
        """
        try:
            # gspread v6.x+ 標準寫法：忽略數值自動轉換，保留原始純字串
            return self.worksheet.get_all_records(numericise_ignore=["all"])
        except TypeError:
            try:
                # 舊版 gspread (v5.x) fallback 寫法
                return self.worksheet.get_all_records(numericise_data=False)
            except TypeError:
                return self.worksheet.get_all_records()

    def fetch_pending_items(self, auto_lock: bool = True) -> List[IntelligenceItem]:
        all_records = self._safe_get_all_records()
        pending_items: List[IntelligenceItem] = []
        rows_to_lock: List[int] = []

        for idx, record in enumerate(all_records, start=2):
            if str(record.get("status", "")).strip().upper() == ItemStatus.PENDING.value:
                raw_meta = record.get("metadata", "{}")
                metadata_dict = (
                    json.loads(raw_meta) if isinstance(raw_meta, str) and raw_meta else {}
                )
                record["metadata"] = metadata_dict

                item = DataHarmonizer.normalize(record)
                pending_items.append(item)
                rows_to_lock.append(idx)

        # 狀態鎖定：防止多 Worker 重複處理
        if auto_lock and rows_to_lock:
            cell_updates = [
                gspread.Cell(row=r, col=3, value=ItemStatus.ANALYZING.value)
                for r in rows_to_lock
            ]
            self.worksheet.update_cells(cell_updates)

        return pending_items

    def update_item_status(
        self,
        item: IntelligenceItem,
        new_status: Optional[ItemStatus] = None
    ) -> bool:
        """接收領域模型物件，自動取用其最新數據回填至 Google Sheets (包含 O 欄 metadata)"""
        status_to_write = new_status or item.status
        target_row = self._find_row_by_entry_id(item.entry_id)
        
        # ⚠️ 關鍵防護：若找不到列號必須發出錯誤警告，避免靜默失敗卡在 ANALYZING
        if not target_row:
            logger.error(f"❌ 回寫失敗：在 Google Sheet A 欄找不到對應的 entry_id [{item.entry_id}]！")
            return False

        status_val = status_to_write.value if hasattr(status_to_write, "value") else str(status_to_write)
        meta_json = json.dumps(item.metadata, ensure_ascii=False) if item.metadata else "{}"

        # 批次更新：Google Sheets API 規範單格值必須包覆為 [[value]] 二維陣列
        update_cells = [
            {"range": f"C{target_row}", "values": [[status_val]]},
            {"range": f"I{target_row}", "values": [[item.metrics or ""]]},
            {"range": f"J{target_row}", "values": [[item.limitations or ""]]},
            {"range": f"L{target_row}", "values": [[item.verified_score or item.raw_score]]},
            {"range": f"M{target_row}", "values": [[item.editorial_summary or ""]]},
            {"range": f"N{target_row}", "values": [[item.cross_impact_notes or ""]]},
            {"range": f"O{target_row}", "values": [[meta_json]]},  # <-- 加入 O 欄寫入！
        ]
        self.worksheet.batch_update(update_cells)
        logger.info(f"✅ 成功回寫 Google Sheet 第 {target_row} 列 [{item.entry_id}]")
        return True

    def _find_row_by_entry_id(self, entry_id: str) -> Optional[int]:
        """透過 entry_id (A 欄) 查找目標資料所在的實體列號"""
        try:
            entry_ids = self.worksheet.col_values(1)
            clean_id = str(entry_id).strip()
            for idx, val in enumerate(entry_ids, start=1):
                if str(val).strip() == clean_id:
                    return idx
            return None
        except Exception:
            return None

三、執行與監控成果

在根目錄直接執行排程入口：

python run_pipeline.py

1. 終端機執行軌跡

2026-10-23 18:30:18 - INFO - 成功鎖定待處理情報數量: 1
2026-10-23 18:30:18 - INFO - 開始處理情資 [82e93867]: Matrix-Game 4.0: Harnessing Video Models for Real-Time Playable Game Worlds
2026-10-23 18:32:58 - INFO - 條目 [82e93867] 通過初審，進入 Domain Expert 深入萃取。
2026-10-23 18:38:11 - INFO - 條目 [82e93867] 進入 Synthesizer 進行跨領域影響推演。
2026-10-23 18:41:36 - INFO - 條目 [82e93867] 進入 Editor 進行最終摘要定稿。
2026-10-23 18:44:48 - INFO - 條目 [82e93867] 流水線全流程執行完畢，狀態切換為 PROCESSED。
2026-10-23 18:44:49 - INFO - ✅ 成功回寫 Google Sheet 第 41 列 [82e93867]

2. 本地落盤的 JSONL Trace (`logs/traces/trace_YYYY-MM-DD.jsonl`)

{"entry_id": "82e93867", "source_title": "Matrix-Game 4.0...", "span": {"stage": "Reviewer", "timestamp": "2026-10-23T10:32:46.900913+00:00", "duration_sec": 148.64, "status": "SUCCESS", "error": null}}
{"entry_id": "82e93867", "source_title": "Matrix-Game 4.0...", "span": {"stage": "DomainExpert", "timestamp": "2026-10-23T10:37:59.115804+00:00", "duration_sec": 300.21, "status": "SUCCESS", "error": null}}
{"entry_id": "82e93867", "source_title": "Matrix-Game 4.0...", "span": {"stage": "Synthesizer", "timestamp": "2026-10-23T10:41:24.408340+00:00", "duration_sec": 193.28, "status": "SUCCESS", "error": null}}
{"entry_id": "82e93867", "source_title": "Matrix-Game 4.0...", "span": {"stage": "Editor", "timestamp": "2026-10-23T10:44:48.789274+00:00", "duration_sec": 192.37, "status": "SUCCESS", "error": null}}

打開 Google Sheet，第 41 列狀態由 ANALYZING 正式翻轉為 PROCESSED，且 O 欄已完整填入包含 execution_trace 陣列的 JSON 字串！

四、💡 實戰避坑指南 (Troubleshooting FAQ)

Q1: 為什麼日誌處理情資顯示 `[inf]` 而不是 `82e93867`？

原因：當產生的 Hash ID 恰巧符合「數字 + e + 數字」格式（例如 82e93867）時，gspread 的 get_all_records() 預設會啟動自動數值轉換，將其誤判為科學記號（$82 \times 10^{93867}$）；由於數值遠超浮點數上限，自動溢位為 float('inf')，字串化後就變成了 "inf"！
解法：在讀取工作表時使用 _safe_get_all_records()，傳入 numericise_ignore=["all"]，強制保留全欄位原始純字串格式。

Q2: 呼叫 `get_all_records()` 報錯 `TypeError: unexpected keyword argument 'numericise_data'`？

原因：gspread 在 v6.x 進行了重大改版，舊版的 numericise_data=False 參數已被廢除，改用 numericise_ignore=["all"]。
解法：使用本篇提供的 _safe_get_all_records() 輔助方法進行雙版本向下相容防護。

Q3: 程式明明跑完四大 Agent 顯示成功，但 Google Sheet 狀態卻卡在 `ANALYZING`，摘要也沒寫入？

原因：這是標準的靜默失敗。當 entry_id 誤變為 "inf" 時，update_item_status 去試算表 A 欄搜尋列號失敗回傳 None，原本程式碼直接 return False 退出而未拋出異常。
解法：在 update_item_status 補上明確的 logger.error，並配合 Q1 修正 ID 溢位問題，回寫即可百分之百命中目標列！

Q4: 寫入 Google Sheets 時出現 `400 Invalid JSON payload: expected a list of lists`？

原因：Google Sheets 官方 REST API 的 batchUpdate 規定更新值必須為二維清單 List[List[Any]]。
解法：單一儲存格的寫入必須嚴格寫成二維格式，例如 {"range": f"O{target_row}", "values": [[meta_json]]}。