iT邦幫忙

2026 iThome 鐵人賽

DAY 26
0
Build on Google AI

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

Day 26:本地可視化除錯:使用 Streamlit 快速打造 Agent Pipeline 戰情檢視器

  • 分享至 

  • xImage
  •  

在 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),將黑盒子裡的思考鏈轉換為直觀的互動式時間軸與決策儀表板!


一、設計藍圖: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              |
+--------------------------------------------------------------------------------+

  1. 全域效能與通過率概覽:直觀呈現該批次中各代理人的累計耗時與決策分布。
  2. Bad Case 快速過濾器:一鍵篩出 ARCHIVED 或評分低於 4 的案例,直接展開 Reviewer 的具體駁回理由。
  3. 思考鏈展開卡片(Expander Flow):依序展示 Reviewer ➔ Domain Expert ➔ Synthesizer ➔ Editor 的推導原貌與量化產出。

二、環境依賴安裝

在 pyproject.toml 與 requirements.txt 中加入 Streamlit 與輕量資料處理依賴:

streamlit>=1.35.0
pandas>=2.2.0


三、實作 Inspector:不到 100 行的極簡戰情室

在 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)


四、啟動與實戰除錯體驗

1. 本地啟動指令

在專案根目錄下執行:

streamlit run scripts/inspector.py

瀏覽器會自動開啟 http://localhost:8501。

2. 實戰場景 A:排查被 Reviewer 刷掉的 Bad Case

當你在側邊欄將過濾器只勾選 ARCHIVED 時:

  • 展開某篇評分為 3 的論文。
  • 點開 Step 1: ⚖️ Reviewer,下方即刻呈現 Reviewer 的完整思維鏈摘要:

「經核查,本文僅在單一合成資料集測試,缺乏基準模型比較,且作者提出的注意力機制相較現有架構未展現出顯著計算複雜度優勢,扣除 2 分,建議歸檔。」

  • 我們立刻能確認:這是代理人具備精準評判標準的正常過濾,而非 Prompt 出錯產生的幻覺。

3. 實戰場景 B:檢視 Synthesizer 的跨界推演深度

展開通過評核的 5 分論文,查看 Step 3: 🔮 Synthesizer:

  • 檢視系統推演出的「二階跨界衝擊」:

「第一階:模型於邊緣裝置記憶體佔用降低 40%;第二階衝擊:原本受限於運算力的小型四足機器人將可本地運行即時具身決策模型,加速智慧巡檢機器人商業化。」

  • 這讓我們在不需要手動翻看整篇論文的情況下,一眼驗證 Synthesizer 是否成功產生高價值的跨領域見解。

五、進階架構優化:從本地快取到雲端 Artifact 下載

如果今天巡航是跑在 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 畫面,側邊欄會自動抓出最新的巡航日期,達成「雲端無人值守運算,本機一鍵拉取視覺化」的極致維運體驗。


總結與下一步

今天我們完成了情報雷達的可視化閉環:

  • 用純 Python 與 Streamlit,以極低代碼量實現了免資料庫依賴、直讀 JSONL 的 Agent Pipeline Inspector。
  • 提供多維過濾、全域指標卡片與多代理人思考鏈時間軸,徹底揭開 LLM 代理人的決策黑盒子。

然而,在真實的高頻巡檢與 API 呼叫中,不可避免會遇到網路抖動、服務商端 503 超載(例如 Gemini API 突發過載)或是 Google Sheets 429 配額告警。

在 Day 27,我們將聚焦系統的高可用性與防禦型架構:「重試、熔斷與配額防禦:打造穩健的 Exponential Backoff 與降級機制」,確保情報雷達在惡劣的外部網路環境下依然堅如磐石!


上一篇
Day 25:可觀測性黑盒子:設計 Agent Trace 日誌與結構化思考過程追蹤
下一篇
Day 27:重試、熔斷與配額防禦:打造穩健的 Exponential Backoff 與降級機制
系列文
打造專屬 AI 參謀群:以 Gemini Spark、Workspace 與 ADK 構建自動化決策雷達 共 29 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言