index4ai

Official Python SDK for index4ai: index documents into a project namespace and get prompt-ready retrieval context for RAG from hybrid search.

Installation

Requires Python 3.10+.

pip install index4ai

Authentication

index4ai uses a project API key. Create one in the index4ai console (https://console.index4.ai) and set it in your environment:

export INDEX4AI_API_KEY="..."

or pass api_key= to the client. The client checks the key's prefix when it is created and raises InvalidApiKeyError if you pass a key that is not a project key, so a misconfiguration fails locally instead of as a 401/403.

Usage

from index4ai import Index4AI

client = Index4AI()  # reads INDEX4AI_API_KEY

# 0. List the project's typed attributes BEFORE declaring any.
#    Attribute names (and their types) are permanent per project and capped at
#    `max_names`, so reuse an existing name instead of minting a near-duplicate.
known = {a.name: a.type for a in client.list_attributes().attributes}

# 1. Ingest a document. It is indexed into your project namespace; there is no
#    archive to download. Typed attributes enable query-time filtering:
#    string/int/float/bool are inferred; datetime and encrypted_text must be
#    declared in attribute_schema.
result = client.ingest(
    "document.pdf",
    attributes={"source": "treasury", "published_at": "2023-01-01T00:00:00Z"},
    attribute_schema={"published_at": {"type": "datetime"}},
)
print(result.doc_id, result.status, result.usage)

# 1b. Or let the server fetch the file from a public URL — no upload.
#     Optionally POST a webhook when the job finishes; the value is
#     "<url>|<header>:<value>|..." (capped at 1024 chars).
result = client.ingest(
    remote_url="https://example.com/document.pdf",
    completion="https://example.com/hook|x-api-key:secret",
)

# 2. Hybrid search across all documents in the namespace. Returns a
#    prompt-ready XML+Markdown block — drop it straight into an LLM prompt.
ctx = client.search(
    "How do I configure retries?",
    target_tokens=2048,
    attribute_filters=[{"attr": "source", "op": "eq", "value": "treasury"}],
)
print(ctx.context)

# 3. List, inspect, and delete documents. list_docs() follows pagination and
#    returns every document in the namespace.
for doc in client.list_docs():
    print(doc.doc_id, doc.filename, doc.pages)

deleted = client.delete_doc(result.doc_id)
print(deleted.vectors_deleted, deleted.storage_deleted)

The document id

ingest() / collect() return an IngestResult whose doc_id is the id to search and manage the document by. If the same content was already indexed, the server keeps the existing document instead: deduplicated is True and doc_id is the pre-existing document's id. doc_id is None only when the server reported no indexing outcome.

Submitting and waiting separately

ingest() is submit() followed by collect(). To submit many files and collect later, split them:

job_id = client.submit("document.pdf")
# ... later, even from another process:
result = client.collect(job_id)

Waiting can always be resumed with the same job_id. Submitting the file again instead creates a second job. Two errors tell you which case you are in:

• JobWaitError — the job exists and keeps running; only the wait failed (network, unreadable status). Resume with client.collect(err.job_id).
• SubmitUncertainError — the connection failed mid-submit, so the server may or may not have created a job. Check list_docs() before submitting again.

Listing attributes: list before you declare

list_attributes() returns an AttributeList: attributes (a list of AttributeInfo(name, type)), max_names (the project's cap on distinct names) and note (the server's guidance).

Attribute names are permanent per project — the first write of a name fixes its type and spends one of the max_names slots, and clearing a value does not free the name. Call list_attributes() before declaring attributes and reuse an existing name with its existing type.

A 503 means the attribute schema is temporarily unreadable. It is raised as ApiError with status_code == 503 — retry later; never treat it as an empty list, or you will mint duplicate names. A response without an attributes list raises InvalidResponseError for the same reason.

Async

AsyncIndex4AI mirrors the sync client method for method:

import asyncio

from index4ai import AsyncIndex4AI


async def main() -> None:
    async with AsyncIndex4AI() as client:
        ctx = await client.search_in_doc("summarize section 3", "doc_abc123")
        print(ctx.context)


asyncio.run(main())

Migrating from poma.Grill

Configuration

The client targets API version 1 (https://api.index4.ai/v1).