Metadata-Version: 2.4
Name: factivelabs
Version: 0.3.0
Summary: Python SDK for the Factive fact-checking API. Stream LLM output and get verified claims back in real time.
Author-email: Factive Labs <support@factivelabs.com>
License: MIT
Project-URL: Homepage, https://www.factivelabs.com
Project-URL: Documentation, https://api.factivelabs.com/docs
Project-URL: Source, https://github.com/jonasphilliplee/claimcheck
Project-URL: Issues, https://github.com/jonasphilliplee/claimcheck/issues
Keywords: fact-check,fact-checking,hallucination,llm,claim-verification,ai,verification,factivelabs
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx<1.0,>=0.25
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21; extra == "dev"
Dynamic: license-file

# factivelabs

> **What changed in 0.3.0** — the streaming flow now uses a single stateless endpoint (`POST /api/v1/verify/paragraph`) instead of the older session-based SSE protocol. The Python API of `verify_stream` / `verify_stream_events` is unchanged — your code keeps working. Under the hood, each paragraph is now an independent HTTP request, so the API can scale horizontally with no shared state.

Python SDK for the [Factive Labs](https://www.factivelabs.com) fact-checking API. Stream an LLM's output through the verifier as it generates and get back fact-checked claims in real time.

## Install

```bash
pip install factivelabs
```

## Quickstart

Get an API key at [factivelabs.com/dashboard](https://www.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:

```python
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

```python
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

```python
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

```python
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

```python
# 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)

```python
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.

```python
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:

```python
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.

```python
# 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`:

```python
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:

```python
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](https://api.factivelabs.com/docs).

## Callbacks

```python
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.

```python
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)

```python
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

```python
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

```python
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

```python
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](https://api.factivelabs.com/docs)
- [Open an issue](https://github.com/jonasphilliplee/claimcheck/issues)
- support@factivelabs.com


## Changelog

### 0.3.0
- **Stateless streaming.** `verify_stream` / `verify_stream_events` now post each paragraph to a single stateless endpoint (`/api/v1/verify/paragraph`) instead of opening a long-lived SSE session. Public Python API unchanged. Removes the worker-affinity bug class entirely; lets the API scale workers freely.

### 0.2.0
- **Async client** (`AsyncFactiveClient`) with the full sync surface mirrored.
- **`verify_url`** + **`verify_file`** + **`extract_text`** for URL/file inputs.
- **Batch jobs** API (`submit_batch`, `get_job`, `wait_for_batch`).
- **Private-corpus management** (`client.corpus.*`).
- **Streaming events iterator** (`verify_stream_events`).
- **Context-manager support** (`with FactiveClient(...) as c:`) for connection reuse.

### 0.1.0
- Initial release: `verify_text` and callback-style `verify_stream`.

## License

MIT
