Build a Brand-Mention Monitor Across Four LLMs in an Afternoon
A working script that runs your prompt set against ChatGPT, Claude, Gemini, and Perplexity on a schedule, then logs where your brand shows up and who gets cited.
You can buy a dashboard that tells you how often an assistant names your company. You can also build a rougher version of the same thing yourself before dinner, and the building teaches you things the dashboard hides. A monitor that runs your own prompts against the four assistants that matter, records the raw answers, and scores mentions on your terms will show you exactly where the visibility comes from and where it leaks. This walkthrough gives you the whole thing in Python: the storage layer, four provider adapters behind one interface, retry logic that survives rate limits, brand matching that catches misspellings, and a roll-up into Share of Voice and Share of Retrieval over time.
The design goal is boring on purpose. Every engine gets wrapped so the rest of the code never knows which vendor it is talking to. That single decision is what keeps the project small, because parsing, matching, and reporting all operate on one normalized record no matter where the answer came from.
What the monitor actually records
Start from the row you want at the end, then work backward to the code that fills it. For each run you want one record per prompt per engine, holding the timestamp, the engine name, the prompt text, the full answer, whether your brand appeared, where it appeared in the answer, and every citation URL the engine returned. Store that and you can compute anything later without re-running a single API call.
The RAG pipeline behind these assistants is the reason citations deserve their own column. A conversational prompt gets rewritten into several sub-queries, each sub-query retrieves a candidate set from a search backend, a reranker trims that pool to a few passages, and only then does the model write an answer with citations stapled on. Being named in the prose and being cited by URL are two different events, and a monitor that collapses them loses the more useful signal. So the schema keeps mentions and citations apart.
# storage.py
import sqlite3
from contextlib import contextmanager
DB_PATH = "mentions.db"
# ts holds an ISO 8601 UTC string. engine is one of chatgpt, claude,
# gemini, perplexity. prompt_id is the stable id from your prompt store.
# mentioned is 0 or 1, mention_ct counts hits, first_pos is a char offset
# or NULL, and error is populated only when the call failed.
SCHEMA = """
CREATE TABLE IF NOT EXISTS runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
engine TEXT NOT NULL,
prompt_id TEXT NOT NULL,
prompt TEXT NOT NULL,
answer TEXT NOT NULL,
mentioned INTEGER NOT NULL,
mention_ct INTEGER NOT NULL,
first_pos INTEGER,
error TEXT
);
CREATE TABLE IF NOT EXISTS citations (
run_id INTEGER NOT NULL REFERENCES runs(id),
url TEXT NOT NULL,
domain TEXT NOT NULL,
is_brand INTEGER NOT NULL
);
"""
@contextmanager
def connect():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
finally:
conn.close()
def init_db():
with connect() as conn:
conn.executescript(SCHEMA)SQLite is the right call for something you run on a laptop or a cron box. One file, no server, and it holds tens of millions of rows without complaint. If you later want the data in a warehouse, a nightly sqlite3 mentions.db .dump or a small pandas export covers it. Do not reach for Postgres until you have a reason.
One interface, four adapters
Every provider speaks a different dialect. OpenAI and Anthropic want you to read a nested content array, Gemini returns candidates, and Perplexity hands back an OpenAI-shaped body with a citations field that the others do not have. Rather than let those differences leak into your logic, define one return shape and make each adapter conform to it.
# adapters.py
from dataclasses import dataclass, field
from typing import Protocol
@dataclass
class EngineResponse:
engine: str
text: str # the answer as plain text
citations: list[str] = field(default_factory=list)
class Engine(Protocol):
name: str
def ask(self, prompt: str) -> EngineResponse: ...With the contract fixed, each adapter is a thin translation layer. The ChatGPT and Claude adapters below use the vendor SDKs; swap in requests if you would rather hold the raw HTTP. Keys come from the environment so nothing sensitive lands in the repo.
import os
from openai import OpenAI # pip install openai
from anthropic import Anthropic # pip install anthropic
class ChatGPT:
name = "chatgpt"
def __init__(self):
# export OPENAI_API_KEY=sk-...
self.client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
def ask(self, prompt: str) -> EngineResponse:
# The web-search tool is what makes citations appear. Without it
# you are testing the base model's memory, which is a different study.
resp = self.client.responses.create(
model="gpt-4.1", # pick the model you care about
tools=[{"type": "web_search"}],
input=prompt,
)
text = resp.output_text
urls = []
for item in resp.output:
for part in getattr(item, "content", []) or []:
for ann in getattr(part, "annotations", []) or []:
if getattr(ann, "url", None):
urls.append(ann.url)
return EngineResponse(self.name, text, dedupe(urls))
class Claude:
name = "claude"
def __init__(self):
# export ANTHROPIC_API_KEY=sk-ant-...
self.client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
def ask(self, prompt: str) -> EngineResponse:
resp = self.client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
tools=[{"type": "web_search_20250305", "name": "web_search"}],
messages=[{"role": "user", "content": prompt}],
)
text, urls = [], []
for block in resp.content:
if block.type == "text":
text.append(block.text)
for cite in getattr(block, "citations", []) or []:
if getattr(cite, "url", None):
urls.append(cite.url)
return EngineResponse(self.name, "".join(text), dedupe(urls))Gemini and Perplexity round out the set. Gemini exposes retrieved sources through its grounding metadata when you enable the search tool, and Perplexity returns citations as a plain list on the response body, so both map cleanly onto the same EngineResponse.
import requests
import google.generativeai as genai # pip install google-generativeai
class Gemini:
name = "gemini"
def __init__(self):
# export GOOGLE_API_KEY=...
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
self.model = genai.GenerativeModel(
"gemini-2.5-flash",
tools="google_search_retrieval",
)
def ask(self, prompt: str) -> EngineResponse:
resp = self.model.generate_content(prompt)
text = resp.text
urls = []
for cand in resp.candidates:
meta = getattr(cand, "grounding_metadata", None)
for chunk in getattr(meta, "grounding_chunks", []) or []:
web = getattr(chunk, "web", None)
if web and getattr(web, "uri", None):
urls.append(web.uri)
return EngineResponse(self.name, text, dedupe(urls))
class Perplexity:
name = "perplexity"
ENDPOINT = "https://api.perplexity.ai/chat/completions"
def __init__(self):
# export PERPLEXITY_API_KEY=pplx-...
self.key = os.environ["PERPLEXITY_API_KEY"]
def ask(self, prompt: str) -> EngineResponse:
headers = {"Authorization": f"Bearer {self.key}"}
body = {
"model": "sonar",
"messages": [{"role": "user", "content": prompt}],
}
r = requests.post(self.ENDPOINT, json=body, headers=headers, timeout=60)
r.raise_for_status()
data = r.json()
text = data["choices"][0]["message"]["content"]
urls = data.get("citations", [])
return EngineResponse(self.name, text, dedupe(urls))
def dedupe(urls: list[str]) -> list[str]:
seen, out = set(), []
for u in urls:
if u not in seen:
seen.add(u)
out.append(u)
return outNotice what the four adapters share once they return: nothing vendor-specific survives past the EngineResponse. Everything downstream reads .text and .citations, so adding a fifth engine later means writing one class and registering it, with no edits anywhere else.
Surviving rate limits without babysitting
Run a hundred prompts across four engines and you will hit a 429 within the first few minutes. The fix is exponential backoff with jitter, wrapped around every call so no adapter has to know about it. Retry on transient failures such as rate limits and 5xx responses, give up fast on a 400 that will never succeed, and cap the total wait so a dead endpoint cannot stall the whole run.
# retry.py
import random
import time
TRANSIENT = {429, 500, 502, 503, 504}
def call_with_retry(fn, *, max_attempts=5, base=1.0, cap=30.0):
"""Run fn(), retrying transient failures with backoff and jitter."""
last_exc = None
for attempt in range(max_attempts):
try:
return fn()
except Exception as exc: # narrow this to your SDK's types
status = getattr(exc, "status_code", None)
if status is not None and status not in TRANSIENT:
raise # a 400 or 401 will not improve
last_exc = exc
if attempt == max_attempts - 1:
break
sleep = min(cap, base * (2 ** attempt))
sleep = sleep * (0.5 + random.random()) # full jitter
time.sleep(sleep)
raise last_excThe jitter matters more than it looks. Without it, every retry across your prompt batch lines up on the same schedule and slams the endpoint in synchronized waves, which is how a transient limit turns into a sustained one. Randomizing the wait spreads the load and lets the batch drain. Keep concurrency modest too: four engines running one prompt each in parallel is fine, forty parallel prompts per engine will get you throttled and teach you nothing extra.
Catching the mention, even when it is misspelled
A brand name is easy to match until it is not. Assistants abbreviate, drop suffixes, and occasionally misspell, so a naive substring search undercounts. Run an exact pass first because it is cheap and unambiguous, then fall back to fuzzy matching on a token window for the cases the exact pass misses. The rapidfuzz library gives you fast, sane ratios for the fuzzy stage.
# matching.py
import re
from rapidfuzz import fuzz # pip install rapidfuzz
def find_mentions(answer: str, brand: str, aliases: list[str],
fuzzy_threshold: int = 88) -> dict:
"""Return mention count and the char offset of the first hit."""
needles = [brand, *aliases]
positions = []
# Exact pass, case-insensitive, on word boundaries.
for needle in needles:
pattern = r"\b" + re.escape(needle) + r"\b"
for m in re.finditer(pattern, answer, flags=re.IGNORECASE):
positions.append(m.start())
# Fuzzy pass: slide over tokens and score against the brand.
if not positions:
tokens = list(re.finditer(r"\S+", answer))
span = max(len(brand.split()), 1)
for i in range(len(tokens)):
window = tokens[i:i + span]
if not window:
continue
phrase = answer[window[0].start():window[-1].end()]
if fuzz.ratio(phrase.lower(), brand.lower()) >= fuzzy_threshold:
positions.append(window[0].start())
positions.sort()
return {
"mentioned": 1 if positions else 0,
"count": len(positions),
"first_pos": positions[0] if positions else None,
}Tune the threshold against your own name rather than trusting the default. A short brand like "Arc" collides with ordinary words and needs a high bar plus a tight alias list, while a distinctive multi-word name tolerates a looser one. Test on real answers before you believe the numbers, because a matcher that over-fires is worse than one that misses: a false mention inflates every metric built on top of it.
Citation matching is simpler and more valuable. For each URL, pull the registered domain and flag whether it is yours. That single boolean drives the split between mentions the model recalled and mentions it retrieved from a live source.
from urllib.parse import urlparse
def classify_citation(url: str, brand_domains: set[str]) -> dict:
host = (urlparse(url).hostname or "").lower().lstrip("www.")
return {
"url": url,
"domain": host,
"is_brand": 1 if any(host.endswith(d) for d in brand_domains) else 0,
}Wiring the run and putting rows in the table
The runner holds no cleverness. It loops over the prompt store, asks each engine through the retry wrapper, scores the answer, and writes one row plus its citations. Failures get recorded rather than raised, so a single dead engine does not lose you the rest of the batch.
# runner.py
from datetime import datetime, timezone
from storage import connect
from retry import call_with_retry
from matching import find_mentions, classify_citation
BRAND = "Acme"
ALIASES = ["Acme Inc", "AcmeHQ", "Acme.io"]
BRAND_DOMAINS = {"acme.io"}
def run_once(engines, prompts):
ts = datetime.now(timezone.utc).isoformat()
with connect() as conn:
for p in prompts: # prompts: list of (id, text)
prompt_id, prompt = p
for engine in engines:
row = {
"ts": ts, "engine": engine.name,
"prompt_id": prompt_id, "prompt": prompt,
"answer": "", "mentioned": 0, "mention_ct": 0,
"first_pos": None, "error": None,
}
citations = []
try:
resp = call_with_retry(lambda: engine.ask(prompt))
m = find_mentions(resp.text, BRAND, ALIASES)
row.update(answer=resp.text, mentioned=m["mentioned"],
mention_ct=m["count"], first_pos=m["first_pos"])
citations = [classify_citation(u, BRAND_DOMAINS)
for u in resp.citations]
except Exception as exc:
row["error"] = repr(exc)
cur = conn.execute(
"""INSERT INTO runs
(ts, engine, prompt_id, prompt, answer,
mentioned, mention_ct, first_pos, error)
VALUES (:ts, :engine, :prompt_id, :prompt, :answer,
:mentioned, :mention_ct, :first_pos, :error)""",
row,
)
for c in citations:
conn.execute(
"""INSERT INTO citations (run_id, url, domain, is_brand)
VALUES (?, ?, ?, ?)""",
(cur.lastrowid, c["url"], c["domain"], c["is_brand"]),
)Your prompt store can be as plain as a YAML file read into that list of tuples. Give every prompt a stable id and never recycle ids across meanings, because the id is the thread that ties a prompt's history together as you edit its wording. Schedule the whole thing with cron and you have a time series building itself:
# run every morning at 06:00, once per prompt per engine
0 6 * * * cd /opt/mention-monitor && /usr/bin/python3 -m runner >> run.log 2>&1From rows to Share of Voice and Share of Retrieval
Now the payoff. Share of Voice, in this context, is the fraction of answers that name you against the total that name you or any competitor you track. Compute it per engine and over time, and the trend line tells you whether your visibility is climbing or eroding on each surface independently.
# report.py
import pandas as pd
from storage import connect
def load_runs() -> pd.DataFrame:
with connect() as conn:
return pd.read_sql_query(
"SELECT ts, engine, prompt_id, mentioned FROM runs "
"WHERE error IS NULL", conn, parse_dates=["ts"])
def share_of_voice(df: pd.DataFrame, freq: str = "W") -> pd.DataFrame:
"""Weekly share of prompts where the brand was mentioned, by engine."""
df = df.copy()
df["bucket"] = df["ts"].dt.to_period(freq).dt.start_time
grp = df.groupby(["bucket", "engine"])
sov = grp["mentioned"].mean().rename("share_of_voice").reset_index()
return sov.pivot(index="bucket", columns="engine",
values="share_of_voice")Share of Retrieval sits upstream of that and is the more honest early warning. It measures how often your content shows up in the material the engine actually pulled, before the model decides whether to name you. You approximate it with the citation table: across all runs, how often did a live answer cite one of your own pages? A brand can score high on voice through the model's training memory while scoring near zero on retrieval, and that gap is fragile, because the moment the model leans on live search you vanish.
def share_of_retrieval(freq: str = "W") -> pd.DataFrame:
"""Weekly fraction of runs whose citations include a brand-owned URL."""
q = """
SELECT r.ts AS ts, r.engine AS engine, r.id AS run_id,
MAX(c.is_brand) AS retrieved
FROM runs r
LEFT JOIN citations c ON c.run_id = r.id
WHERE r.error IS NULL
GROUP BY r.id
"""
with connect() as conn:
df = pd.read_sql_query(q, conn, parse_dates=["ts"])
df["retrieved"] = df["retrieved"].fillna(0).astype(int)
df["bucket"] = df["ts"].dt.to_period(freq).dt.start_time
return (df.groupby(["bucket", "engine"])["retrieved"]
.mean().reset_index()
.pivot(index="bucket", columns="engine", values="retrieved"))Read the two series side by side. When voice holds steady and retrieval climbs, your Citation Surface is widening and the position is getting sturdier under it. When voice holds and retrieval falls, you are living on borrowed memory and the next model refresh can wipe the gain. The reason to watch retrieval first is that it moves earlier: it is the leading indicator, and by the time voice drops the cause is already weeks old.
One caveat keeps this honest. Most brand mentions in AI answers trace back to third-party pages rather than your own domain, so a strict "is it my URL" test undercounts your true retrieval footprint. Widen BRAND_DOMAINS into a broader Citation Surface set that includes the review sites, docs, and community pages where your name reliably appears, and the retrieval number starts reflecting the ground the model actually stands on. Track the raw domains too, because the list of who gets cited for your category is itself a map of where to go earn coverage.
Where a hosted tool earns its keep
Everything above runs in an afternoon and is genuinely yours after that. The maintenance is the part nobody warns you about. Provider SDKs change their response shapes, model names get deprecated, the citation format on one engine shifts and silently zeroes a column, and your prompt set needs constant expansion to stay honest about what customers ask. A monitor is a living system, and keeping four vendor integrations current is a standing tax on your week.
That is the burden a hosted service like shareof.ai absorbs: the adapters stay current, the runs happen on schedule across ChatGPT, Claude, Gemini, and Perplexity, and Share of Voice and Share of Retrieval roll up without a person re-running the batch or patching a broken parser. The framework in this article stands on its own and is worth building at least once, because owning the raw records teaches you how visibility is actually produced. When the plumbing becomes a chore rather than a lesson, hand it off and spend the time on the content that moves the numbers.
A note on measurement
Two numbers deserve your skepticism before you act on them. Sampling variance is real: assistants are stochastic, so one run per prompt is a noisy estimate, and you should average several runs per prompt per day before you trust a small move. And the retrieval approximation from citations is a floor, not a full count, since an engine can read a page it never links. Treat the trend as the signal and any single day as noise, widen your Citation Surface deliberately, and verify the matcher against real answers in your own category rather than assuming the defaults fit your name.
Primary sources and evidence
- OpenAI web search tool — source annotations and grounded responses.
- Gemini grounding with Google Search — search grounding and citation metadata.
- Perplexity Search API — retrieving ranked web results programmatically.
Continue through the system
- Measure Your Own Share of Retrieval With the Model APIs
- Reading Your AI Crawler Logs: GPTBot, ClaudeBot, PerplexityBot and Friends
- Share of Voice vs Share of Model vs Share of Retrieval
Run the practical layer with the free AI visibility scan, explore the AI search library, or see the AI visibility benchmarks.