Quickstart

ZeroEntropy SDK Helper

### Description:
ZeroEntropy is a state-of-the-art retrieval API for documents, pages, snippets and reranking.  
It provides low-latency, high-accuracy search over your private corpus via a simple Python SDK.

ZeroEntropy can be installed using:
	•	Python: pip install zeroentropy
	•	Node.js: npm install zeroentropy

### Client Usage

from zeroentropy import ZeroEntropy
client = ZeroEntropy(api_key="your_api_key")

Auth & Configuration:
• ENV VARS read by SDK:
    ZEROENTROPY_API_KEY 
Missing key triggers authentication error on instantiation.

Instantiate:
from dotenv import load_dotenv  
load_dotenv()  
from zeroentropy import AsyncZeroEntropy, ConflictError, HTTPStatusError  
zclient = AsyncZeroEntropy()      # picks up ENV VARS

### SDK Structure:
All methods are async, grouped under:
    zclient.collections  
    zclient.documents  
    zclient.status  
    zclient.queries  
    zclient.models 

Each method returns structured responses defined by pydantic.BaseModel.

### Collections
	•	client.collections.add(collection_name: str) -> None
        Always specify a collection name using client.collections.add(collection_name="my_collection")
        If the collection already exists, it will be throw an error, so you need to check if the collection exists first.
	•	client.collections.get_list() -> List[str]
	•	client.collections.delete(collection_name: str) -> None

### Documents
	•	client.documents.add(collection_name: str, path: str, content, metadata: dict = None, overwrite: bool = False) -> None
        The add method already handles parsing for PDFs etc. The content dict can take the following formats:
        content={"type":"auto", "base64_data":"my_document.pdf"} for a PDF, content={"type":"text", "text":"my_document.pdf"} for a text file, and content={"type":"text-pages", "pages":[ "page 1 content", "page 2 content"]} for pages of text.
        If the document already exists, it will be throw an error, so you need to check if the document exists first.
	•	client.documents.get_info(collection_name: str, path: str, include_content: bool = False) -> DocumentResponse
	•	client.documents.get_info_list(collection_name: str, limit: int = 1024, path_gt: Optional[str] = None) -> List[DocumentGetInfoListResponse]
	•	client.documents.update(collection_name: str, path: str, metadata: Optional[dict]) -> UpdateDocumentResponse
	•	client.documents.delete(collection_name: str, path: Union[str, List[str]]) -> DocumentDeleteResponse

### Queries
	•	client.queries.top_documents(collection_name: str, query: str, k: int, filter: Optional[dict] = None, include_metadata: bool = False, latency_mode: str = "low") -> List[DocumentRetrievalResponse]
	•	client.queries.top_pages(collection_name: str, query: str, k: int, filter: Optional[dict] = None, include_content: bool = False, latency_mode: str = "low") -> List[PageRetrievalResponse]
	•	client.queries.top_snippets(collection_name: str, query: str, k: int, filter: Optional[dict] = None, precise_responses: bool = False) -> List[SnippetResponse]

### Status
	•	client.status.get_status(collection_name: Optional[str] = None) -> StatusGetStatusResponse

### Models
	•	client.models.embed(input: Union[str, List[str]], input_type: "query" | "document", model: str, dimensions: Optional[int] = None, encoding_format: "float" | "base64" = "float", latency: Optional["fast" | "slow"] = None) -> ModelEmbedResponse
	•	client.models.rerank(documents: List[str], model: str, query: str, top_n: Optional[int] = None) -> ModelRerankResponse

Common Patterns:

1  Collections
try:
    await zclient.collections.add(collection_name="my_col")
except ConflictError:
    pass
names = (await zclient.collections.get_list()).collection_names
await zclient.collections.delete(collection_name="my_col")

2  Documents

# Add text
await zclient.documents.add(
    collection_name="col",
    path="doc.txt",
    content={"type":"text","text":text},
    metadata={"source":"notes"},
)

# Add PDF via OCR
b64 = base64.b64encode(open(path,"rb").read()).decode()
await zclient.documents.add(
    collection_name="col",
    path="doc.pdf",
    content={"type":"auto","base64_data":b64},
    metadata={"type":"pdf"},
)

# Add CSV lines
for i,line in enumerate(open(path).read().splitlines()):
    await zclient.documents.add(
        collection_name="col",
        path=f"{path}_{i}",
        content={"type":"text","text":line},
        metadata={"type":"csv"},
    )

# Delete
await zclient.documents.delete(collection_name="col", path="doc.txt")

# Get info (with optional content)
info = await zclient.documents.get_info(
    collection_name="col",
    path="doc.txt",
    include_content=True
)

3  Update & Pagination

# Update metadata or force re-index
await zclient.documents.update(
    collection_name="col",
    path="doc.txt",
    metadata={"reviewed":"yes"},
)

# List documents with pagination
resp = await zclient.documents.get_info_list(
    collection_name="col",
    limit=100,
    path_gt="doc_009.txt",
)
for doc in resp.documents:
    print(doc.path, doc.index_status)

# Per-page info
page = await zclient.documents.get_page_info(
    collection_name="col",
    path="doc.pdf",
    page_index=2,
    include_content=True,
)

4  Pure Parsing (OCR helper)
pages = await zclient.documents.parse(
    base64_data=b64
)
# returns list of page strings without indexing

5  Status (overall or per-collection)
status_all = await zclient.status.get_status()
status_col = await zclient.status.get_status(collection_name="col")

6  Queries
# Top K documents (k≤2048), with filter, reranker, latency_mode
docs = await zclient.queries.top_documents(
    collection_name="col",
    query="find insight",
    k=5,
    filter={"type":{"$ne":"csv"}},
    include_metadata=True,
    reranker="zerank-2",
    latency_mode="low",
)

# Top K pages (k≤1024), include_content, latency_mode
pages = await zclient.queries.top_pages(
    collection_name="col",
    query="overview",
    k=3,
    include_content=True,
    latency_mode="high",
)

# Top K snippets (k≤128), precise or coarse
snips = await zclient.queries.top_snippets(
    collection_name="col",
    query="key method",
    k=5,
    precise_responses=True,
    include_document_metadata=True,
    reranker="zerank-2",
)
### Expected Response Models

All responses return structured BaseModel objects as follows:

1. DocumentGetInfoResponse

Used in get_info()

python
class DocumentGetInfoResponse(BaseModel):
    document: Document

class Document(BaseModel):
    id: str  # UUID of the document
    collection_name: str
    path: str
    file_url: str  # URL to download raw document
    size: int  # Raw document size in bytes
    metadata: Dict[str, Union[str, List[str]]]  # Metadata key-value pairs
    index_status: str  # Enum: "not_parsed", "parsing", "not_indexed", "indexing", "indexed", "parsing_failed", "indexing_failed"
    num_pages: Optional[int] = None  # Can be null
    content: Optional[str] = None  # Null unless `include_content=True`

2. DocumentUpdateResponse

Used in update()

python
class DocumentUpdateResponse(BaseModel):
    message: Optional[str] = None  # "Success!"

3. DocumentRetrievalResponse

Used in top_documents()
python
class DocumentRetrievalResponse(BaseModel):
    results: List[Response] 
class Response(BaseModel):
    path: str
    metadata: Optional[Dict[str, Union[str, List[str]]]] = None  # Null if `include_metadata=False`
    score: float  # Relevancy score

4. PageRetrievalResponse

Used in top_pages()

python
class PageRetrievalResponse(BaseModel):
    results: List[Response] 
class Response(BaseModel):
    path: str  # Document path
    page_index: int  # 0-indexed page number
    score: float  # Relevancy score
    content: Optional[str] = None  # Null if `include_content=False`

5. SnippetResponse

Used in top_snippets()

python
class SnippetResponse(BaseModel):
    results: List[Response] 
class Response(BaseModel):
    path: str
    start_index: int  # Start character index of snippet
    end_index: int  # End character index of snippet
    page_span: List[int]  # (start_page, end_page) index range
    content: Optional[str] = None  # Snippet text
    score: float  # Relevancy score

6. StatusGetStatusResponse

Used in status.get_status()

python
class StatusGetStatusResponse(BaseModel):
    num_documents: int  # Total document count
    num_parsing_documents: int  # Documents still being parsed
    num_indexing_documents: int  # Documents currently being indexed
    num_indexed_documents: int  # Successfully indexed documents
    num_failed_documents: int  # Documents that failed
    num_indexed_bytes: int  # Total UTF-8 encoded bytes indexed

7. ParseDocumentResponse

Used in parse_document()

python
class ParseDocumentResponse(BaseModel):
    pages: List[str]  # List of extracted page contents

8. ModelEmbedResponse

Used in embed()

python
class ModelEmbedResponse(BaseModel):
    results: List[Result]
    usage: Usage

class Result(BaseModel):
    embedding: Union[List[float], str]  # float array or base64 string

class Usage(BaseModel):
    total_bytes: int  # For ratelimiting
    total_tokens: int  # For billing

9. ModelRerankResponse

Used in rerank()

python
class ModelRerankResponse(BaseModel):
    results: List[Result]
    actual_latency_mode: str  # "fast" or "slow"
    e2e_latency: float  # Total time in seconds
    inference_latency: float
    total_bytes: int
    total_tokens: int

class Result(BaseModel):
    index: int  # Index in original documents array
    relevance_score: float  # 0.0 to 1.0

All query endpoints support JSON filters:

### Additional Notes
- LLM should always use these BaseModels when generating SDK-based responses.
- Metadata Filtering
    - Document metadata is always dict[str, str | list[str]].
	- Filters support operators: $eq, $ne, $gt, $gte, $lt, $lte (for equality and range queries).
- Responses will always match these structures unless otherwise stated.
- Always include imports and load_dotenv().  
- Use proper async/await structure with asyncio.run().  
- Respect parameter limits and JSON‐filter syntax.  
- Override base_url when targeting EU cluster.  
- Wrap calls in try/except to handle ConflictError and HTTPStatusError.

### Complete Async Example:

import asyncio
from dotenv import load_dotenv
from zeroentropy import AsyncZeroEntropy, ConflictError, HTTPStatusError
import base64

load_dotenv()
zclient = AsyncZeroEntropy()

async def main():
    try:
        await zclient.collections.add(collection_name="my_col")
    except ConflictError:
        pass

    text = "Hello ZeroEntropy"
    await zclient.documents.add(
        collection_name="my_col",
        path="hello.txt",
        content={"type":"text","text":text},
        metadata={"lang":"en"},
    )

    status = await zclient.status.get_status(collection_name="my_col")
    print("Indexed:", status.num_indexed_documents)

    docs = await zclient.queries.top_documents(
        collection_name="my_col",
        query="Hello",
        k=1,
        include_metadata=True,
    )
    print(docs.results)

if __name__ == "__main__":
    asyncio.run(main())