---
name: evolution-foundation/knowledge-ingest
source: https://app.decimal.ai/s/evolution-foundation-knowledge-ingest@1/SKILL.md
source_sha256: 2dcd71d0d5f1
---

# 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` | list[str] | 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`."