Knowledge² Python SDK

Official Python client for the Knowledge² retrieval platform. The supported customer journey is:

create corpus -> ingest documents -> build indexes -> search -> optimize retrieval

Installation

From PyPI:

pip install knowledge2
pip install "knowledge2[config]"
pip install "knowledge2[yaml]"

From source:

pip install -e .
pip install -e ".[config]"
pip install -e ".[yaml]"

pip install knowledge2 now includes the typed response model dependency (pydantic) out of the box. Install knowledge2[config] only if you want K2Config environment/file loading via pydantic-settings.

Before You Start

• Use a normal org-scoped API key for the standard retrieval workflow: projects, corpora, documents, indexes, search, and optimize.
• optimize_indexes() and some enterprise/preview surfaces can return feature-flag or quota errors (403, 409, 429) even when the payload is correct. Check environment entitlements early.

Surface Categories

The main docs and examples below focus on the core retrieval workflow.

Quick Start

from sdk import Knowledge2

client = Knowledge2(api_key="k2_...")

project = client.create_project("My Project")
corpus = client.create_corpus(project["id"], "My Corpus")

batch = client.upload_documents_batch_and_wait(
    corpus["id"],
    [
        {
            "source_uri": "doc://overview",
            "raw_text": "Knowledge² builds dense and sparse indexes for hybrid retrieval.",
            "metadata": {"topic": "overview"},
        },
        {
            "source_uri": "doc://search",
            "raw_text": "Hybrid retrieval combines semantic similarity with exact keyword matching.",
            "metadata": {"topic": "search"},
        },
    ],
    auto_index=False,
)
client.sync_indexes(corpus["id"], wait=True)

results = client.search(
    corpus["id"],
    "what is hybrid retrieval",
    top_k=3,
    return_config={"include_text": True, "include_scores": True},
)

for hit in results["results"]:
    print(hit["score"], hit.get("text", "")[:80])

upload_documents_batch_and_wait(...) is the canonical onboarding helper for raw-text JSON batch ingestion. It blocks until the batch finishes and returns the final batch payload, including doc_ids.

If you intentionally want enqueue-first control, use wait=False and then resolve the batch with wait_for_document_batch(...):

docs = [
    {
        "source_uri": "doc://overview",
        "raw_text": "Knowledge² builds dense and sparse indexes for hybrid retrieval.",
    },
]

enqueue = client.upload_documents_batch(corpus["id"], docs, wait=False)
batch = client.wait_for_document_batch(corpus["id"], enqueue["batch_id"])
print(batch["status"], batch["doc_ids"])

For large in-flight imports, get_document_batch(...) and wait_for_document_batch(...) are the canonical batch APIs. Once the batch is visible they return stable doc_ids, terminal resolution, and live batch counters that track admitted documents as processing advances. For broader operational context during a large import, you can still pair them with get_corpus_status(...), get_job(...), or document-level status checks.

Indexing semantics

Uploads run in two stages and get_corpus_status(...) exposes both:

1. Ingestion — chunking + embedding writes. Tracked by ingesting. Once ingesting=false, all chunks for the upload are persisted.
2. Indexing — the dense + sparse retrieval indexes are rebuilt. Tracked by indexing and search_status. Search and chat reflect the new chunks once search_status="ready" (or retrieval_ready=true).

When auto_index=True (the default), uploads automatically queue an incremental index build after ingestion completes. Search keeps serving the previous ready index until the new one finishes — so a doc you just uploaded will not appear in search/chat between ingesting=false and search_status="ready". Poll get_corpus_status for search_status="ready" (or retrieval_ready=true) before relying on the new chunks:

import time

deadline = time.monotonic() + 300  # cap the wait — adjust to your SLA
while time.monotonic() < deadline:
    status = client.get_corpus_status(corpus["id"])
    if status.get("retrieval_ready") or status.get("search_status") == "ready":
        break
    if status.get("search_status") in ("failed", "error"):
        raise RuntimeError(f"Index build failed: {status}")
    time.sleep(3)
else:
    raise TimeoutError("Corpus did not reach search-ready before deadline")

If you set auto_index=False, call client.sync_indexes(corpus["id"], wait=True) yourself when you are ready to publish the new chunks to search.

Improve Retrieval Quality

profile = client.get_query_profile(corpus["id"])
print(profile["example_queries"])

job = client.optimize_indexes(
    corpus["id"],
    example_queries=[
        "how does hybrid retrieval work",
        "what is bm25 tuning",
        "how does rrf combine dense and sparse search",
    ],
    query_count=25,
    top_k=10,
    metric="ndcg",
    wait=True,
)
print(job["job_id"], job["job_type"])

Examples

• sdk/examples/retrieval_quickstart.py: minimal happy path from empty corpus to working hybrid search
• sdk/examples/e2e_lifecycle.py: full retrieval-quality workflow with query profile inspection and indexes:optimize

Run either example with:

export K2_BASE_URL=https://api.knowledge2.ai
export K2_API_KEY=<api-key>
python sdk/examples/retrieval_quickstart.py
python sdk/examples/e2e_lifecycle.py

Authentication

client = Knowledge2(api_key="k2_...")
client = Knowledge2.from_env()
client = Knowledge2(bearer_token="...")

Configuration

Important constructor knobs:

• api_host: defaults to https://api.knowledge2.ai
• api_key: API key for programmatic access
• org_id: auto-detected from GET /v1/auth/whoami when omitted
• timeout: float or ClientTimeouts
• limits: connection-pool settings via ClientLimits
• max_retries: transient retry budget
• validate_responses: enable Pydantic response validation
• http_client: bring your own httpx.Client

from sdk import ClientTimeouts, Knowledge2

client = Knowledge2(
    api_key="k2_...",
    timeout=ClientTimeouts(connect=5, read=120, write=30, pool=10),
)

Namespaces

The flat client API is canonical. The sync client also exposes namespace helpers that group the same methods without changing behavior:

• client.documents.*
• client.documents.upload_batch_and_wait(...)
• client.documents.wait_for_batch(...)
• client.corpora.*
• client.search_ns.*
• client.jobs.*
• client.auth.*

AsyncKnowledge2 currently stays flat-only.

Framework Integrations

The SDK ships LangChain and LlamaIndex integration modules in-package. Install the framework dependency separately, then import the adapter:

from sdk.integrations.langchain import K2LangChainRetriever
from sdk.integrations.llamaindex import K2LlamaIndexRetriever

Enterprise Capabilities

Agents, feeds, pipelines, and A2A are available for enterprise deployments. Keep the primary examples focused on the core retrieval flow.

Agent Declared Schema and Harvest Policy (Preview)

create_agent and update_agent accept a typed declared_schema envelope and a harvest_policy literal. The server requires declared_schema to wrap the JSON Schema under a fields_schema key — bare schemas are rejected with HTTP 422.

agent = client.create_agent(
    name="Case Brief Extractor",
    corpus_id=corpus["id"],
    task_type="extract",
    declared_schema={
        "fields_schema": {
            "type": "object",
            "properties": {
                "court": {"type": "string"},
                "decision_date": {"type": "string", "format": "date"},
            },
            "required": ["court"],
        }
    },
    harvest_policy="declared-only",
)

harvest_policy is one of "off", "inferred", or "declared-only":

• "off" (default) — agent runs do not harvest any output back into the corpus.
• "inferred" — harvest the LLM-inferred fields (and, if declared_schema is set, the declared fields too) into the corpus on each run.
• "declared-only" — harvest only the declared fields from declared_schema (no inferred fields). Requires declared_schema to be set; if it's not, no structured output is requested and the agent behaves as if harvest_policy="off".

The legacy aliases "always" (→ "declared-only") and "declared_and_inferred" (→ "inferred") were normalised by Alembic 0100_harvest_policy_dedupe and are no longer accepted as of #2286.

Running an Agent (Preview)

run_agent(...) enqueues a background job and by default returns the 202 handle immediately. Pass wait=True to block until the run reaches a terminal state and receive the full job dict (including result):

# Fire-and-forget (default) — returns {"job_id": ..., "status": "queued"}.
run = client.run_agent(agent_id, input_chunks=[{"text": "..."}])

# Synchronous — blocks until the run reaches a terminal state.
job = client.run_agent(
    agent_id,
    input_chunks=[{"text": "..."}],
    wait=True,
    poll_s=2,
    timeout_s=120,
)
envelope = job.get("result")  # AgentRunEnvelope | None: content / fields / metadata

timeout_s bounds the full wall-clock budget from enqueue through polling (a slow POST eats into the polling budget; once timeout_s elapses the call raises TimeoutError regardless of which phase is in flight). wait=True raises RuntimeError (carrying the job's error_message when present, otherwise a synthesized "Job <id> ended with status=<status>" string) if the run ends failed or canceled — same exception contract as upload_documents_batch(..., wait=True) and optimize_indexes(..., wait=True). If the server returns a 202 without a job_id (contract violation), wait=True raises RuntimeError rather than silently returning the enqueue handle.

Under client.with_raw_response.run_agent(..., wait=True) the return value is a RawResponse wrapping the terminal job (status_code, headers, and parsed dict) — the trailing GET /v1/jobs/{id} honors raw mode so callers who opt into raw responses keep getting envelopes on the wait path.

Corpus Extraction Template (Preview)

API path convention: All v1 endpoints use hyphens in the path. Use /v1/extraction-templates, not /v1/extraction_templates. The underscore variant returns 404.

create_corpus and update_corpus accept extraction_template_id and re_extraction_policy ("lazy" or "eager"). The template must be a seed template or owned by the target corpus/project's organization — for ordinary org-scoped keys this is the caller's org, but global-scope/admin keys can bind templates owned by the cross-org project they're targeting.

Typed-contract note for declared_schema: AgentDeclaredSchema is a Python TypedDict with total=False. The server requires at least one of fields_schema / metadata_schema (an empty envelope {} is rejected with 422), but TypedDict cannot express that constraint statically — so static type checkers will accept declared_schema={} even though it will fail at runtime. The TS SDK models the same constraint as a discriminated union, which does catch {} at compile time. If you need stricter Python validation, build the request through the Pydantic model layer (sdk.models.agents.AgentDeclaredSchemaModel) before passing it through.

corpus = client.create_corpus(
    project["id"],
    "Legal Decisions",
    extraction_template_id="tpl_legal_v1",
    re_extraction_policy="eager",
)

re_extraction_policy has no effect at create time (a fresh corpus has no documents) but is persisted so that it governs re-extraction behavior when the template binding is later changed on a non-empty corpus.

To unbind a previously-bound template, pass extraction_template_id=None to update_corpus(...) — the SDK distinguishes "omit" (leave unchanged) from "explicitly null" (clear the binding) via a sentinel default. The same pattern applies to update_agent(declared_schema=None, ...) and update_agent(harvest_policy=None, ...).

Per-corpus intent-routing kill-switch

create_corpus and update_corpus accept intent_routing_enabled (bool), echoed back on get_corpus. It defaults to True (routing on — opt-out). Set it to False to disable the whole intent pre-pass + tiered-filter pipeline for the corpus, so chat/agent-run against it skips query classification and tier assignment entirely — useful for incident response and eval baselines. Omit the kwarg (leave None) on update_corpus to leave the current value unchanged. This is broader than a per-field hard-threshold disable, which only turns off the hard tier.

client.update_corpus(corpus["id"], intent_routing_enabled=False)  # routing off

Subscription Modes (Preview)

Agent-feed subscriptions support four authoring modes on create_subscription, gated behind the knowledge_agents_enabled feature flag:

The create response echoes the compiled match_spec and the raw match_spec_description, so no separate /preview endpoint is required:

# Natural-language mode
sub = client.create_subscription(
    agent_id,
    feed_id=feed_id,
    role="input",
    mode="nl_semantic",
    match_spec_description="documents about security incidents",
)
print(sub["match_spec"])            # compiled semantic_like predicate
print(sub["match_spec_description"])  # raw NL description (echoed)

# Structured filter mode
sub = client.create_subscription(
    agent_id,
    feed_id=feed_id,
    role="output",
    mode="structured",
    structured_filters=[
        {"path": "metadata.declared.claim_type", "op": "==", "value": "theft"},
    ],
    structured_logic="and",
)
print(sub["match_spec"])  # compiled explicit predicate

Feed Drafts, Subscriptions, and Feedback (Preview)

In addition to CRUD and run_feed, the Knowledge2 client exposes the full editing and feedback surface of the Feeds API as flat methods on client (the same mixin-based pattern used by every other resource).

draft = client.create_feed_draft(feed_id)
client.update_feed(draft["id"], name="new name")
client.activate_feed_draft(feed_id)  # applies the draft; returns the parent

# Cron-scheduled feed (UTC; mutually exclusive with schedule_interval) — #2305
client.create_feed(
    project_id=project_id,
    source_agent_id=agent_id,
    name="weekly digest",
    persistent=True,
    target_corpus={"existing": corpus_id},
    schedule_cron="0 9 * * MON",  # Mondays at 09:00 UTC
)

# execution_mode='answer' returns an LLM-generated answer over the chunks (#2303)
run = client.run_feed(feed_id, return_results=True)
if run.get("generated_answer"):
    answer_text = run["generated_answer"]["content"]  # envelope shape

# Re-backfill in place (#2328); the response carries the previous cursor for
# forensic correlation in audit logs.
client.backfill_feed(feed_id, start_from="2026-01-01T00:00:00Z")

# `results` is only populated for non-persistent feeds run with
# `return_results=True`; guard the example for safe use.
if run.get("results"):
    client.submit_feed_feedback(
        feed_id,
        rating=1,
        chunk_id=run["results"][0]["chunk_id"],
        feed_run_id=run["feed_run_id"],
    )
stats = client.get_feed_feedback_stats(feed_id)  # org-wide for this feed

All methods above are fully mirrored on AsyncKnowledge2 under the same names.

Error Handling

All SDK exceptions inherit from Knowledge2Error.

from sdk.errors import Knowledge2Error, NotFoundError, RateLimitError

try:
    client.get_corpus("missing")
except NotFoundError:
    ...
except RateLimitError as exc:
    print(exc.retry_after)
except Knowledge2Error as exc:
    print(exc)