Building a citation verification step into research agents

Citation verification research agents must do more than emit URLs alongside claims; they need a deterministic post-generation check that the sources actually support the statements. This tutorial builds a minimal but production-minded pipeline that generates a cited answer, fetches each source, and asks a model to confirm the citation holds.

Prerequisites

• Python 3.11 or newer
• pip install openai httpx beautifulsoup4
• An API key for an OpenAI-compatible endpoint. We’ll use n4n.ai’s gateway, which exposes one OpenAI-compatible endpoint addressing 240+ models and provides automatic fallback when a provider is degraded.
• Environment variable N4N_API_KEY set.

Architecture

The pipeline has four stages:

1. Draft – an LLM answers a query with inline markdown citations [1](https://...).
2. Parse – extract each citation’s URL and the sentence that contains it.
3. Fetch – retrieve the HTML, strip markup, and keep raw text.
4. Verify – a second LLM call judges whether the source text supports the claim.

This separation keeps the verification step independent of the generator’s biases.

Step 1: Generate a draft with citations

We point the OpenAI SDK at the compatible base URL and force the model to cite.

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"],
)

def generate_draft(query: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.2,
        messages=[
            {"role": "system", "content": 
             "Answer the user concisely. Use inline markdown citations as [n](url). "
             "Only cite URLs you are confident exist and are relevant."},
            {"role": "user", "content": query}
        ],
    )
    return resp.choices[0].message.content

draft = generate_draft("What are the main failure modes of distributed SQLite?")
print(draft)

Expected output resembles:

Distributed SQLite deployments often struggle with write contention [1](https://sqlite.org/whentouse.html)
and network partition tolerance [2](https://en.wikipedia.org/wiki/CAP_theorem).

Step 2: Extract citations and claims

We map each URL to the sentence that cited it. A regex catches [n](url); sentence splitting is naive but adequate.

import re

def extract_citations(text: str):
    cit_re = re.compile(r"\[(\d+)\]\((https?://[^)]+)\)")
    sentences = re.split(r"(?<=[.!?])\s+", text)
    out = []
    for sent in sentences:
        for m in cit_re.finditer(sent):
            out.append({
                "id": m.group(1),
                "url": m.group(2),
                "sentence": sent.strip()
            })
    return out

citations = extract_citations(draft)
print(citations)

Output:

[
  {"id": "1", "url": "https://sqlite.org/whentouse.html", "sentence": "Distributed SQLite deployments often struggle with write contention [1](https://sqlite.org/whentouse.html)"},
  {"id": "2", "url": "https://en.wikipedia.org/wiki/CAP_theorem", "sentence": "and network partition tolerance [2](https://en.wikipedia.org/wiki/CAP_theorem)."}
]

Step 3: Fetch and sanitize source pages

Never trust a URL without a fetch. Use httpx with a short timeout and BeautifulSoup to drop scripts.

import httpx
from bs4 import BeautifulSoup

def fetch_text(url: str, timeout: int = 10) -> str | None:
    try:
        r = httpx.get(
            url,
            timeout=timeout,
            follow_redirects=True,
            headers={"User-Agent": "Mozilla/5.0 (citation-verifier)"}
        )
        r.raise_for_status()
        soup = BeautifulSoup(r.text, "html.parser")
        for tag in soup(["script", "style", "noscript"]):
            tag.extract()
        return soup.get_text(separator=" ", strip=True)[:15000]
    except Exception as e:
        print(f"fetch failed {url}: {e}")
        return None

For the two example URLs this returns the cleaned article text (truncated to 15k chars to bound token spend).

Step 4: Verify each citation against the source

A second model call acts as a binary judge. We request strict JSON.

import json

def verify_claim(claim: str, source: str) -> dict:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": 
             "You are a strict fact checker. Given a claim and source text, "
             "return JSON {\"supported\": boolean, \"reason\": string}."},
            {"role": "user", "content": f"CLAIM: {claim}\n\nSOURCE: {source}"}
        ],
    )
    return json.loads(resp.choices[0].message.content)

def verify_citations(answer: str):
    report = []
    for c in extract_citations(answer):
        text = fetch_text(c["url"])
        if not text:
            report.append({**c, "status": "fetch_failed"})
            continue
        res = verify_claim(c["sentence"], text)
        report.append({
            **c,
            "status": "verified" if res.get("supported") else "unsupported",
            "reason": res.get("reason", "")
        })
    return report

report = verify_citations(draft)
print(json.dumps(report, indent=2))

Expected report snippet:

[
  {
    "id": "1",
    "url": "https://sqlite.org/whentouse.html",
    "sentence": "Distributed SQLite deployments often struggle with write contention [1](https://sqlite.org/whentouse.html)",
    "status": "verified",
    "reason": "Page states SQLite is not designed for high concurrency writes."
  },
  {
    "id": "2",
    "url": "https://en.wikipedia.org/wiki/CAP_theorem",
    "sentence": "and network partition tolerance [2](https://en.wikipedia.org/wiki/CAP_theorem).",
    "status": "unsupported",
    "reason": "CAP theorem page does not mention SQLite specifically."
  }
]

The second citation is flagged because the model bolted a generic claim onto an unrelated source. That is exactly the failure citation verification research agents must catch.

Step 5: Act on the report

Verification is useless if the agent ignores it. Two options:

• Filter: drop sentences with unsupported or fetch_failed status from the final answer.
• Annotate: append a warning footnote.

def rebuild_answer(answer: str, report):
    bad_ids = {r["id"] for r in report if r["status"] != "verified"}
    if not bad_ids:
        return answer
    sents = re.split(r"(?<=[.!?])\s+", answer)
    kept = [s for s in sents if not any(f"[{i}](" in s for i in bad_ids)]
    return " ".join(kept) + "\n\n⚠ Removed claims with unverified citations."

Production considerations

Most citation verification research agents skip the fetch step because it adds latency. In practice, the fetch is the only ground truth you have.

• Concurrency: wrap fetch_text and verify_claim in asyncio with httpx.AsyncClient to verify dozens of citations in parallel.
• Caching: store fetched text keyed by URL+etag. Source pages rarely change within a research run.
• Cost control: the verification calls can double token spend. Using n4n.ai’s per-token usage metering you can attribute cost per agent step and set budgets.
• Robust parsing: real drafts mix parentheses, footnotes, and HTML. Upgrade the extractor to a markdown AST (e.g., markdown-it-py) before shipping.
• Model routing: for verification, a smaller model is fine. Honor client routing directives to send verification to a cheap model while the draft uses a flagship.

Where this breaks

The judge model inherits source truncation limits; a claim supported only in a comment section past byte 15k will be marked unsupported. Increase the fetch window or use a search-in-page index if that matters.

Also, some sites block bots. Respect robots.txt and provide a fetch_failed fallback that down-weights rather than deletes the claim.

Wrap-up

You now have a working loop for citation verification research agents: generate, extract, fetch, judge, and rebuild. The code is ~120 lines and swaps into any OpenAI-compatible stack. Start with the synchronous version, then harden the fetch layer before scaling to autonomous deep-research workloads.