在 Day 25 中，我們在中樞服務實現了無侵入式的 Trace 收集器，將 Reviewer、Domain Expert、Synthesizer 與 Editor 四大 Agent 的呼叫耗時與執行狀態寫入了 JSONL 檔案中。

然而，面對日積月累的 .jsonl 結構化日誌，直接用純文字閱讀不僅效率低落，也很難一眼看出哪一篇情報在初審被刷掉、哪一個階段花費了最多時間，以及Agent 在當下的思考推演內容。

今天我們將使用輕量且強大的 Streamlit，打造專屬的 Radar Inspector 巡航戰情儀表板，支援：

1. 依巡航日期批次動態切換檢視

2. 審查狀態（PROCESSED / ARCHIVED / ERROR）多維度過濾

3. 可展開式思考鏈（Step 1 ~ Step 4）視覺化卡片

4. Agent 思考摘要與關鍵成果即時預覽

一、系統架構與資料流向

Inspector 的定位是離線/本地巡檢工作台（Offline Observability Dashboard）：

[GitHub Actions 雲端巡航] ──(上傳產出物)──> [Artifact: agent-execution-traces]
                                                    │
                                                    ▼ (gh run download 下載)
                                            [logs/traces/*.jsonl]
                                                    ▲
[本地 python run_pipeline.py] ──(直接產出)───────────┘
                                                    │
                                                    ▼
                                     [Streamlit: scripts/inspector.py]
                                                    │
                                                    ▼
                                        [http://localhost:8501 戰情面板]

二、前置增強：在 Trace 中擷取 Agent 思考摘要

在 Day 25 的實作中，如果 _record_trace 只記錄了純耗時指標（duration_sec），Inspector 就只能呈現「執行花了多久」，無法展示「思考了什麼」。

請確認 src/services/pipeline.py 的 _record_trace 已補上 output_preview 欄位：

    def _record_trace(self, item: IntelligenceItem, stage_name: str, duration: float, error=None):
        """將單一 Agent 執行軌跡追加到 metadata 與本地日誌中（含思考鏈輸出預覽）"""
        
        # 依階段提取該 Agent 剛產出的思維摘要
        output_preview = ""
        if stage_name == "Reviewer":
            score_text = f"評分: {item.raw_score} 分" if hasattr(item, "raw_score") else ""
            reason = item.metadata.get("reviewer_reason") or "通過初審"
            output_preview = f"{score_text} | 審核結果: {reason}"
        elif stage_name == "DomainExpert":
            m = item.metrics or "無特殊量化指標"
            lim = item.limitations or "無特殊技術限制"
            output_preview = f"【指標】{m}\n【限制】{lim}"
        elif stage_name == "Synthesizer":
            output_preview = item.cross_impact_notes or "無跨領域推演"
        elif stage_name == "Editor":
            output_preview = item.editorial_summary or "無定稿摘要"

        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,
            "output_preview": output_preview,  # 🌟 關鍵思考鏈文字
        }
        
        if "execution_trace" not in item.metadata:
            item.metadata["execution_trace"] = []
        item.metadata["execution_trace"].append(span)

        # 即時寫入本地 JSONL
        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")

三、實作 Streamlit Inspector (scripts/inspector.py)

因為流水線採用「即時串流落盤」，每個 Agent 執行完畢都會寫入一行獨立的 Span JSON。
在 Inspector 中，我們必須依據 entry_id 進行分組聚合（Aggregation），才能正確還原每篇情資的完整執行流：

# scripts/inspector.py
import json
from pathlib import Path
import streamlit as st

st.set_page_config(page_title="Radar Inspector", page_icon="📡", layout="wide")

st.sidebar.title("📡 Radar Inspector")

# 1. 動態掃描 logs/traces 目錄下的所有批次檔
trace_dir = Path("logs/traces")
trace_files = sorted(list(trace_dir.glob("trace_*.jsonl")), reverse=True)

if not trace_files:
    st.warning("⚠️ 尚未找到任何 Trace 紀錄檔，請先執行 pipeline 產生日誌。")
    st.stop()

# 提取日期列表
date_options = [f.stem.replace("trace_", "") for f in trace_files]
selected_date = st.sidebar.selectbox("📅 選擇巡邏批次", date_options)

selected_file = trace_dir / f"trace_{selected_date}.jsonl"

# 2. 讀取並將同一情資的 Spans 聚合（以 entry_id 為單位）
raw_spans = []
if selected_file.exists():
    with open(selected_file, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                try:
                    raw_spans.append(json.loads(line))
                except Exception:
                    continue

# 聚合邏輯：將同一個 entry_id 的所有 Span 打包在一起
aggregated_items = {}
for entry in raw_spans:
    eid = entry.get("entry_id")
    if not eid:
        continue
    if eid not in aggregated_items:
        aggregated_items[eid] = {
            "entry_id": eid,
            "source_title": entry.get("source_title", "Untitled"),
            "spans": [],
            "status": "ARCHIVED",  # 預設狀態
        }
    
    if "span" in entry:
        aggregated_items[eid]["spans"].append(entry["span"])
    elif "traces" in entry:
        aggregated_items[eid]["spans"].extend(entry.get("traces", []))

# 狀態研判：若跑過 Editor 成功代表已定稿 (PROCESSED)
for eid, item in aggregated_items.items():
    stage_names = [s.get("stage") for s in item["spans"]]
    if "Editor" in stage_names:
        item["status"] = "PROCESSED"
    elif any(s.get("status") == "ERROR" for s in item["spans"]):
        item["status"] = "ERROR"
    else:
        item["status"] = "ARCHIVED"

items_list = list(aggregated_items.values())

# 3. 側邊欄過濾控制項
status_filter = st.sidebar.multiselect(
    "🏷️ 狀態過濾",
    options=["PROCESSED", "ARCHIVED", "ERROR"],
    default=["PROCESSED", "ARCHIVED", "ERROR"]  # 預設全選，絕不漏看
)

search_query = st.sidebar.text_input("🔍 搜尋標題或 entry_id", "")

filtered_items = [
    item for item in items_list
    if item["status"] in status_filter
    and (search_query.lower() in item["source_title"].lower() or search_query.lower() in item["entry_id"].lower())
]

# 4. 頂部儀表板指標 (Metrics)
st.title(f"巡航除錯檢視：{selected_date}")

total_count = len(items_list)
processed_count = sum(1 for i in items_list if i["status"] == "PROCESSED")
pass_rate = (processed_count / total_count * 100) if total_count > 0 else 0.0

col1, col2, col3 = st.columns(3)
col1.metric("採集情資總數", f"{total_count} 筆")
col2.metric("通過初審 (PROCESSED)", f"{processed_count} 筆")
col3.metric("初審通過率", f"{pass_rate:.1f}%")

st.divider()

# 5. 渲染情資審查串流與思考鏈卡片
st.subheader("📄 情資審查串流與代理人思考鏈")

if not filtered_items:
    st.info("💡 目前篩選條件下無符合的情資項目，請調整側邊欄的過濾條件。")
else:
    for item in filtered_items:
        with st.expander(f"【{item['status']}】 [{item['entry_id']}] {item['source_title']}", expanded=(item["status"] == "PROCESSED")):
            st.markdown(f"**項目識別碼 (Entry ID):** `{item['entry_id']}`")
            
            stages = item["spans"]
            if not stages:
                st.write("尚無詳細思考鏈軌跡。")
            else:
                for idx, span in enumerate(stages, 1):
                    stage_name = span.get("stage", f"Step {idx}")
                    duration = span.get("duration_sec", 0)
                    status_icon = "✅" if span.get("status") == "SUCCESS" else "❌"
                    
                    icons = {
                        "Reviewer": "⚖️ Reviewer 初審守門員",
                        "DomainExpert": "🔬 Domain Expert 領域提煉",
                        "Synthesizer": "🔮 Synthesizer 跨領域推演",
                        "Editor": "✍️ Editor 總編排版定稿",
                    }
                    display_title = icons.get(stage_name, f"階段: {stage_name}")
                    
                    st.markdown(f"#### Step {idx}: {display_title} ({status_icon} 耗時: `{duration}s`)")
                    
                    if span.get("error"):
                        st.error(f"遭遇錯誤: {span.get('error')}")
                    else:
                        st.caption(f"執行時間: {span.get('timestamp')}")
                        output_content = span.get("output_preview")
                        if output_content:
                            st.markdown("**🧠 代理人思考與萃取摘要：**")
                            st.info(output_content)
                        else:
                            st.caption("*(此階段無文字預覽)*")
                            
                    st.write("---")

四、啟動戰情室與操作指南

1. 啟動 Streamlit

在專案虛擬環境中執行：

streamlit run scripts/inspector.py

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

2. 同步雲端 GitHub Actions 巡航結果（若由 CI/CD 執行）

若巡航管線是在 GitHub Actions 雲端定期執行，可透過 GitHub CLI 下載最新產出的 Artifact：

# 1. 登入 GitHub（只需初次設定一次，選 Web 瀏覽器認證最快速）
gh auth login

# 2. 下載產出物，加上 --clobber 防止檔案存在時衝突
gh run download --name agent-execution-traces --dir logs/traces/ --clobber

下載後回到瀏覽器按下鍵盤 R 重新整理，即可在下拉選單看到最新的雲端巡檢紀錄！

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

Q1: 為什麼點開卡片只有「執行時間與耗時」，卻沒有思考鏈文字？

• 原因：在基礎版 Trace 設計中，Span 字典只記錄了 duration_sec、timestamp 與 status，沒有將模型生成的推演摘要放入物件中。

• 解法：在 pipeline.py 的 _record_trace() 中加入 output_preview，動態從 item 提取 reviewer_reason、metrics、limitations、cross_impact_notes 與 editorial_summary 存入 Span 結構中。

Q2: 為什麼明明有跑過的情資，畫面上「通過初審 (PROCESSED)」卻顯示為 0 筆？

• 原因：即時串流落盤時，每一行寫入的是「Span 紀錄」，本身沒有外層的 status: "PROCESSED"。若直接依單行做狀態篩選，會全部判定為空值；若側邊欄剛好只勾選 PROCESSED，就會被全部過濾掉导致畫面空白。

• 解法：在 inspector.py 讀取時，透過 aggregated_items 字典以 entry_id 進行聚合，檢查是否具備 Editor 階段來判定最終狀態。

Q3: 執行 gh run download 出現 error extracting ... openat trace_xxx.jsonl: file exists？

• 原因：GitHub CLI 為防止本機檔案被無預警覆蓋，預設遇到同名檔案會拒絕解壓縮。
• 解法：在指令後方加上 --clobber 參數（例如 gh run download --name ... --dir ... --clobber）強制覆蓋同名檔案。

Q4: 提示 To get started with GitHub CLI, please run: gh auth login，如何配置？

• 說明：這是 GitHub CLI 所需的驗證。
• 解法：直接在終端機輸入 gh auth login，依序選擇 GitHub.com $\rightarrow$ HTTPS $\rightarrow$ Login with a web browser，在瀏覽器輸入 8 碼配對代碼即可完成授權，完全不需要手動建立複雜的 Personal Access Token。

六、結語

透過今天實作的 Radar Inspector，我們把冷冰冰的日誌檔案轉化為具備可觀測性的互動戰情儀表板：

• 能在秒級掌握每一次巡航的初審通過率

• 能清楚診斷四大 Agent 中哪一個環節耗時過長

• 能展開每一篇情報的「決策推理過程」，為 Prompt 調優與邊界規則迭代提供最直觀的依據！