iT邦幫忙

2026 iThome 鐵人賽

DAY 21
0
Build on Google AI

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

Day 21:情報沉澱:將戰情室情資無縫同步至個人第二大腦 (Obsidian Vault 雙鏈知識網)

  • 分享至 

  • xImage
  •  

在 Day 20 中,我們在 Google Sheets 打造了兼具統計指標與動態檢視的戰情室儀表板。然而,Google Sheets 本質上是一個關聯型態的快速篩檢與樞紐分析平台,並不適合進行長期的知識內化、非線性網狀連結與深度寫作。

當一篇技術論文或產業研報被標記為 PROCESSED 且評分達到高標(例如 $\ge 4$)時,它代表具有長期複利價值的資產。我們需要將這些情報自動轉譯為具備 YAML Frontmatter、雙向鏈接(Wikilinks)與結構化 Callout 的 Markdown 檔案,無縫沉澱到個人知識管理(PKM)系統——以 Obsidian Vault 為例。

本篇將以六角架構延伸出專屬的輸出端口(Driven Port)與轉接器(Adapter),實作自動化同步管線。


一、架構設計:從「流動情報」到「靜態資產」

在 PKM 體系中,情報的生命週期包含三層:

  1. Inbox / Raw Feed:Gemini Spark 採集的高雜訊來源(Google Sheets raw_feed)。
  2. Triaged / Triage Hub:Pipeline 四大代理人過濾與深化後的研報(Google Sheets dashboard)。
  3. Permanent Notes / Second Brain:沉澱至本地 Vault,具備雙向鏈結與語義標籤,隨時可與個人過往筆記碰撞的原子化筆記。
+------------------------+       +-------------------------+       +------------------------+
| Google Sheets          | ----> | ObsidianSyncAdapter     | ----> | Local Vault / Git Repo |
| (PROCESSED & Score>=4) |       | (Jinja2 / Frontmatter)  |       | (Atomic Markdown Notes)|
+------------------------+       +-------------------------+       +------------------------+

為什麼選擇 Obsidian Vault 本地優先(Local-first)模式?

  • 持久性與零鎖定:純文字 Markdown 配合 Git 版控,不依賴任何專有資料庫或雲端服務商的 API 變更。
  • 圖譜與雙向鏈接:利用 [[技術標籤]]、[[關鍵作者]] 與 [[所屬領域]],在本地自動織成知識網。
  • 資料夾與 MOC 架構:同步後的情報放置於 00_Inbox/Radar/ 或 10_Sources/Papers/,並自動掛載到領域 MOC(Map of Content)。

二、領域接口定義:KnowledgeSyncPort

秉持六角架構的精神,核心管線不應直接依賴檔案寫入細節。我們在 src/ports/knowledge_sync.py 定義統一的輸出接口:

# src/ports/knowledge_sync.py
from abc import ABC, abstractmethod
from src.models.intelligence import IntelligenceItem

class KnowledgeSyncPort(ABC):
    @abstractmethod
    def sync_item(self, item: IntelligenceItem) -> bool:
        """將單筆已處理的情報同步至外部知識庫。"""
        pass

    @abstractmethod
    def batch_sync(self, items: list[IntelligenceItem]) -> dict[str, int]:
        """批次同步情報清單,回傳成功與略過筆數。"""
        pass


三、適配器實作:ObsidianVaultAdapter

適配器負責三件事:

  1. 防重複同步(Idempotency):檢查目標檔案是否存在,若存在且未開啟強制覆寫則略過。
  2. 檔名安全性清理(Sanitization):移除非法字元(如 :、/、\ 等),格式化為 YYYY-MM-DD - [分類] 標題.md。
  3. 渲染結構化 Markdown:包含標準 Frontmatter(供 Obsidian Dataview 插件查詢)與 Callouts(突出 Benchmarks、限制與二階效應)。

1. 筆記模板定義

---
id: "{{ item.id }}"
title: "{{ item.title | replace('"', '\\"') }}"
source: "{{ item.source_url }}"
domain: "{{ item.domain }}"
score: {{ item.review_score }}
created_date: {{ item.created_at.strftime('%Y-%m-%d') }}
tags:
  - intelligence-radar
  - domain/{{ item.domain | lower }}
{% for tag in item.tags %}
  - {{ tag }}
{% endfor %}
---

# {{ item.title }}

> [!abstract] 核心提煉
> {{ item.executive_summary }}

## 📊 量化對比與效能指標
{{ item.benchmarks_and_metrics if item.benchmarks_and_metrics else "無顯著量化指標揭露。" }}

## ⚠️ 技術邊界與潛在缺陷
{{ item.limitations if item.limitations else "作者未於論文中強調顯著邊界。" }}

## 🔮 二階效應與跨領域衝擊
> [!tip] 跨界推演
> {{ item.second_order_effects if item.second_order_effects else "暫無推演。" }}

---
## 關聯實體與雙鏈
- 領域節點: [[{{ item.domain }}]]
- 來源追蹤: [原始文章連結]({{ item.source_url }})

2. 轉接器完整代碼

# src/adapters/obsidian_adapter.py
from __future__ import annotations

import re
from pathlib import Path
from jinja2 import Template
from src.ports.knowledge_sync import KnowledgeSyncPort
from src.models.intelligence import IntelligenceItem

OBSIDIAN_TEMPLATE = """---
id: "{{ item.id }}"
title: "{{ item.title | replace('"', '\\"') }}"
source: "{{ item.source_url }}"
domain: "{{ item.domain }}"
score: {{ item.review_score }}
created_date: {{ item.created_at.strftime('%Y-%m-%d') }}
tags:
  - intelligence-radar
  - domain/{{ item.domain | lower }}
{% for tag in item.tags %}
  - {{ tag }}
{% endfor %}
---

# {{ item.title }}

> [!abstract] 核心提煉
> {{ item.executive_summary }}

## 📊 量化對比與效能指標
{{ item.benchmarks_and_metrics if item.benchmarks_and_metrics else "無顯著量化指標揭露。" }}

## ⚠️ 技術邊界與潛在缺陷
{{ item.limitations if item.limitations else "無特別載明。" }}

## 🔮 二階效應與跨界推演
> [!tip] 衝擊預測
> {{ item.second_order_effects if item.second_order_effects else "暫無推演。" }}

---
## 關聯實體
- 領域: [[{{ item.domain }}]]
- 原始連結: [來源網址]({{ item.source_url }})
"""

class ObsidianVaultAdapter(KnowledgeSyncPort):
    def __init__(self, vault_path: str | Path, subfolder: str = "00_Inbox/Radar"):
        self.vault_path = Path(vault_path).resolve()
        self.target_dir = self.vault_path / subfolder
        self.target_dir.mkdir(parents=True, exist_ok=True)
        self.template = Template(OBSIDIAN_TEMPLATE)

    @staticmethod
    def _sanitize_filename(name: str) -> str:
        # 去除 Windows / POSIX 檔案系統禁用的字元
        sanitized = re.sub(r'[\\/*?:"<>|]', "", name)
        # 壓縮過長檔名,避免超出作業系統 255 字元上限
        return sanitized.strip()[:100]

    def _build_filepath(self, item: IntelligenceItem) -> Path:
        date_prefix = item.created_at.strftime("%Y-%m-%d")
        safe_title = self._sanitize_filename(item.title)
        filename = f"{date_prefix} - [{item.domain}] {safe_title}.md"
        return self.target_dir / filename

    def sync_item(self, item: IntelligenceItem, overwrite: bool = False) -> bool:
        target_file = self._build_filepath(item)
        if target_file.exists() and not overwrite:
            return False

        content = self.template.render(item=item)
        target_file.write_text(content, encoding="utf-8")
        return True

    def batch_sync(self, items: list[IntelligenceItem], overwrite: bool = False) -> dict[str, int]:
        stats = {"synced": 0, "skipped": 0}
        for item in items:
            if self.sync_item(item, overwrite=overwrite):
                stats["synced"] += 1
            else:
                stats["skipped"] += 1
        return stats


四、整合至排程巡航與本機工作流程

在實際運行中,情報沉澱有兩種模式:

  1. GitHub Actions 遠端自動化 Commit 模式:
    若個人 Obsidian Vault 代管於私有 GitHub 儲存庫,可在排程巡航中建立專屬 Git Submodule,執行完畢後以 Deploy Key 自動將沉澱的 Markdown 提交(Push)回 Vault。
  2. 本機 CLI 同步工具模式:
    在工作站上以排程(crontab)或手動執行 Python CLI 指令,調用 SheetsAdapter 取得尚未同步至本地的項目,直接寫入本機硬碟的 Obsidian 資料夾中。

以下展示本地單元同步腳本範例:

# scripts/sync_to_obsidian.py
import os
from pathlib import Path
from src.adapters.sheets_adapter import GoogleSheetsAdapter
from src.adapters.obsidian_adapter import ObsidianVaultAdapter

def main():
    vault_path = os.getenv("OBSIDIAN_VAULT_PATH", "~/ObsidianVault")
    vault_adapter = ObsidianVaultAdapter(
        vault_path=Path(vault_path).expanduser(),
        subfolder="00_Inbox/Radar"
    )

    sheets_adapter = GoogleSheetsAdapter()
    
    # 僅提取 PROCESSED 且高分(review_score >= 4)之項目
    candidates = sheets_adapter.fetch_processed_items(min_score=4)
    print(f"[*] 找到 {len(candidates)} 筆高價值情報候選項目...")

    results = vault_adapter.batch_sync(candidates)
    print(f"[+] 同步完成!新增: {results['synced']} 篇,略過: {results['skipped']} 篇。")

if __name__ == "__main__":
    main()


五、在 Obsidian 內的自動化應用:Dataview 查詢

筆記沉澱至 Obsidian 後,即可利用 Dataview 插件 在個人每週回顧或特定技術看板上動態聚合這些情資:

```dataview
TABLE domain AS "領域", score AS "評分", created_date AS "收錄日期"
FROM "00_Inbox/Radar"
WHERE score >= 4
SORT created_date DESC
LIMIT 10

藉由統一規範的 Frontmatter,每一篇情報不再只是 Google Sheets 裡的橫向表格資料,而是具備獨立生命週期的原子化知識筆記,隨時可與個人的架構方案、工作日誌與論文速讀建立雙向關聯。

上一篇
Day 20:戰情室總覽:從 Google Sheets 到輕量動態情報儀表板
下一篇
Day 22:雙腦共振:Obsidian DataviewJS 看板與雙向鏈結實戰
系列文
打造專屬 AI 參謀群:以 Gemini Spark、Workspace 與 ADK 構建自動化決策雷達 共 22 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言