Polyvia Python SDK

pip install polyvia

pip install "polyvia[langchain]"

from polyvia import Polyvia

client = Polyvia(api_key="poly_<key>")

# Ingest a batch into a group, then ask one question across all of it.
items = client.ingest.batch(
    ["q1.pdf", "q2.pdf", "q3.pdf", "q4.pdf"],
    group="FY24 Earnings",
)
for item in items:
    client.ingest.wait(item.task_id)

# Answers cite the exact page in each document.
print(client.query("How did revenue trend across the four quarters?",
                   group="FY24 Earnings").answer)

# Single file — accepts a file path, bytes, or file-like object.
# `group` takes the group name; it's created if it doesn't exist yet.
result = client.ingest.file("report.pdf", group="Finance")
# → IngestResult(document_id="<id>", task_id="<id>", status="pending")

# Multiple files (the group is resolved once for the batch)
items = client.ingest.batch(
    ["q3.pdf", "q4.pdf"],
    names=["Q3 Report", "Q4 Report"],
    group="Finance",
)

# Check status
status = client.ingest.status(result.task_id)
# → IngestionStatus(task_id="<id>", document_id="<id>", status="completed")

# Block until done — raises IngestionError on failure, IngestionTimeout on timeout
client.ingest.wait(result.task_id, poll_interval=5, timeout=300)

# All completed documents
answer = client.query("What risks are mentioned across all reports?")

# Single document (fastest)
answer = client.query("Summarise section 3.", document_id="doc_<id>")

# Scoped to a group — by name (must already exist)
answer = client.query("Key findings?", group="Finance")

# Multiple groups — by id
answer = client.query("Compare results.", group_ids=["g_<id>", "g_<id>"])

print(answer.answer)

# Idempotent: returns the existing "Finance" group or creates it (matched by
# exact name, so never a duplicate). Returns a Group.
group = client.groups.get_or_create("Finance")
group.id      # backend id, if you ever need it
group.name    # "Finance"

# Look one up without creating it (None if there isn't one)
existing = client.groups.find("Finance")

# List
groups = client.groups.list()

# Delete all documents in a group, then the group itself
client.groups.delete(group.id, delete_documents=True)

# Or separately
client.groups.delete_documents(group.id)
client.groups.delete(group.id)

# List — filter by status and/or group
docs = client.documents.list(status="completed", group_id="g_<id>")
docs = client.documents.list(group_ids=["g_<id>", "g_<id>"])

# Get one
doc = client.documents.get("doc_<id>")

# Move to a different group / remove from group
client.documents.update("doc_<id>", group_id="g_other")
client.documents.update("doc_<id>", group_id=None)

# Delete
client.documents.delete("doc_<id>")

usage = client.usage()
print(usage.usage.requests.period)     # requests this calendar month
print(usage.usage.documents_stored)    # live document count

limits = client.rate_limits()
print(limits.limits["requests_per_minute"])
print(limits.current["remaining_this_minute"])
print(limits.resets_at.month)          # ISO timestamp of next monthly reset

import asyncio
from polyvia import AsyncPolyvia

async def main():
    async with AsyncPolyvia(api_key="poly_<key>") as client:
        result = await client.ingest.file("report.pdf")
        await client.ingest.wait(result.task_id)
        answer = await client.query("Key findings?")
        print(answer.answer)

asyncio.run(main())

from polyvia import (
    AuthenticationError,  # 401 — bad or missing API key
    ForbiddenError,        # 403 — document belongs to another user
    NotFoundError,         # 404 — document, group, or task not found
    RateLimitError,        # 429 — too many requests
    IngestionError,        # task finished with status='failed'
    IngestionTimeout,      # ingest.wait() exceeded its timeout
)

try:
    client.ingest.wait(task_id, timeout=60)
except IngestionError as e:
    print(f"Parsing failed: {e.error}")
except IngestionTimeout:
    print("Timed out — document may still be processing")
except RateLimitError:
    print("Rate limit hit — back off and retry")
except NotFoundError:
    print("Document or task not found")
except AuthenticationError:
    print("Invalid API key")