knowledge-ingest

Group: Ingestion. Upload + automatic classification via pipeline parse → chunk → embed → enqueue classify.

When to trigger

• "Index this PDF"
• "Add this URL to the knowledge base"
• "Upload these files to Academy"
• User passes a file path with ingestion intent

Arguments

| Name | Type | Required | Description | |---|---|---|---| | file_path | str | one of two | Local path | | url | str | one of two | URL to download first | | connection | str | no | Defaults to first ready | | space | str | yes | Destination space slug | | unit_id | str | no | Associated unit | | title | str | no | Derived from filename if absent | | tags | liststr] | no | User-defined tags |

Workflow

Step 1 — Validate connection + space

If connection not provided, use first ready. If none: error ("Run knowledge-admin action=connect").

Validate space via GET /spaces. If not found: list spaces + ask for confirmation.

Step 2 — Resolve file

If url:

python
import requests, tempfile
from pathlib import Path
from urllib.parse import urlparse

parsed = urlparse(url)
filename = Path(parsed.path).name or "downloaded"
tmp = Path(tempfile.gettempdir()) / filename
with requests.get(url, stream=True, timeout=60) as r:
    r.raise_for_status()
    with open(tmp, "wb") as f:
        for chunk in r.iter_content(8192):
            f.write(chunk)
file_path = str(tmp)

If file_path: validate existence.

Step 3 — Multipart upload

python
from dashboard.backend.sdk_client import evo

with open(file_path, "rb") as f:
    result = evo.post(
        "/api/knowledge/v1/documents",
        files={"file": f},
        data={
            "space": space,
            "unit_id": unit_id,
            "title": title or Path(file_path).stem,
            "tags": ",".join(tags or []),
        },
        headers={"X-Knowledge-Connection": connection},
    )

document_id = result["document_id"]

Endpoint returns 202 Accepted + document_id. Async worker.

Step 4 — Poll status

Interval 2s, timeout 10min:

python
import time
deadline = time.time() + 600
while time.time() < deadline:
    status = evo.get(
        f"/api/knowledge/v1/documents/{document_id}/status",
        headers={"X-Knowledge-Connection": connection},
    )
    phase = status.get("phase")
    if phase in ("done", "ready"):
        break
    if phase == "error":
        raise RuntimeError(status.get("error"))
    time.sleep(2)

Step 5 — Fetch classification (non-blocking)

Classification is asynchronous. 1 extra GET on /documents/{id}:

• content_type != null: show full classification
• Else: "Classification pending — will appear in seconds via async worker"

Output

✓ Document uploaded: {title}
  document_id: {uuid}
  space: {connection}/{space}
  unit: {unit_title or "none"}
  status: ready
  chunks: {N}
  classification:
    content_type: {lesson|tutorial|faq|...}
    difficulty: {...}
    topics: [...]
  elapsed: {X}s

Actionable failures

• File not found → "File does not exist: {path}"
• URL fetch failed → "Download failed: {status_code}"
• Space not found → list available spaces
• Marker models missing → "Run knowledge-admin action=install-parser"
• Timeout → "Timeout after 10min. Status: {phase}. Check knowledge-browse."