Knowhere Python SDK

Installation

pip install knowhere-python-sdk

uv add knowhere-python-sdk

Usage

import knowhere

client = knowhere.Knowhere(api_key="sk_...")

result = client.parse(
    url="https://example.com/report.pdf",
)

print(result.statistics.total_chunks)
print(result.full_markdown[:200])

for chunk in result.text_chunks:
    print(chunk.content[:80])

for page in result.page_chunks:
    print(page.content_source)       # "summary"
    print(page.content[:120])        # page-level summary
    print(page.metadata.page_nums)   # [4, 5, 6]

Retrieval and document lifecycle

job = client.jobs.create(
    source_type="url",
    source_url="https://example.com/manual.pdf",
    namespace="support-center",
    document_metadata={"title": "Support manual"},
)

document_id = job.document_id
job_result = client.jobs.wait(job.job_id)
document_id = document_id or job_result.document_id

if document_id is None:
    raise RuntimeError("Expected document_id after successful publication.")

response = client.retrieval.query(
    namespace="support-center",
    query="How do I reset Bluetooth pairing?",
    chunk_types=["page"],
    top_k=5,
    channels=["path", "term"],
    filter_mode="keep",
    signal_paths=["Bluetooth", "Pairing"],
)

print(response.router_used)
print(response.answer_text)
print(response.evidence)  # grouped [E n] parts to consume
print(response.evidence_text)  # deprecated; always empty. Use evidence.
print(response.stop_reason)
print(response.failure_reason)

for reference in response.referenced_chunks:
    print(reference.chunk_id, reference.chunk_type, reference.content_source)
    print(reference.metadata, reference.asset_url)

for result in response.results:  # raw path chunks for debug
    print(result.chunk_id, result.chunk_type, result.content_source)
    print(result.content)
    print(result.score)
    print(result.source.source_file_name, result.source.section_path)

update_job = client.jobs.create(
    source_type="url",
    source_url="https://example.com/manual-v2.pdf",
    document_id=document_id,
)

document = client.documents.get(document_id)
print(document.status)

chunks = client.documents.list_chunks(
    document_id,
    page=1,
    page_size=50,
    chunk_type="page",
    include_asset_urls=True,
)
print(chunks.pagination.total)
if chunks.chunks:
    chunk = client.documents.get_chunk(
        document_id,
        chunks.chunks[0].id,
        include_asset_urls=True,
    )
    print(chunk.chunk.content)
    print(chunk.chunk.metadata.get("page_nums"))  # Page citations.
    print(chunk.chunk.asset_url)  # Requested 7-day URL when available.
    page_assets = chunk.chunk.metadata.get("pageAssets") or []
    print(page_assets)

source = client.documents.get_page_citation_source(document_id)
print(source.url)

client.documents.archive(document_id)

documents = client.documents.list(
    namespace="support-center",
    page=1,
    page_size=50,
)
for document in documents.documents:
    print(document.document_id, document.status)
print(documents.pagination.total_pages)

response = client.retrieval.query(
    namespace="support-center",
    query="battery charging",
    include_document_ids=["doc_123", "doc_old"],
    exclude_document_ids=["doc_old"],
    exclude_sections=[
        {"document_id": "doc_123", "section_path": "Appendix / Legal"}
    ],
)

client = knowhere.Knowhere(auth_token_provider=lambda: current_access_token())

Parse a local file

from pathlib import Path

result = client.parse(
    file=Path("report.pdf"),
    parsing_params={"model": "advanced", "ocr_enabled": True},
)

print(result.manifest.source_file_name)  # "report.pdf"
print(len(result.chunks))                # 152
print(result.namespace)                  # "default" or your explicit namespace
print(result.document_id)                # Published canonical document id

Bring your own LLM keys (BYOK)

# Multimodal shorthand — one model for text + vision
llm_config = {
    "api_key": "sk-...",
    "model": "gpt-4o",
    "base_url": "https://api.openai.com/v1",
}

# Same endpoint, different models per channel
llm_config = {
    "api_key": "sk-...",
    "base_url": "https://api.openai.com/v1",
    "models": {"text": "gpt-4o-mini", "vision": "gpt-4o"},
}

# Or two different endpoints
llm_config = {
    "text": {
        "api_key": "sk-...",
        "model": "gpt-4o-mini",
        "base_url": "https://api.openai.com/v1",
    },
    "vision": {
        "api_key": "sk-ali-...",
        "model": "qwen-vl-max",
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    },
}

result = client.parse(file=Path("report.pdf"), llm_config=llm_config)

response = client.retrieval.query(
    namespace="support-center",
    query="refund policy",
    use_agentic=True,
    llm_config=llm_config,
)

Access different chunk types

result = client.parse(url="https://example.com/report.pdf")

# Text chunks
for chunk in result.text_chunks:
    print(chunk.metadata.keywords)
    print(chunk.metadata.summary)

# Page chunks (v2 page-memory results)
for chunk in result.page_chunks:
    print(chunk.content_source)      # "summary"
    print(chunk.content[:120])
    print(chunk.metadata.page_nums)  # citation pages
    print(chunk.metadata.entities)

# Image chunks (raw bytes loaded from ZIP)
for chunk in result.image_chunks:
    print(chunk.file_path)
    print(len(chunk.data))       # bytes
    chunk.save("./output/")      # writes image to disk

# Table chunks (HTML loaded from ZIP)
for chunk in result.table_chunks:
    print(chunk.file_path)
    print(chunk.html[:100])

Save all results to disk

result = client.parse(file=Path("report.pdf"))
result.save("./output/report/")

Async usage

import asyncio
import knowhere

async def main():
    async with knowhere.AsyncKnowhere(api_key="sk_...") as client:
        result = await client.parse(url="https://example.com/report.pdf")
        print(result.statistics.total_chunks)

        for chunk in result.text_chunks:
            print(chunk.summary)

asyncio.run(main())

Step-by-step control

from pathlib import Path

# Step 1: Create a parsing job
job = client.jobs.create(
    source_type="file",
    file_name="report.pdf",
    namespace="support-center",
    parsing_params={"model": "advanced", "ocr_enabled": True},
)

# Step 2: Upload file to presigned URL
client.jobs.upload(job, file=Path("report.pdf"))

# Step 3: Poll until done (adaptive backoff)
job_result = client.jobs.wait(job.job_id, poll_interval=10.0, poll_timeout=1800.0)

print(job_result.document_id)  # Persist this to update/archive the document later.

# Step 4: Download and parse results
result = client.jobs.load(job_result)
print(result.statistics)

Handling errors

import knowhere

try:
    result = client.parse(url="https://example.com/report.pdf")
except knowhere.AuthenticationError:
    print("Invalid API key")
except knowhere.APIStatusError as e:
    print(f"{e.status_code}: {e.message}")

Configuration

# Uses environment variables automatically
client = knowhere.Knowhere()

# Or configure explicitly
client = knowhere.Knowhere(
    api_key="sk_...",
    base_url="https://api.knowhereto.ai",
    timeout=30.0,           # HTTP request timeout (default: 60s)
    upload_timeout=300.0,   # File upload timeout (default: 600s)
    max_retries=3,          # Max retry attempts (default: 5)
)

Retries

client = knowhere.Knowhere(
    api_key="sk_...",
    max_retries=3,  # default is 5
)

Determining the installed version

import knowhere
print(knowhere.__version__)

Versioning

Requirements

Community

License