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, id_gt: Optional[str] = None) -> List[DocumentMetadataResponse]
• client.documents.update(collection_name: str, path: str, metadata: Optional[dict]) -> UpdateDocumentResponse
• client.documents.delete(collection_name: str, path: str) -> None
### 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(collection_name: Optional[str] = None) -> StatusResponse
### Parsers
• client.parsers.parse_document(base64_data: str) -> ParseDocumentResponse
### Models
• client.models.rerank(query: str, documents: str[], model:str, top_n:int) -> RerankResult
Common Patterns:
1 Collections
try:
await zclient.collections.add(collection_name="my_col")
except ConflictError:
pass
names = (await zclient.collections.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.list_info(
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(
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_col = await zclient.status.get(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-1-small",
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-1-small",
)
### Expected Response Models
All responses return structured BaseModel objects as follows:
1. DocumentResponse
Used in get_info()
python
class DocumentResponse(BaseModel):
id: str # UUID of the document
collection_name: str
path: str
metadata: Dict[str, str] # Metadata key-value pairs
index_status: str # Enum: "parsing_failed", "not_parsed", "parsing", "not_indexed", "indexing", "indexed"
num_pages: Optional[int] = None # Can be null
content: Optional[str] = None # Null unless `include_content=True`
2. UpdateDocumentResponse
Used in update()
python
class UpdateDocumentResponse(BaseModel):
previous_id: str # Old document UUID
new_id: str # New updated document UUID
3. DocumentRetrievalResponse
Used in top_documents()
python
class DocumentRetrievalResponse(BaseModel):
results: List[Response]
class Response(BaseModel):
path: str
metadata: Optional[Dict[str, 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. StatusResponse
Used in status.get()
python
class StatusResponse(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
7. ParseDocumentResponse
Used in parse_document()
python
class ParseDocumentResponse(BaseModel):
pages: List[str] # List of extracted page contents
8. RerankResponse
Used in rerank()
python
class RerankResponse(BaseModel):
results: RerankResult
class RerankResult(BaseModel):
index: int
relevance_score: number
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("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("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())
