自建第一手 LLM 引用率監測 — 為什麼不要只依賴二手數據

為什麼需要第一手數據

GEO 監測工具市場上有兩條路：

路 1：商業 SaaS（Profound、Otterly、Peec 等）幫你跑通用查詢、給你看趨勢、提供 dashboard。優點是上手快、不用工程資源；缺點是它給你的查詢清單是它認為你該追的——不是你目標客戶實際在問的。

路 2：自建第一手量測——你自己決定查哪些問題、用哪個模型、怎麼解析。

兩條路不互斥，但對「真正想理解 AI 怎麼推薦自家品牌」的團隊，第一手訊號不可省。原因：

• 查詢清單客製化：你的目標客戶的實際提問方式（語氣、長度、上下文）只有你最懂；通用 SaaS 用的是泛化模板
• 模型版本即時對照：每代新模型上線時，你想知道的是「新模型對我的引用變化」，不是 SaaS 廠商的整合時間表
• 查詢 + 回應全文留底：之後要追根究柢「為什麼這個月引用率掉」時，有原始 response text 可以分析才追得回去
• 跨產品資料串接：你的 GA4、表單、CRM 資料要跟 LLM 引用資料 join 分析時，第一手數據才能直接 SQL JOIN

這篇給的是「自己跑、自己存、自己看」的最小架構。不是要你取代商業 SaaS——多數團隊兩條路並行最務實：SaaS 看大盤趨勢，自建看深度問題。

自建會給你什麼 vs 不會給你什麼

換言之：自建解決資料採集，不解決資料解讀與策略。後者仍需要 GEO 領域知識——這也是顧問服務的核心價值（見文末）。

下面是最小實作架構。

系統架構

┌─────────────────┐
│ collector.py    │ ← 每天 03:00 執行（cron / Task Scheduler）
│ • 讀 queries    │
│ • 呼叫 3 API    │
│ • 解析回應      │
└────────┬────────┘
         │ 寫入
         ▼
┌─────────────────┐
│ monitoring.db   │ ← SQLite，3 張表
│ • queries       │
│ • runs          │
│ • results       │
└────────┬────────┘
         │ 讀取
         ▼
┌─────────────────┐
│ dashboard.py    │ ← 隨時跑（或定期）
│ • 產 HTML       │ ← 用 Plotly 畫趨勢
│ • 開瀏覽器      │
└─────────────────┘

Step 1：環境準備

# Python 3.10+
pip install openai anthropic httpx plotly

.env 檔（永遠不要 commit 進 git）：

OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
PERPLEXITY_API_KEY=pplx-...
BRAND_NAME=您的公司名
BRAND_DOMAIN=example.com

.gitignore：

.env
*.db
__pycache__/

Step 2：定義你的 20 個目標查詢

queries.yaml：

queries:
  - id: 1
    text: "中小企業適用的客戶管理系統推薦"
    category: "product_recommend"
  - id: 2
    text: "2026 年最值得用的 CRM 軟體"
    category: "product_recommend"
  - id: 3
    text: "B2B 行銷自動化工具比較"
    category: "comparison"
  - id: 4
    text: "如何選擇 CRM 廠商"
    category: "how_to"
  # ... 共 20 個

選查詢的原則：

• 使用者真的會問：避開「我希望被推薦」但實際沒人查的
• 有業務轉化潛力：問題對應你的產品 / 服務
• 跨類型分布：產品推薦 / 比較 / 教學 / 趨勢分析等

Step 3：DB schema

schema.sql：

-- 查詢清單（穩定）
CREATE TABLE IF NOT EXISTS queries (
    id INTEGER PRIMARY KEY,
    text TEXT NOT NULL,
    category TEXT,
    active INTEGER DEFAULT 1
);

-- 每次執行的 metadata
CREATE TABLE IF NOT EXISTS runs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    started_at TEXT NOT NULL,
    completed_at TEXT,
    queries_count INTEGER,
    platforms TEXT,
    status TEXT  -- ok / partial / failed
);

-- 個別查詢結果
CREATE TABLE IF NOT EXISTS results (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id INTEGER NOT NULL,
    query_id INTEGER NOT NULL,
    platform TEXT NOT NULL,  -- openai / anthropic / perplexity
    response_text TEXT,      -- 完整回應原文（debug 用）
    brand_mentioned INTEGER, -- 0/1
    brand_position INTEGER,  -- 出現第幾個（沒出現 NULL）
    citation_url INTEGER,    -- 是否含我網域 URL（Perplexity 特有）
    error TEXT,              -- API 錯誤訊息（如有）
    created_at TEXT,
    FOREIGN KEY (run_id) REFERENCES runs(id),
    FOREIGN KEY (query_id) REFERENCES queries(id)
);

CREATE INDEX IF NOT EXISTS idx_results_run ON results(run_id);
CREATE INDEX IF NOT EXISTS idx_results_query ON results(query_id);

Step 4：collector.py — 採集腳本

"""每日跑：對每個 query × 每個 platform 執行查詢，存入 DB。"""

import os
import re
import sqlite3
import time
from datetime import datetime, timezone

import httpx
import yaml
from anthropic import Anthropic
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

BRAND_NAME = os.environ["BRAND_NAME"]
BRAND_DOMAIN = os.environ["BRAND_DOMAIN"]
DB_PATH = "monitoring.db"

openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
anthropic_client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
perplexity_key = os.environ["PERPLEXITY_API_KEY"]


def init_db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    with open("schema.sql") as f:
        conn.executescript(f.read())
    return conn


def load_queries(conn: sqlite3.Connection) -> list[dict]:
    """從 yaml 同步到 DB，回傳 active queries。"""
    with open("queries.yaml") as f:
        config = yaml.safe_load(f)
    for q in config["queries"]:
        conn.execute(
            "INSERT OR REPLACE INTO queries (id, text, category, active) VALUES (?, ?, ?, 1)",
            (q["id"], q["text"], q.get("category", "")),
        )
    conn.commit()
    rows = conn.execute(
        "SELECT id, text FROM queries WHERE active = 1"
    ).fetchall()
    return [{"id": r[0], "text": r[1]} for r in rows]


def parse_brand_appearance(text: str) -> tuple[bool, int | None]:
    """從回應文字中找品牌名出現位置。
    回傳 (mentioned, position)。position 是「第幾個列出的選項」。"""
    if not text or BRAND_NAME not in text:
        return False, None

    # 簡化：找出所有條列項，看品牌在第幾項出現
    # 真實情境可改用 LLM 自己解析（成本高但更準）
    items = re.findall(r"^\s*[\d•\-\*]+[\.\)]\s*(.+)$", text, re.MULTILINE)
    for idx, item in enumerate(items, start=1):
        if BRAND_NAME in item:
            return True, idx

    # 沒在條列中但提到 → 視為提及但無排序
    return True, None


def query_openai(query_text: str) -> dict:
    try:
        resp = openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "user",
                    "content": query_text,
                }
            ],
            temperature=0,
        )
        text = resp.choices[0].message.content or ""
        mentioned, pos = parse_brand_appearance(text)
        return {
            "response_text": text,
            "brand_mentioned": int(mentioned),
            "brand_position": pos,
            "error": None,
        }
    except Exception as exc:
        return {
            "response_text": None, "brand_mentioned": 0,
            "brand_position": None, "error": str(exc),
        }


def query_anthropic(query_text: str) -> dict:
    try:
        resp = anthropic_client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": query_text}],
        )
        text = resp.content[0].text if resp.content else ""
        mentioned, pos = parse_brand_appearance(text)
        return {
            "response_text": text,
            "brand_mentioned": int(mentioned),
            "brand_position": pos,
            "error": None,
        }
    except Exception as exc:
        return {
            "response_text": None, "brand_mentioned": 0,
            "brand_position": None, "error": str(exc),
        }


def query_perplexity(query_text: str) -> dict:
    """Perplexity API 與 OpenAI compatible，但會回傳 citation URLs。"""
    try:
        with httpx.Client(timeout=30.0) as client:
            r = client.post(
                "https://api.perplexity.ai/chat/completions",
                headers={"Authorization": f"Bearer {perplexity_key}"},
                json={
                    "model": "llama-3.1-sonar-large-128k-online",
                    "messages": [{"role": "user", "content": query_text}],
                },
            )
            r.raise_for_status()
            data = r.json()
            text = data["choices"][0]["message"]["content"] or ""
            citations = data.get("citations", [])
            mentioned, pos = parse_brand_appearance(text)
            cited_url = int(any(BRAND_DOMAIN in c for c in citations))
            return {
                "response_text": text,
                "brand_mentioned": int(mentioned),
                "brand_position": pos,
                "citation_url": cited_url,
                "error": None,
            }
    except Exception as exc:
        return {
            "response_text": None, "brand_mentioned": 0,
            "brand_position": None, "citation_url": 0, "error": str(exc),
        }


def main() -> None:
    conn = init_db()
    queries = load_queries(conn)

    cur = conn.execute(
        "INSERT INTO runs (started_at, queries_count, platforms, status) "
        "VALUES (?, ?, ?, 'in_progress')",
        (datetime.now(timezone.utc).isoformat(), len(queries),
         "openai,anthropic,perplexity"),
    )
    run_id = cur.lastrowid
    conn.commit()

    fail_count = 0
    for q in queries:
        for platform_name, query_fn in [
            ("openai", query_openai),
            ("anthropic", query_anthropic),
            ("perplexity", query_perplexity),
        ]:
            print(f"[{platform_name}] q={q['id']}: {q['text'][:40]}...")
            r = query_fn(q["text"])
            if r.get("error"):
                fail_count += 1
            conn.execute(
                "INSERT INTO results "
                "(run_id, query_id, platform, response_text, brand_mentioned, "
                " brand_position, citation_url, error, created_at) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                (
                    run_id, q["id"], platform_name,
                    r["response_text"], r["brand_mentioned"], r["brand_position"],
                    r.get("citation_url", 0), r.get("error"),
                    datetime.now(timezone.utc).isoformat(),
                ),
            )
            conn.commit()
            time.sleep(1)  # rate limit 友善

    status = "ok" if fail_count == 0 else ("partial" if fail_count < 10 else "failed")
    conn.execute(
        "UPDATE runs SET completed_at = ?, status = ? WHERE id = ?",
        (datetime.now(timezone.utc).isoformat(), status, run_id),
    )
    conn.commit()
    print(f"Run {run_id} done: {status} ({fail_count} errors)")


if __name__ == "__main__":
    main()

Step 5：dashboard.py — 視覺化

"""讀 monitoring.db，產出 HTML dashboard 開瀏覽器。"""

import sqlite3
import webbrowser
from pathlib import Path

import plotly.graph_objects as go

DB_PATH = "monitoring.db"
OUTPUT = "dashboard.html"


def main() -> None:
    conn = sqlite3.connect(DB_PATH)

    # 1. 每日整體出現率（across all platforms + queries）
    rows = conn.execute("""
        SELECT
            DATE(r.started_at) as day,
            res.platform,
            AVG(res.brand_mentioned) * 100 as appearance_rate,
            AVG(CASE WHEN res.brand_position IS NOT NULL
                THEN res.brand_position ELSE NULL END) as avg_position
        FROM runs r
        JOIN results res ON res.run_id = r.id
        WHERE r.status IN ('ok', 'partial')
        GROUP BY day, res.platform
        ORDER BY day
    """).fetchall()

    # 重整成 platform → [(day, rate)]
    platform_data: dict[str, dict] = {}
    for day, plat, rate, avg_pos in rows:
        platform_data.setdefault(plat, {"days": [], "rates": [], "positions": []})
        platform_data[plat]["days"].append(day)
        platform_data[plat]["rates"].append(rate)
        platform_data[plat]["positions"].append(avg_pos)

    # 2. 出現率趨勢圖
    fig1 = go.Figure()
    for plat, data in platform_data.items():
        fig1.add_trace(go.Scatter(
            x=data["days"], y=data["rates"],
            mode="lines+markers", name=plat,
        ))
    fig1.update_layout(
        title="每日品牌出現率 (%) — 各平台",
        xaxis_title="日期", yaxis_title="出現率 (%)",
        height=400,
    )

    # 3. 平均位次趨勢
    fig2 = go.Figure()
    for plat, data in platform_data.items():
        fig2.add_trace(go.Scatter(
            x=data["days"], y=data["positions"],
            mode="lines+markers", name=plat,
        ))
    fig2.update_layout(
        title="平均推薦位次 — 各平台（越低越好）",
        xaxis_title="日期", yaxis_title="平均位次",
        yaxis=dict(autorange="reversed"),  # 1 在上面
        height=400,
    )

    # 4. 最近 7 天每查詢的命中熱圖
    rows = conn.execute("""
        SELECT q.text, res.platform,
               AVG(res.brand_mentioned) * 100 as rate
        FROM results res
        JOIN queries q ON q.id = res.query_id
        JOIN runs r ON r.id = res.run_id
        WHERE r.started_at >= datetime('now', '-7 days')
        GROUP BY q.text, res.platform
    """).fetchall()

    queries_seen = sorted(set(r[0] for r in rows))
    platforms_seen = sorted(set(r[1] for r in rows))
    matrix = [[0] * len(platforms_seen) for _ in queries_seen]
    for q_text, plat, rate in rows:
        i = queries_seen.index(q_text)
        j = platforms_seen.index(plat)
        matrix[i][j] = rate

    fig3 = go.Figure(data=go.Heatmap(
        z=matrix, x=platforms_seen, y=queries_seen,
        colorscale="RdYlGn", zmin=0, zmax=100,
    ))
    fig3.update_layout(
        title="最近 7 天命中率熱圖",
        height=600,
    )

    # 5. 寫 HTML
    html = f"""<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>LLM Citation Monitor</title>
<style>
body {{ font-family: system-ui; max-width: 1200px; margin: 2em auto; padding: 0 1em; }}
h1 {{ color: #1a56db; }}
section {{ margin: 2em 0; }}
</style></head>
<body>
<h1>LLM 引用率監測 dashboard</h1>
<p>更新時間：<code>{__import__('datetime').datetime.now().isoformat()}</code></p>
<section>{fig1.to_html(full_html=False, include_plotlyjs="cdn")}</section>
<section>{fig2.to_html(full_html=False, include_plotlyjs=False)}</section>
<section>{fig3.to_html(full_html=False, include_plotlyjs=False)}</section>
</body></html>"""

    Path(OUTPUT).write_text(html, encoding="utf-8")
    print(f"Dashboard 已產出：{OUTPUT}")
    webbrowser.open(f"file://{Path(OUTPUT).absolute()}")


if __name__ == "__main__":
    main()

Step 6：排程

Linux / macOS（cron）

# 每天凌晨 03:00 跑採集
0 3 * * * cd /path/to/monitor && /path/to/python collector.py >> collector.log 2>&1

Windows（Task Scheduler）

# 透過 schtasks 註冊
schtasks /Create /SC DAILY /ST 03:00 /TN "LLM Monitor" `
  /TR "C:\Python312\python.exe C:\monitor\collector.py"

成本估算

每天 20 query × 3 platform = 60 API 呼叫。

加上若用 cloud function 每日跑一次、SQLite 用本地：總成本約 $7–15/月。

vs 商業 SaaS $200–500/月，省 95%+。

進階改良方向

改良 1：用 LLM 自己解析回應品牌位置

parse_brand_appearance 用 regex 對中文 + 不規則格式 LLM 回應準確度有限。改用一個輕模型（gpt-4o-mini / claude-haiku）讀回應問「品牌 X 在第幾個被提到」，每次成本 + $0.001 但準確度大幅提升。

改良 2：多品牌（含競爭者）追蹤

把 BRAND_NAME 改成 list（自家 + 5 個主要競爭者），同時追蹤每家在每查詢的出現率與位次。競爭者比對是 GEO 量測中最有說服力的數據。

改良 3：警示通知

當：

• 某平台出現率連續 3 天下降 > 20%
• 某高優先 query 連續 7 天 0 出現
• run 狀態為 failed

→ 寄 email / 推 Slack。

改良 4：歷史回溯與週月報表

• 週報：每週六自動寄出本週趨勢 + top 改善 / 倒退 query
• 月報：自動產 PDF（matplotlib 或 weasyprint）寄給老闆

第一步：先做 Step 1–4（最小可行版本）

把 collector.py + queries.yaml + schema.sql 跑起來，連續 7 天有資料。第一週的數字會作為「基線」。

👉 GeoWeb 健檢 提供站內訊號的量化評估，但站外即時引用率需要這種主動量測腳本才看得到。

如果你想做完整客製化監測系統（含多品牌追蹤、警示通知、自動週月報、團隊 dashboard 整合），這是 GEO 顧問服務的範圍：[email protected]

GEO 深度系列。前一篇：「子網域 vs 子目錄 vs 多站架構 — 對 GEO 的影響與決策框架」