ingestion-testing

name	ingestion-testing
description	Test and verify the [ORG_NAME] document ingestion pipeline. Use when ingesting files, testing the watcher, verifying Qdrant/SQLite data, running recall accuracy tests, or when the user mentions ingestion, pipeline testing, or dropping files.

Ingestion Pipeline Testing

Product root for commands below: [PRODUCT_CODE_ROOT]/ inside your bound product workspace. Example script names, default DB paths, and Qdrant collection names below are placeholders — substitute values from PRODUCT_CONTEXT / .env for the product you are testing (this portable Method pack does not ship a real ingest tree).

Pre-Flight Checks

Before any ingestion work, run these in order:

1. Check for duplicate watcher processes

ps aux | grep watch_ingest | grep -v grep

If multiple processes exist, stop extras (adjust pattern to your watcher script name):

pkill -f watch_ingest.py

2. Verify infrastructure

# Qdrant
curl -s http://127.0.0.1:6333/healthz

# llama.cpp embed lane (OpenAI-compatible `llama-server`) — not Ollama
EMBED_URL="${[ORG_NAME_UPPER]_INFERENCE_EMBED_URL:-http://127.0.0.1:11437}"
curl -s -o /dev/null -w "embed_lane_http=%{http_code}\n" "${EMBED_URL}/v1/models"

If that fails, start lanes per your product scripts (e.g. scripts/start-llama-lanes.sh) and docs/runbooks/ for inference.

3. Capture memory baseline

free -h | head -3

4. VRAM / lanes (ingestion uses embed lane only)

Adopter note: [ORG_NAME_UPPER] is a placeholder — replace it with your org name in UPPER_SNAKE_CASE (e.g. if your org is ACME, the env var becomes ACME_INFERENCE_EMBED_URL). Set these in your .env bootstrap and they propagate everywhere below.

Ingestion and RAG embeddings use [ORG_NAME_UPPER]_INFERENCE_EMBED_URL — the port in the example above (11437) is illustrative; use whatever port your embed lane runs on. Chat lanes (e.g. :11435, :11436) are separate llama-server processes — port numbers vary by your setup.

If embed and chat share one GPU and VRAM is tight before a big ingest, stop or avoid starting chat lanes for the ingest window — use your stack control script / lane runbook. Do not use Ollama keep_alive tricks unless Ollama is your documented inference path.

5. Pre-warm embedding model

Match EMBEDDING_MODEL and any product-specific text-embedding env vars from .env (must match the model loaded on the embed llama-server):

EMBED_URL="${[ORG_NAME_UPPER]_INFERENCE_EMBED_URL:-http://127.0.0.1:11437}"
EMBED_MODEL="${EMBEDDING_MODEL:-nomic-embed-text}"
curl -s "${EMBED_URL}/v1/embeddings" \
  -H "Content-Type: application/json" \
  -d "{\"model\":\"${EMBED_MODEL}\",\"input\":\"warmup\"}" > /dev/null

Clean State (if doing a fresh test)

Only wipe if testing from scratch. Skip if appending to existing data.

# SQLite (path from [ORG_NAME_UPPER]_SQLITE_PATH or your default under [PRODUCT_CODE_ROOT]/)
rm -f tmp/app.db*

# Qdrant — collection name from QDRANT_COLLECTION (example: documents)
COL="${QDRANT_COLLECTION:-documents}"
curl -s -X DELETE "http://127.0.0.1:6333/collections/${COL}"

# Watcher drop folders (under [PRODUCT_CODE_ROOT]/)
rm -f data/ingest/processed/* data/ingest/failed/* data/ingest/incoming/*

Start API + watcher

1. Start the API (however you normally run it) so POST /api/v1/ingest/upload is available (API_PORT default 8000).
2. Set env: [ORG_NAME_UPPER]_WATCH_TENANT_ID, and either [ORG_NAME_UPPER]_WATCH_TOKEN or [ORG_NAME_UPPER]_WATCH_USERNAME + [ORG_NAME_UPPER]_WATCH_PASSWORD + optional [ORG_NAME_UPPER]_WATCH_TENANT_ID for login. Optionally [ORG_NAME_UPPER]_API_URL (default http://127.0.0.1:8000).

cd [ORG_NAME]
export [ORG_NAME_UPPER]_WATCH_TENANT_ID=example-tenant
export [ORG_NAME_UPPER]_WATCH_USERNAME=operator
export [ORG_NAME_UPPER]_WATCH_PASSWORD='<from .env bootstrap>'
python3 -u scripts/watch_ingest.py

Expected console lines (prefixes vary by product):

• Watcher logs API URL and tenant
• Incoming path activity
• Per-file ingest OK / FAIL

Failed uploads may get a sidecar JSON next to the file in data/ingest/failed/ (exact name per your implementation).

Drop Files

Copy or move files into data/ingest/incoming/ (create dirs if missing). Allowed suffixes: .md, .txt, .json, .jsonl, .csv, .log, .pdf, .yaml, .yml.

cp path/to/sample.pdf data/ingest/incoming/

Monitor Ingestion

Watch API logs and watcher stdout. On success the file moves to data/ingest/processed/; on failure to data/ingest/failed/.

Text embeddings use the env vars your API documents (commonly EMBEDDING_MODEL plus any org-prefixed override). Vision models for PDF/image paths are product-specific — see your services/api/ tree. Do not treat this pack as naming those variables.

Verify Data

Qdrant

Default collection: from QDRANT_COLLECTION (example below uses documents).

import os
import requests, json

col = os.environ.get("QDRANT_COLLECTION", "documents")
r = requests.get(f"http://127.0.0.1:6333/collections/{col}", timeout=10)
info = r.json()["result"]
print(f"Points: {info['points_count']}, Status: {info['status']}")

SQLite

Default DB: path from your *_SQLITE_PATH env var or dev default (example tmp/app.db).

import os
import sqlite3
from pathlib import Path

db = Path(os.environ.get("YOUR_ORG_SQLITE_PATH", "tmp/app.db"))
conn = sqlite3.connect(db)
conn.row_factory = sqlite3.Row

docs = conn.execute("SELECT doc_id, title, doc_type, status FROM documents").fetchall()
print(f"Documents: {len(docs)}")
chunks = conn.execute("SELECT COUNT(*) AS c FROM vector_chunks").fetchone()["c"]
print(f"vector_chunks: {chunks}")
conn.close()

Run the snippet from [PRODUCT_CODE_ROOT]/ so the DB path resolves.

Recall Accuracy Test

Point queries at QDRANT_COLLECTION and filter on payload fields your ingest sets (e.g. document_id, tenant_id). Adjust model name to match .env.

import os
import requests

EMBED_BASE = os.environ.get("YOUR_ORG_INFERENCE_EMBED_URL", "http://127.0.0.1:11437").rstrip("/")
QDRANT = "http://127.0.0.1:6333"
COLLECTION = os.environ.get("QDRANT_COLLECTION", "documents")
EMBED_MODEL = os.environ.get("EMBEDDING_MODEL") or "nomic-embed-text"

def embed(text):
    r = requests.post(
        f"{EMBED_BASE}/v1/embeddings",
        json={"model": EMBED_MODEL, "input": text},
        timeout=120,
    )
    r.raise_for_status()
    return r.json()["data"][0]["embedding"]

def search(query, tenant_id=None, limit=5):
    vec = embed(query)
    body = {"vector": vec, "limit": limit, "with_payload": True}
    if tenant_id:
        body["filter"] = {"must": [{"key": "tenant_id", "match": {"value": tenant_id}}]}
    r = requests.post(
        f"{QDRANT}/collections/{COLLECTION}/points/search",
        json=body,
        timeout=30,
    )
    r.raise_for_status()
    return r.json().get("result", [])

for q in ["your procedural query here", "another domain-specific query"]:
    hits = search(q, tenant_id="example-tenant")
    top = hits[0] if hits else None
    score = top.get("score") if top else None
    title = (top.get("payload") or {}).get("title", "?")[:80] if top else "?"
    print(f"Q: {q}\n  score={score} title={title}\n")

Score Expectations

Rough guidance only; depends on model and corpus.

Query Type	Expected Score	Rating
Procedural ("how do I remove...")	>= 0.70	Excellent
Specific component/tool	>= 0.65	Good
Abstract/general	>= 0.55	Fair

Post-Test

1. Capture memory: free -h | head -3
2. Log results to notes/session_YYYY-MM-DD.txt (workspace root)
3. Confirm data/ingest/incoming/ is empty; inspect data/ingest/failed/ if anything failed