2026-09-15, Rank Tracking · Tutorials

Rank-tracking pipeline for AI answers with async batches

A rank tracker for AI answers asks the same questions on a schedule and records where your domain appears among the sources each answer cites. The one built here uses async batches: a submission that is safe to re-run, collection by signed webhook, a reconciliation pass for anything that did not arrive, and the metrics you compute from the stored rows.

The code is Python with requests (plus a TypeScript variant of the submission using @answerline/sdk). It reads the API key from ANSWERLINE_API_KEY and talks to https://api.answerline.dev. For the system this pipeline sits inside at thousands of prompts, catalog and planner included, see programmatic AI visibility tracking.

What you are measuring

Citation rank is the position of the first cited source on your domain in an answer’s sources list, for one prompt, in one market, on one run. Track mention alongside it: whether the answer names your brand at all, cited or not. The two are different events with different fixes; mentions vs citations covers why.

Three properties of AI answers shape the pipeline:

• Answers vary between runs. The same prompt can cite you today and not tomorrow, so a single run is an observation, not a ranking. Store every run and aggregate later; see why one check is an anecdote.
• Answers are local. The market changes what gets cited, so it is part of the key, not a filter you apply afterwards. ChatGPT and Gemini answer from 213 countries, and Gemini also takes a US state.
• Prompts replace keywords. The prompt set defines the metric. Build it deliberately with prompt set design before you schedule anything.

Step 1: define the unit of work and its key

One task is one engine, one prompt, one market, one day. That tuple is also the task’s identity, so it makes the idempotency key:

def task_key(engine: str, prompt_id: str, country: str, day: str) -> str:
    return f"rank:{engine}:{prompt_id}:{country}:{day}"

Keys are unique across your account. Re-running a day’s submission after a crash, a timeout or a deploy creates nothing twice: items that already exist come back as RESOURCE_ALREADY_EXISTS inside the batch response. Include the engine in the key; a key of prompt, market and day alone collides as soon as you track a second engine. Use a stable prompt_id from your own table rather than the prompt text, so rewording a prompt is a deliberate new series. Idempotency keys explained goes deeper on key design.

Step 2: build and submit batches

POST /v1/async/task/batch takes a JSON array of 1 to 500 tasks. Each task carries taskType, the engine request as payload, and optionally priority (1 to 10, default 1, higher runs first), idempotencyKey and webhook: {"url": ...}.

import os
import requests

API = "https://api.answerline.dev"
MAX_BATCH = 500
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['ANSWERLINE_API_KEY']}"

def build_tasks(prompts: dict, countries: list, engines: list, day: str, webhook_url: str) -> list:
    return [
        {
            "taskType": engine,
            "payload": {"prompt": prompt, "country": country},
            "idempotencyKey": task_key(engine, prompt_id, country, day),
            "webhook": {"url": webhook_url},
        }
        for prompt_id, prompt in prompts.items()
        for country in countries
        for engine in engines
    ]

class OutOfBalance(Exception):
    pass

def submit(tasks: list) -> dict:
    """Submits every task; returns {idempotencyKey: task id} for tasks created by this call."""
    created = {}
    for start in range(0, len(tasks), MAX_BATCH):
        chunk = tasks[start:start + MAX_BATCH]
        res = session.post(f"{API}/v1/async/task/batch", json=chunk, timeout=120)
        res.raise_for_status()  # a 429 QUEUE_LIMIT_EXCEEDED lands here: nothing in this chunk was accepted
        for item in res.json()["results"]:
            key = chunk[item["index"]]["idempotencyKey"]
            if item["success"]:
                created[key] = item["task"]["id"]
                continue
            code = item["error"]["code"]
            if code == "INSUFFICIENT_CREDITS":
                raise OutOfBalance(key)
            if code != "RESOURCE_ALREADY_EXISTS":
                print("rejected", key, item["error"]["message"])
    return created

Persist created as soon as each chunk returns; reconciliation needs the ids, and a crash between chunks should not lose them. A re-run returns RESOURCE_ALREADY_EXISTS for tasks the first run created, so keep the mapping from the first run rather than expecting the second to rebuild it.

The same submission in TypeScript with the published SDK, whose createBatch refuses a list outside 1 to 500 before sending it:

import { Client, MAX_BATCH, type TaskInput } from "@answerline/sdk";

const client = new Client({ apiKey: process.env.ANSWERLINE_API_KEY! });

export async function submit(tasks: TaskInput[]): Promise<Map<string, string>> {
  const created = new Map<string, string>();
  for (let start = 0; start < tasks.length; start += MAX_BATCH) {
    const chunk = tasks.slice(start, start + MAX_BATCH);
    const batch = await client.createBatch(chunk);
    for (const item of batch.results) {
      const key = chunk[item.index].idempotencyKey!;
      if (item.success) created.set(key, item.task.id);
      else if (item.error.code === "INSUFFICIENT_CREDITS") throw new Error(`balance too low at ${key}`);
      else if (item.error.code !== "RESOURCE_ALREADY_EXISTS") console.warn(key, item.error.message);
    }
  }
  return created;
}

Always set the key yourself. The SDK generates a random one when you omit it, which protects a single call’s retries but not a re-run of the whole job the next hour.

The three submission failures

Before a large run, GET /v1/credits returns remaining, your balance in dollars, and every created item returns credits.creditsToCharge, the most it can cost in dollars. GET /v1/async/status shows queuedTasks, processingTasks and a priorityBreakdown while the run progresses.

To run some prompts first, for example a launch week’s, set "priority": 10 on their tasks. Priority orders only your own pending tasks.

Step 3: receive results by webhook

Each finished task is POSTed to the webhook URL. The body is exactly what GET /v1/async/task/{taskId} returns: task (with status COMPLETED or FAILED), credits, and response, which holds result on success or the error on failure. task.idempotencyKey is included, so the receiver can file a result under its prompt, market and day without a lookup.

The receiver has four jobs: verify the signature on the raw body, ignore dashboard test deliveries, store idempotently, and answer 2xx within 15 seconds.

import hashlib
import hmac
import json
import sqlite3
import time
from flask import Flask, abort, request

app = Flask(__name__)
db = sqlite3.connect("ranks.db", check_same_thread=False)
db.execute("create table if not exists deliveries (task_id text primary key, key text, status text, body text)")

def signature_valid(raw: bytes, header: str | None, secret: str, tolerance: int = 300) -> bool:
    pairs = [p.split("=", 1) for p in (header or "").split(",") if "=" in p]
    stamps = [v for k, v in pairs if k == "t"]
    sigs = [v for k, v in pairs if k == "v1"]
    if len(stamps) != 1 or not stamps[0].isdigit() or not sigs:
        return False
    if abs(time.time() - int(stamps[0])) > tolerance:
        return False
    expected = hmac.new(secret.encode(), stamps[0].encode() + b"." + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, s) for s in sigs)

@app.post("/hooks/ranks")
def receive():
    raw = request.get_data()
    if not signature_valid(raw, request.headers.get("Webhook-Signature"), os.environ["WEBHOOK_SECRET"]):
        abort(400)
    delivery = json.loads(raw)
    if delivery.get("test"):
        return "", 204
    task = delivery["task"]
    db.execute(
        "insert or ignore into deliveries values (?, ?, ?, ?)",
        (task["id"], task["idempotencyKey"], task["status"], raw.decode()),
    )
    db.commit()
    return "", 204

Store the raw delivery and parse it into metrics in a separate job. That keeps the response fast, and lets you recompute ranks later with a different rule without paying for the answers again.

Details that matter at volume, all from the webhooks guide:

• The signing secret (whsec_...) is on the Webhooks page of the dashboard. During a rotation the header carries two v1 values; accepting any match, as above, keeps deliveries valid through it.
• Webhook-Id is not signed. Deduplicate on task.id, which is inside the signed body; the insert or ignore on the primary key does that.
• Deliveries are concurrent and retried independently, so they arrive out of order and sometimes twice.
• URLs must be https and resolve to public addresses.

Designing a webhook receiver that survives outages covers buffering, backpressure and redelivery in more depth.

Step 4: reconcile what did not arrive

A delivery gets up to 10 attempts, with the last one roughly 21 to 43 minutes after the first. If your endpoint is down longer than that, the delivery is marked dead. The dashboard can redeliver it, but a pipeline should not depend on someone clicking. After each run, poll the tasks you created but never stored:

FINAL = ("COMPLETED", "FAILED")

def reconcile(created: dict) -> int:
    stored = {row[0] for row in db.execute("select task_id from deliveries")}
    filed = 0
    for key, task_id in created.items():
        if task_id in stored:
            continue
        res = session.get(f"{API}/v1/async/task/{task_id}", timeout=30)
        res.raise_for_status()
        body = res.json()
        if body["task"]["status"] in FINAL:
            db.execute("insert or ignore into deliveries values (?, ?, ?, ?)",
                       (task_id, key, body["task"]["status"], res.text))
            filed += 1
    db.commit()
    return filed

Only COMPLETED and FAILED are final. A poller can see PROCESSING go back to QUEUED, so do not treat PROCESSING as a promise of completion. A task that has not started within 4 hours of creation fails uncharged with details.cause QUEUE_WAIT_EXCEEDED; resubmit those with a new key.

Step 5: turn answers into ranks

Parse stored deliveries into one row per task:

from urllib.parse import urlsplit

def on_domain(url: str, domain: str) -> bool:
    host = (urlsplit(url).hostname or "").lower()
    return host == domain or host.endswith("." + domain)

def citation_rank(result: dict, domain: str) -> int | None:
    for source in sorted(result.get("sources", []), key=lambda s: s["position"]):
        if on_domain(source["url"], domain):
            return int(source["position"])
    return None

def to_row(delivery: dict, domain: str, brand: str) -> dict:
    task = delivery["task"]
    row = {"key": task["idempotencyKey"], "task_id": task["id"], "status": task["status"],
           "rank": None, "mentioned": None, "cited_sources": None}
    if task["status"] != "COMPLETED":
        return row
    result = delivery["response"]["result"]
    row["rank"] = citation_rank(result, domain)
    row["cited_sources"] = len(result.get("sources", []))
    row["mentioned"] = brand.lower() in (result.get("text") or "").lower()
    return row

Notes on the fields:

• sources[].position is the source’s place in that engine’s source list, and every chat engine plus AI Mode returns sources[] with position and url. Positions are comparable within an engine over time, not across engines: each engine lays out citations differently.
• entities[] (type, name) exists on ChatGPT responses only, where it lists the brands and products the answer names. The sketch matches the brand in result.text so the method is the same on every engine; tune it for brand name variants.
• citationPills[] records the inline chips inside the answer text. Group entries by citationPillId to rebuild each chip. ChatGPT’s ids are 0-based; the other engines that return pills use 1-based ids. A source shown as an inline chip is more visible than one only in the source list; store both if your reporting distinguishes them.
• A FAILED task is not charged and its response holds the error. Keep its row with an empty rank, so gaps in coverage stay visible instead of silently reading as “not cited”.

Step 6: aggregate into metrics

Because answers vary between runs, report rates and distributions over a window, not a single day’s number:

• Citation rate: share of completed runs where rank is not empty, per prompt cluster, engine and market.
• Median rank when cited: the typical position, computed only over runs where you were cited.
• Mention rate: share of runs that name the brand, cited or not.
• Coverage: completed runs divided by scheduled runs. Falling coverage is a pipeline problem, not a visibility change.

For how many runs a prompt needs before a rate is worth reporting, see sampling vs census; for share-of-voice rollups, see the AI share of voice framework.

Scheduling and cost

Start the submission from any scheduler once per period: cron, a CI schedule, or an n8n workflow. Pick the period per use case with monitoring cadence.

There is no concurrency cap and no per-account request rate limit. The async queue holds about 100,000 tasks of a typical prompt, more of shorter ones. See the rate limits page.

Worked example:

Run the first row daily for 30 days and it costs $8.10 at Starter prices. Every new account gets 100 free requests on any live engine, enough to validate the pipeline end to end before you top up; the tiers are on the pricing page. Failed tasks cost nothing, so these are upper bounds. Estimating your monitoring bill walks through sizing in more detail.

Pitfalls

1. Random keys. A key generated per attempt makes every retry a new paid task. Derive keys from meaning.
2. Parsing before verifying. Re-serialized JSON does not match the signature; verify the raw bytes first.
3. Slow webhook handlers. Anything that does not answer 2xx within 15 seconds is retried and duplicates your work. Store, answer, then process.
4. Treating a missing row as “not cited”. Distinguish failed and unreconciled tasks from answers that cited no one on your domain.
5. Comparing positions across engines. Normalize per engine, or report citation rate, which is comparable.
6. Reusing a failed task’s key inside 24 hours. The key stays taken until then; retry sooner with a suffix such as :retry1.

The rank tracking use case shows the same approach for Google organic results; start building with the quickstart and the async tasks docs.

Keep reading