Build Your Own First-Party LLM Citation Monitor — Why You Shouldn't Rely on Second-Hand Data Alone

Why You Need First-Party Data

There are two paths in the GEO monitoring tool market:

Path 1: Commercial SaaS (Profound, Otterly, Peec, etc.) runs generic queries for you, shows you trends, and provides a dashboard. The upside is that it’s quick to get started and requires no engineering resources; the downside is that the query list it gives you is the one it thinks you should track — not the one your target customers are actually asking.

Path 2: Build your own first-party measurement — you decide which questions to query, which model to use, and how to parse the results.

The two paths aren’t mutually exclusive, but for teams that genuinely want to understand how AI recommends their own brand, the first-party signal can’t be skipped. The reasons:

• Customized query list: The way your target customers actually phrase questions (tone, length, context) is something only you understand best; generic SaaS uses generalized templates
• Real-time comparison across model versions: When each new model generation launches, what you want to know is “how this new model changes citations of me,” not the SaaS vendor’s integration timeline
• Full record of query + response text: When you later need to get to the bottom of “why did citation rate drop this month,” you can only trace it back if you have the raw response text to analyze
• Cross-product data joins: When you want to join your GA4, form, and CRM data with LLM citation data for analysis, only first-party data can be directly SQL JOINed

What this article provides is a minimal architecture for “run it yourself, store it yourself, view it yourself.” It’s not meant to replace commercial SaaS — for most teams, running both paths in parallel is the most pragmatic: SaaS for the big-picture trends, your own build for the deep questions.

What Building Your Own Gives You vs. What It Doesn’t

In other words: building your own solves data collection, not data interpretation or strategy. The latter still requires GEO domain knowledge — and that’s the core value of consulting services (see the end of the article).

Below is the minimal implementation architecture.

System Architecture

┌─────────────────┐
│ collector.py    │ ← Runs daily at 03:00 (cron / Task Scheduler)
│ • Read queries  │
│ • Call 3 APIs   │
│ • Parse responses│
└────────┬────────┘
         │ Write
         ▼
┌─────────────────┐
│ monitoring.db   │ ← SQLite, 3 tables
│ • queries       │
│ • runs          │
│ • results       │
└────────┬────────┘
         │ Read
         ▼
┌─────────────────┐
│ dashboard.py    │ ← Run anytime (or on a schedule)
│ • Generate HTML │ ← Plot trends with Plotly
│ • Open browser  │
└─────────────────┘

Step 1: Environment Setup

# Python 3.10+
pip install openai anthropic httpx plotly

The .env file (never commit this into git):

OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
PERPLEXITY_API_KEY=pplx-...
BRAND_NAME=Your Company Name
BRAND_DOMAIN=example.com

The .gitignore:

.env
*.db
__pycache__/

Step 2: Define Your 20 Target Queries

queries.yaml:

queries:
  - id: 1
    text: "Recommended customer management systems for small and medium businesses"
    category: "product_recommend"
  - id: 2
    text: "Best CRM software worth using in 2026"
    category: "product_recommend"
  - id: 3
    text: "B2B marketing automation tool comparison"
    category: "comparison"
  - id: 4
    text: "How to choose a CRM vendor"
    category: "how_to"
  # ... 20 in total

Principles for choosing queries:

• Users actually ask them: Avoid ones where “I wish I were recommended” but nobody actually searches
• Have business conversion potential: The question maps to your product / service
• Spread across types: Product recommendation / comparison / how-to / trend analysis, etc.

Step 3: DB Schema

schema.sql:

-- Query list (stable)
CREATE TABLE IF NOT EXISTS queries (
    id INTEGER PRIMARY KEY,
    text TEXT NOT NULL,
    category TEXT,
    active INTEGER DEFAULT 1
);

-- Metadata for each run
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
);

-- Individual query results
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,      -- Full original response (for debugging)
    brand_mentioned INTEGER, -- 0/1
    brand_position INTEGER,  -- Which position it appeared in (NULL if not appearing)
    citation_url INTEGER,    -- Whether it contains a URL on my domain (Perplexity-specific)
    error TEXT,              -- API error message (if any)
    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 — The Collection Script

"""Runs daily: executes a query for each query × each platform, stores into the 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]:
    """Syncs from yaml to the DB, returns 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]:
    """Finds the position where the brand name appears in the response text.
    Returns (mentioned, position). position is "which listed option it is."""
    if not text or BRAND_NAME not in text:
        return False, None

    # Simplified: find all list items, see which item the brand appears in
    # In a real scenario you can switch to having the LLM parse it (higher cost but more accurate)
    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

    # Not in a list but mentioned → treat as mentioned but without ranking
    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:
    """The Perplexity API is OpenAI-compatible, but it returns 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)  # be rate-limit friendly

    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 — Visualization

"""Reads monitoring.db, generates an HTML dashboard and opens the browser."""

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. Daily overall appearance rate (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()

    # Reshape into 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. Appearance-rate trend chart
    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="Daily brand appearance rate (%) — by platform",
        xaxis_title="Date", yaxis_title="Appearance rate (%)",
        height=400,
    )

    # 3. Average-position trend
    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="Average recommendation position — by platform (lower is better)",
        xaxis_title="Date", yaxis_title="Average position",
        yaxis=dict(autorange="reversed"),  # 1 on top
        height=400,
    )

    # 4. Per-query hit heatmap for the last 7 days
    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="Hit-rate heatmap for the last 7 days",
        height=600,
    )

    # 5. Write the 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 Citation Monitor dashboard</h1>
<p>Updated at: <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 generated: {OUTPUT}")
    webbrowser.open(f"file://{Path(OUTPUT).absolute()}")


if __name__ == "__main__":
    main()

Step 6: Scheduling

Linux / macOS (cron)

# Run the collector daily at 03:00 in the morning
0 3 * * * cd /path/to/monitor && /path/to/python collector.py >> collector.log 2>&1

Windows (Task Scheduler)

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

Cost Estimate

20 queries × 3 platforms per day = 60 API calls.

Plus, if you run it once daily via a cloud function and use SQLite locally: total cost is about $7–15/month.

vs. commercial SaaS at $200–500/month, that’s a 95%+ saving.

Directions for Advanced Improvement

Improvement 1: Use an LLM to Parse the Brand Position in the Response

parse_brand_appearance has limited accuracy with regex against Chinese text + irregularly formatted LLM responses. Switch to a lightweight model (gpt-4o-mini / claude-haiku) that reads the response and asks “in which position is brand X mentioned.” Each call adds + $0.001 in cost, but accuracy improves dramatically.

Improvement 2: Multi-Brand (Including Competitor) Tracking

Change BRAND_NAME into a list (your own + 5 main competitors), and track each company’s appearance rate and position across each query simultaneously. Competitor comparison is the most persuasive data in GEO measurement.

Improvement 3: Alert Notifications

When:

• A platform’s appearance rate declines by > 20% for 3 consecutive days
• A high-priority query has 0 appearances for 7 consecutive days
• A run’s status is failed

→ send an email / push to Slack.

Improvement 4: Historical Backtracking and Weekly/Monthly Reports

• Weekly report: automatically email this week’s trends + the top improving / declining queries every Saturday
• Monthly report: automatically generate a PDF (matplotlib or weasyprint) and email it to the boss

First Step: Start with Steps 1–4 (the Minimal Viable Version)

Get collector.py + queries.yaml + schema.sql running, and gather data for 7 consecutive days. The first week’s numbers will serve as your “baseline.”

👉 GeoWeb Health Check provides a quantitative assessment of on-site signals, but real-time off-site citation rate can only be seen with this kind of active measurement script.

If you want to build a fully customized monitoring system (including multi-brand tracking, alert notifications, automated weekly/monthly reports, and team dashboard integration), that falls within the scope of GEO consulting services: [email protected]

The GEO deep-dive series. Previous article: “Subdomain vs. Subdirectory vs. Multi-Site Architecture — Impact on GEO and a Decision Framework”