factivelabs 0.2.0

Python SDK for the Factive fact-checking API. Stream LLM output and get verified claims back in real time.

factivelabs

Python SDK for the Factive Labs fact-checking API. Stream an LLM's output through the verifier as it generates and get back fact-checked claims in real time.

Install

pip install factivelabs

Quickstart

Get an API key at factivelabs.com/dashboard.

Reusing connections across many calls

For tight loops or long-lived workers, use the client as a context manager — one TCP/TLS connection is kept alive across method calls instead of being torn down and remade every time:

with FactiveClient(api_key="fk_...") as client:
    for doc in many_docs:
        result = client.verify_text(doc)

(Outside a with block, every method call still creates and tears down its own client — fine for one-shots, just not optimal for hot loops.)

Streaming an LLM's output

from factivelabs import FactiveClient

client = FactiveClient(api_key="fk_...")

# Any iterable of strings works — Anthropic, OpenAI, Perplexity, your own.
def my_llm_stream():
    for token in some_llm.stream("What's happening in tech this week?"):
        yield token

result = client.verify_stream(
    my_llm_stream(),
    context="What's happening in tech this week?",
    on_token=lambda t: print(t, end="", flush=True),
    on_claim=lambda c: print(f"\n  → [{c.verdict}] {c.text}"),
)

print(f"\n\nFact-checked {len(result.claims)} claims.")
print(f"Confirmed: {len(result.confirmed)} | Disputed: {len(result.disputed)}")

if not result.byte_equality:
    # Sanity check: every byte of the LLM's output should have made it
    # through to the API. If not, the result is incomplete.
    print(f"WARNING: only {result.chars_received}/{result.chars_sent} chars confirmed")

One-shot fact-check of static text

result = client.verify_text(
    "The Great Wall of China is visible from low Earth orbit with the naked eye.",
)
for claim in result.claims:
    print(claim.verdict, "—", claim.text)

Fact-check a URL

result = client.verify_url("https://example.com/article")
print(result.title)               # auto-extracted document title
print(result.extracted_text[:200])  # the plain text we fact-checked
for claim in result.claims:
    print(claim.verdict, "—", claim.text)

YouTube and TikTok URLs are auto-detected — the SDK uses the right transcript extractor under the hood.

Fact-check a PDF, Word doc, or image

# From a path on disk:
result = client.verify_file("/path/to/report.pdf")

# Or from raw bytes already in memory:
with open("/path/to/photo.jpg", "rb") as f:
    result = client.verify_file(f.read(), filename="photo.jpg")

print(result.title, "—", len(result.claims), "claims")

Supported formats: PDF, Word (.docx), and images (PNG / JPG / etc. — text is extracted via OCR).

Just extract text (no fact-checking)

extracted = client.extract_text(url="https://example.com/article")
print(extracted.title, extracted.char_count)
print(extracted.text)

Batch fact-checking (1–100 documents)

For backlog processing — submit many documents at once and poll or wait for results.

submission = client.submit_batch([
    {"text": "First doc."},
    {"url": "https://example.com/article"},
    {"file": "/path/to/report.pdf"},
])

# Block until everything finishes:
results = client.wait_for_batch(
    submission,
    timeout=600,                              # seconds
    on_progress=lambda jid, st: print(jid, st.status),
)

for job_id, status in results.items():
    if status.is_complete:
        for claim in status.result.claims:
            print(claim.verdict, claim.text)
    elif status.is_failed:
        print(job_id, "FAILED:", status.error)

Or submit and poll yourself:

submission = client.submit_batch([{"text": "doc"}], webhook_url="https://my-app/hook")
status = client.get_job(submission.jobs[0].job_id)
print(status.status, status.progress)

Per-plan batch limits: free=5, pro=50, enterprise=100.

Private corpus management

For customers using Athena's private-corpus mode — upload documents the verifier will use as the source of truth instead of the public web.

# Upload one or more files. Accepts paths or (filename, bytes) tuples.
result = client.corpus.upload([
    "/path/to/handbook.pdf",
    ("policy.docx", open("/path/to/policy.docx", "rb").read()),
])
# {"queued": [...], "failed": [...]}

# List the docs currently in the corpus:
for doc in client.corpus.list():
    print(doc.file_name, "—", doc.status)   # queued → processing → embedding → ready

# Delete one:
client.corpus.delete("doc_abc123")

# Wipe everything (requires explicit confirm):
client.corpus.clear(confirm=True)

# Scope description (improves routing accuracy):
scope = client.corpus.get_scope()
print(scope.scope_text)

client.corpus.update_scope("Internal HR policies, benefits, and the employee handbook.")
client.corpus.regenerate_scope()  # auto-generate from current corpus contents

Once the corpus is populated and the scope is set, fact-check against it with use_private_corpus=True:

result = client.verify_text(
    "Some document.",
    use_private_corpus=True,
    corpus_scope="Internal HR policies and benefits.",  # optional but helps routing
)

Async / await

For FastAPI, asyncio scripts, or any async codebase:

import asyncio
from factivelabs import AsyncFactiveClient

async def main():
    client = AsyncFactiveClient(api_key="fk_...")

    # Same methods, all awaitable:
    result = await client.verify_text("Some document.")
    result = await client.verify_url("https://example.com/article")
    result = await client.verify_file("/path/to/report.pdf")

    # Streaming works with sync OR async iterables, and accepts async callbacks:
    async def on_claim(claim):
        await save_to_db(claim)

    result = await client.verify_stream(my_async_llm_stream(), on_claim=on_claim)

asyncio.run(main())

Why streaming?

Most fact-check APIs make you wait for your LLM to finish before they can do anything. By the time the answer is on screen, the user has already read it — and any hallucinations along with it.

verify_stream lets the verifier work paragraph-by-paragraph as the LLM generates. Verified claims start arriving on on_claim while the LLM is still typing, so your UI can highlight, annotate, or block content in real time.

How the streaming protocol works (and why you don't have to think about it)

Under the hood: the SDK buffers your LLM's tokens locally until it sees a paragraph break (\n\n), then posts that paragraph as a single atomic HTTP request to the API. Each post carries a complete, self-contained unit of text. The SDK retries on transient failures (5xx, network blips), surfaces hard rejects (4xx) immediately, and reports a byte-equality check at the end so you can confirm nothing was dropped.

You shouldn't need to think about any of this — it's why the SDK exists. But if you want to talk to the API directly, the contract is documented at api.factivelabs.com/docs.

Callbacks

client.verify_stream(
    token_iterator,
    context="user question here",
    on_token=lambda token: ...,             # every token, as it arrives
    on_paragraph_sent=lambda paragraph: ..., # after each paragraph POST succeeds
    on_claim=lambda claim: ...,             # each verified claim
    on_event=lambda name, data: ...,        # low-level SSE event hook
)

Threading note (sync client only): on_token fires from your main thread (where the iterator runs). on_claim, on_event, and on_paragraph_sent fire from a background thread that reads the API's SSE stream. If you mutate shared state inside multiple callbacks, use a thread-safe primitive (queue.Queue, threading.Lock, etc.) — bare dict[k] += 1 is not safe across these threads. The async client (AsyncFactiveClient) sidesteps this entirely: every callback runs on the event loop.

Streaming as an iterator (recommended for web frameworks)

When you want to plumb streaming events into a Flask SSE response, FastAPI StreamingResponse, websocket frame, or anything else that wants a generator, use verify_stream_events. It yields typed StreamEvent objects as they happen — no callbacks, no thread management.

for ev in client.verify_stream_events(my_llm_tokens):
    if ev.type == "token":
        print(ev.text, end="", flush=True)
    elif ev.type == "claim":
        print(f"\n→ [{ev.claim.verdict}] {ev.claim.text}")
    elif ev.type == "done":
        print(f"\nbyte_equality={ev.result.byte_equality}")

Flask SSE proxy (one-screen recipe)

import json
from flask import Response, stream_with_context
from factivelabs import FactiveClient

client = FactiveClient(api_key=...)

@app.route("/run-stream")
def run_stream():
    @stream_with_context
    def generate():
        for ev in client.verify_stream_events(my_llm_tokens()):
            if ev.type == "token":
                yield f"event: token\ndata: {json.dumps({'text': ev.text})}\n\n"
            elif ev.type == "claim":
                yield "event: claim\ndata: " + json.dumps({
                    "text": ev.claim.text, "verdict": ev.claim.verdict,
                }) + "\n\n"
            elif ev.type == "done":
                yield "event: done\ndata: " + json.dumps({
                    "chars_sent": ev.result.chars_sent,
                    "byte_equality": ev.result.byte_equality,
                }) + "\n\n"

    return Response(generate(), mimetype="text/event-stream")

FastAPI / async

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from factivelabs import AsyncFactiveClient

@app.get("/stream")
async def stream():
    async def gen():
        async for ev in client.verify_stream_events(my_async_tokens()):
            if ev.type == "token":
                yield f"data: {ev.text}\n\n"
            elif ev.type == "claim":
                yield f"event: claim\ndata: {ev.claim.verdict}: {ev.claim.text}\n\n"
            elif ev.type == "done":
                yield f"event: done\ndata: byte_equality={ev.result.byte_equality}\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

A complete worked example (Flask app + mock API + customer journey) lives in examples/demo_app/.

Lower-level callback API

verify_stream(tokens, on_token=..., on_claim=...) is still available — use it when you just want the final result and don't need an iterator. Notes on its callback threading model are in the docstring.

Errors

from factivelabs import (
    FactiveClient,
    AuthenticationError,    # 401 — bad / missing key
    QuotaExceededError,     # 402 — out of credits
    RateLimitError,         # 429 — slow down
    ParagraphRejectedError, # 4xx — paragraph dropped, will not be retried
    StreamError,            # server-side error during a stream
    StreamTimeoutError,     # session didn't complete in time
)

try:
    result = client.verify_stream(my_stream(), context="...")
except QuotaExceededError as e:
    print("Out of credits:", e.detail.get("credits_used"), "of", e.detail.get("credits_quota"))
except RateLimitError as e:
    print("Rate limited; retry after", e.retry_after, "seconds")

Configuring the client

client = FactiveClient(
    api_key="fk_...",
    base_url="https://api.factivelabs.com",  # override for staging/local
    timeout=300.0,                            # seconds
)

Examples

The examples/ folder contains drop-in scripts for the three most common LLM providers:

• examples/anthropic_stream.py — Claude streaming
• examples/openai_stream.py — GPT-4 / GPT-5 streaming
• examples/perplexity_stream.py — Perplexity Sonar streaming

What's not in v0

• Section filtering for very large docs — for documents over ~1000 words that need section-by-section processing. Use the REST API directly (/api/v1/analyze-structure plus include_sections/exclude_sections on /verify).
• Claim extraction without verification — for previewing claims before paying for verification. Use the REST API directly (POST /api/v1/extract).

Getting help

• API docs
• Open an issue
• [e-mail]

License

MIT