Metadata-Version: 2.4
Name: knowledge2
Version: 0.8.0
Summary: Python SDK for the Knowledge² retrieval platform
Author-email: Knowledge2 <contact@knowledge2.ai>
License: MIT
Project-URL: Homepage, https://knowledge2.ai
Project-URL: Documentation, https://knowledge2.ai/docs
Project-URL: Repository, https://github.com/knowledge2-ai/knowledge2-python-sdk
Project-URL: Changelog, https://github.com/knowledge2-ai/knowledge2-python-sdk/blob/main/CHANGELOG.md
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic<3,>=2
Provides-Extra: config
Requires-Dist: pydantic-settings>=2.0; extra == "config"
Provides-Extra: pydantic
Requires-Dist: pydantic<3,>=2; extra == "pydantic"
Provides-Extra: yaml
Requires-Dist: pyyaml>=6.0; extra == "yaml"

# Knowledge² Python SDK

[![PyPI version](https://img.shields.io/pypi/v/knowledge2.svg)](https://pypi.org/project/knowledge2/)
[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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:

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

From source:

```bash
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

| Category | Surface |
|---|---|
| Core retrieval workflow | orgs, auth, projects, corpora, documents, indexes, search, jobs, metadata, onboarding, audit, usage, console, generation models |
| Enterprise capabilities | agents, feeds, destinations, pipelines, A2A, extraction templates |
| Quality & schema evolution (preview) | quality (engagement, metrics, retrieval outcomes, proposals) — proposals require the `schema_evolution_enabled` org flag; metrics and retrieval-outcomes reads are ungated |

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

## Quick Start

```python
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(...)`:

```python
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:

```python
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

```python
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:

```bash
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

| Method | Header | Typical use |
|---|---|---|
| API key | `X-API-Key` | primary programmatic access for retrieval workflows |
| Bearer token | `Authorization: Bearer <token>` | console / Auth0 session |

```python
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`

```python
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:

```python
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.

```python
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`):

```python
# 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.

```python
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.

```python
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:

| Mode | Use | Required fields |
|------|-----|-----------------|
| `always` | Route every envelope from the feed | `feed_id`, `role` |
| `explicit` | Evaluate a predicate DSL against the envelope | `feed_id`, `role`, `match_spec` |
| `nl_semantic` | Describe the match in plain English; compiled server-side into a `semantic_like` predicate against `content` | `feed_id`, `role`, `match_spec_description` (10-500 chars); optional `threshold` (default 0.75) |
| `structured` | Compose filter clauses without writing raw DSL; compiled server-side into an explicit MatchSpec | `feed_id`, `role`, `structured_filters` (list of filter dicts); optional `structured_logic` (`"and"` default or `"or"`) |

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

```python
# 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).

| Method | Endpoint | Notes |
|--------|----------|-------|
| `create_feed_draft(feed_id)` | `POST /v1/feeds/{id}/draft` | Returns a draft feed with `parent_feed_id` set |
| `get_feed_draft(feed_id)` | `GET /v1/feeds/{id}/draft` | 404 when no draft exists |
| `activate_feed_draft(feed_id)` | `POST /v1/feeds/{id}/draft/activate` | Returns the updated **parent** feed (draft is deleted) |
| `discard_feed_draft(feed_id)` | `DELETE /v1/feeds/{id}/draft` | Returns `None` |
| `list_feed_subscriptions(feed_id)` | Read-only view | Returns subscriptions embedded on the feed record; use `create_subscription` on the Agents mixin to attach new ones |
| `submit_feed_feedback(feed_id, *, rating, chunk_id, feed_run_id)` | `POST /v1/feeds/{id}/feedback` | `rating` is `1` (thumbs up) or `0` (thumbs down) |
| `get_feed_feedback_stats(feed_id, *, feed_run_id=None)` | `GET /v1/feeds/{id}/feedback` | Optional `feed_run_id` scopes stats to a single run |
| `backfill_feed(feed_id, *, start_from)` | `POST /v1/feeds/{id}/backfill` | Re-enqueue a backfill on an existing feed without delete-and-recreate. Resets the feed cursor and updates `start_from`. Returns `202` with `{job_id, start_from, previous_last_checked_seq}`; raises `409 feed_backfill_in_flight` if a run_feed job is already queued or running (#2328) |

```python
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`.

```python
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)
```
