Polyvia Python SDK

pip install polyvia

pip install "polyvia[langchain]"

from polyvia import Polyvia

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

# Ingest → wait → query
result = client.ingest.file("report.pdf", name="Q4 Report")
client.ingest.wait(result.task_id)
print(client.query("What are the key findings?").answer)

export POLYVIA_API_KEY=poly_<key>

client = Polyvia()

# Single file — accepts a file path, bytes, or file-like object
result = client.ingest.file("report.pdf", name="Q4 Report", group_id="g_<id>")
# → IngestResult(document_id="<id>", task_id="<id>", status="pending")

# Multiple files
items = client.ingest.batch(
    ["q3.pdf", "q4.pdf"],
    names=["Q3 Report", "Q4 Report"],
    group_id="g_<id>",
)

# 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
answer = client.query("Key findings?", group_id="g_<id>")

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

print(answer.answer)

# Create
group = client.groups.create("Finance")
group_id = group.group_id

# 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")