在 Day 25 中,我們為四大代理人(Reviewer、Domain Expert、Synthesizer、Editor)掛載了無侵入式的追蹤裝飾器 @trace_agent_step,將每筆情報的思考鏈(Chain-of-Thought)、決策依據、執行耗時與 Token 消耗以結構化的 JSONL 格式落盤保存。
然而,當系統累積了數十場巡航、上百條 Trace 紀錄時,光靠 cat、grep 或 VS Code 打開幾千行的 JSON 文字逐行翻找,排查效率依然低下。尤其是當我們需要對初審被封存(ARCHIVED)的 Bad Case 進行根因分析,或是向團隊展示多代理人協同推演過程時,純文字日誌顯得缺乏直觀性。
今天我們將利用 Python 生態圈最敏捷的資料視覺化框架——Streamlit,以不到 100 行代碼打造專屬的本地除錯戰情室(Agent Pipeline Inspector),將黑盒子裡的思考鏈轉換為直觀的互動式時間軸與決策儀表板!
我們的本地除錯戰情室旨在提供零學習成本、專為 AI Agent 最佳化的視覺化除錯體驗:
+--------------------------------------------------------------------------------+
| 📡 Intelligence Radar - Agent Pipeline Inspector |
+--------------------------------------------------------------------------------+
| 側邊欄 (Sidebar) | 主檢視區 (Main Dashboard) |
| | |
| 📅 選擇巡邏批次 | 📊 全域指標卡片 (KPI Metrics) |
| [2026-10-10 ▼] | [ 處理總數: 6 | 通過率: 66% | 平均耗時: 6,440ms ] |
| | ------------------------------------------------------- |
| 🏷️ 狀態過濾 | 📑 情資清單與快速過濾 (Interactive Table) |
| [x] PROCESSED | 選擇特定情報 ➔ 查看詳細執行流 |
| [x] ARCHIVED | ------------------------------------------------------- |
| [ ] ERROR | 🔍 代理人思考時間軸 (Agent Execution Timeline) |
| | [Step 1] Reviewer: PASS (Score=5) - 耗時 1,420ms |
| 🔍 搜尋 entry_id | [Step 2] Domain Expert: 指標萃取完成 - 耗時 2,150ms |
| [ ] | [Step 3] Synthesizer: 二階跨界推演 - 耗時 1,890ms |
| | [Step 4] Editor: 決策層摘要精煉 - 耗時 980ms |
+--------------------------------------------------------------------------------+
ARCHIVED 或評分低於 4 的案例,直接展開 Reviewer 的具體駁回理由。在 pyproject.toml 與 requirements.txt 中加入 Streamlit 與輕量資料處理依賴:
streamlit>=1.35.0
pandas>=2.2.0
在 scripts/inspector.py 建立視覺化工具。程式碼直接讀取 Day 25 產出的 logs/traces/*.jsonl 檔案:
# scripts/inspector.py
from __future__ import annotations
import json
from pathlib import Path
import pandas as pd
import streamlit as st
st.set_page_config(
page_title="Agent Pipeline Inspector",
page_icon="📡",
layout="wide"
)
# 1. 載入並快取追蹤紀錄
@st.cache_data
def load_trace_files() -> dict[str, Path]:
log_dir = Path("logs/traces")
if not log_dir.exists():
return {}
files = sorted(log_dir.glob("trace_*.jsonl"), reverse=True)
return {f.stem.replace("trace_", ""): f for f in files}
def parse_trace_data(file_path: Path) -> list[dict]:
records = []
with file_path.open("r", encoding="utf-8") as f:
for line in f:
if line.strip():
records.append(json.loads(line))
return records
# 2. 側邊欄:巡檢批次與條件過濾
st.sidebar.title("📡 Radar Inspector")
available_dates = load_trace_files()
if not available_dates:
st.warning("⚠️ 尚未偵測到任何追蹤日誌 (logs/traces/trace_*.jsonl)!請先執行巡航管線。")
st.stop()
selected_date = st.sidebar.selectbox("📅 選擇巡邏批次", list(available_dates.keys()))
raw_data = parse_trace_data(available_dates[selected_date])
status_filter = st.sidebar.multiselect(
"🏷️ 狀態過濾",
options=["PROCESSED", "ARCHIVED", "ERROR"],
default=["PROCESSED", "ARCHIVED", "ERROR"]
)
search_query = st.sidebar.text_input("🔍 搜尋標題或 entry_id", "")
filtered_data = [
item for item in raw_data
if item.get("status") in status_filter
and (search_query.lower() in item.get("source_title", "").lower() or search_query in item.get("entry_id", ""))
]
# 3. 主畫面:指標卡片概覽
st.title(f"巡航除錯檢視:{selected_date}")
col1, col2, col3, col4 = st.columns(4)
total_count = len(raw_data)
processed_count = sum(1 for d in raw_data if d.get("status") == "PROCESSED")
pass_rate = (processed_count / total_count * 100) if total_count > 0 else 0
avg_score = (
sum(d.get("verified_score", 0) for d in raw_data if d.get("verified_score")) / total_count
) if total_count > 0 else 0
col1.metric("採集情資總數", f"{total_count} 筆")
col2.metric("通過初審 (PROCESSED)", f"{processed_count} 筆")
col3.metric("初審通過率", f"{pass_rate:.1f}%")
col4.metric("平均技術評分", f"{avg_score:.2f} / 5.0")
st.divider()
# 4. 條目列表與思考鏈明細
st.subheader("📑 情資審查串流與代理人思考鏈")
for item in filtered_data:
status = item.get("status")
score = item.get("verified_score", "N/A")
title = item.get("source_title", "未命名標題")
entry_id = item.get("entry_id", "未知ID")
# 狀態標記與顏色區分
badge = "🟢 [PROCESSED]" if status == "PROCESSED" else ("🟡 [ARCHIVED]" if status == "ARCHIVED" else "🔴 [ERROR]")
with st.expander(f"{badge} (Score: {score}) | {title} [{entry_id}]"):
st.markdown(f"**來源網址**:{item.get('source_url', '無')}")
traces = item.get("traces", [])
if not traces:
st.info("此項目無代理人追蹤記錄。")
continue
# 呈現代理人依序執行的思考流
st.markdown("#### 🧠 代理人思考時間軸")
for idx, span in enumerate(traces, 1):
agent_name = span.get("agent_name")
duration = span.get("duration_ms", 0)
decision = span.get("decision", "OK")
thought = span.get("thought_process", "無詳細記錄")
err = span.get("error")
icon = "🤖"
if agent_name == "Reviewer":
icon = "⚖️"
elif agent_name == "DomainExpert":
icon = "🔬"
elif agent_name == "Synthesizer":
icon = "🔮"
elif agent_name == "Editor":
icon = "✍️"
st.markdown(f"**Step {idx}:{icon} {agent_name}** `({duration} ms)` — 決策: `{decision}`")
if err:
st.error(f"執行錯誤:{err}")
else:
st.info(thought)
在專案根目錄下執行:
streamlit run scripts/inspector.py
瀏覽器會自動開啟 http://localhost:8501。
當你在側邊欄將過濾器只勾選 ARCHIVED 時:
Step 1: ⚖️ Reviewer,下方即刻呈現 Reviewer 的完整思維鏈摘要:「經核查,本文僅在單一合成資料集測試,缺乏基準模型比較,且作者提出的注意力機制相較現有架構未展現出顯著計算複雜度優勢,扣除 2 分,建議歸檔。」
展開通過評核的 5 分論文,查看 Step 3: 🔮 Synthesizer:
「第一階:模型於邊緣裝置記憶體佔用降低 40%;第二階衝擊:原本受限於運算力的小型四足機器人將可本地運行即時具身決策模型,加速智慧巡檢機器人商業化。」
如果今天巡航是跑在 GitHub Actions 雲端 Runner 上,我們如何在本機檢視當天的 Trace?
在 Day 25 中,我們已經在 Actions 中設定了 Artifact 保存:
- name: Upload Agent Execution Traces
uses: actions/upload-artifact@v4
with:
name: agent-execution-traces
path: logs/traces/*.jsonl
我們可以使用 GitHub 官方 CLI gh,一行指令將雲端的最新追蹤日誌同步回本地:
# 下載最新一場 Workflow 的 Trace 產物至本地 logs/traces/
gh run download --name agent-execution-traces --dir logs/traces/
下載完成後,重新整理 Streamlit 畫面,側邊欄會自動抓出最新的巡航日期,達成「雲端無人值守運算,本機一鍵拉取視覺化」的極致維運體驗。
今天我們完成了情報雷達的可視化閉環:
然而,在真實的高頻巡檢與 API 呼叫中,不可避免會遇到網路抖動、服務商端 503 超載(例如 Gemini API 突發過載)或是 Google Sheets 429 配額告警。
在 Day 27,我們將聚焦系統的高可用性與防禦型架構:「重試、熔斷與配額防禦:打造穩健的 Exponential Backoff 與降級機制」,確保情報雷達在惡劣的外部網路環境下依然堅如磐石!