Files
ai-resume/corpus/payloads.py
Sam Rolfe f9dc574172 Phase 4: corpus and retrieval — chunks, Qdrant, FTS5, freshness audit
Satisfies docs/phase-4-corpus.md. content/ stays the single source of truth:
one pass over it writes both indexes, so the static pages and the retrieval
corpus cannot drift.

- corpus/: stdlib-only pipeline. validate -> chunk -> enrich -> embed -> upsert.
  Chunking is block-based, so a boundary cannot fall inside a fenced block, a
  table or front-matter (verified structurally, not by inspection).
- Deterministic ids: sha256(source_path + chunk_index + content_hash), rendered
  as a UUID. Re-running updates points; re-chunking deletes the old ones by
  source filter, so duplicates and orphans are both impossible.
- Incremental by content hash against a SQLite manifest. A provider, dimension
  or chunker change is refused instead of mixed: it needs a rebuild.
- Payloads per section 4.2 plus provenance, validated before any vector is
  written; a missing field fails the run and names the document and chunk.
- SQLite FTS5 over chunks (the keyword half of hybrid retrieval, decision D10)
  with area/confidence filters and a `terms` column so front-matter tech and
  tags are findable.
- Freshness audit: stale documents, due soon, index drift, contract drift;
  optional Apprise alert; exit code for Prefect.
- Embeddings behind one interface: omniroute (OpenAI-compatible, retries, key
  from the environment only) and local-hashing (offline placeholder). The
  provider decision is still Sam's; the pipeline does not depend on it.
- deploy/qdrant.yml: mem_limit = memswap_limit = 400m, loopback only.
- docs/phase-4-notes.md: operator manual, DoD evidence, measured memory.
- 76 unit tests, no container and no network.

Measured, Qdrant 1.20.0 under a 400 m cap at 1024 dimensions: 500 / 5,000 /
20,000 chunks settle at 128 / 138 / 151 MiB, peaking at 149 / 305 / 292 MiB
during ingest. Qdrant's default settings are OOM-killed between 5,000 and
20,000, so the collection is created with HNSW and payloads on disk; recorded in
reference/stack.md with the ceiling and the escalation order.

Note: the `make corpus*` targets were swept into the Phase 3 deploy commit
24590cc by concurrent work in this tree; the Makefile content is correct, and
this commit contains the pipeline they call.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-26 21:04:52 +11:00

156 lines
5.7 KiB
Python

"""Chunk payloads: the thing the retrieval filter runs against.
`docs/phase-4-corpus.md` §4.2 fixes the payload. `area` and `confidence` are the
two that matter: they let the query path narrow *before* it searches, and they
let an answer qualify unfinished work instead of presenting it as delivered.
**A missing or invalid field fails the run.** Nothing in this module supplies a
default value: `validate_payload` raises on the first field that is absent,
empty or out of enum, and the pipeline reports which document and chunk it was.
That is the DoD item "a missing field fails the run rather than defaulting".
"""
from __future__ import annotations
import datetime
AREAS = ("ai", "platform", "edge", "products", "history", "about")
LEVELS = (1, 2, 3)
SOURCE_TYPES = ("doc", "code", "adr", "metric")
CONFIDENCE = ("delivered", "running", "in-progress", "abandoned", "historic")
# Fields every payload must carry (docs/phase-4-corpus.md §4.2), plus the ones
# the index needs to answer "where did this come from?" without a second lookup.
REQUIRED_FIELDS = (
"area",
"level",
"source_type",
"confidence",
"period",
"artefacts",
"last_verified",
"text",
)
PROVENANCE_FIELDS = (
"id",
"source_path",
"source_id",
"chunk_index",
"title",
"claim",
"url",
"content_hash",
"embedding_model",
"embedding_dim",
"chunker_version",
)
class PayloadError(RuntimeError):
"""A payload is incomplete or out of enum. The run must stop."""
def source_type_for(doc) -> str:
"""Classify a document. Everything in content/ is a doc unless it says so.
The schema has no `source_type` field and the schema is frozen, so the
classification is derived from tags the schema does allow.
"""
tags = {str(tag).lower() for tag in doc.tags}
if "adr" in tags:
return "adr"
if "metric" in tags:
return "metric"
if "code" in tags:
return "code"
return "doc"
def build_payload(doc, chunk, cfg) -> dict:
"""Build one chunk payload from a document and a chunk.
Values are copied, never guessed: if the document does not carry a field,
the payload does not gain one.
"""
payload = {
# -- the retrieval filter (docs/phase-4-corpus.md §4.2) ------------
"area": doc.area,
"level": doc.level,
"source_type": source_type_for(doc),
"confidence": doc.confidence,
"period": doc.period,
"artefacts": [str(a.get("url", "")) for a in doc.artefacts],
"last_verified": doc.last_verified,
"text": chunk.text,
# -- provenance ----------------------------------------------------
"id": None, # filled by the pipeline once the id is computed
"source_path": doc.path,
"source_id": doc.source_id,
"chunk_index": chunk.index,
"title": doc.title,
"claim": doc.claim,
"url": doc.url,
"content_hash": doc.digest,
"embedding_model": cfg.model,
"embedding_dim": cfg.dim,
"chunker_version": cfg.chunker_version,
# -- supporting evidence, used by the answer to cite ---------------
"public": bool(doc.public),
"tech": list(doc.tech),
"tags": list(doc.tags),
"chunk_tokens": chunk.tokens,
}
return payload
def validate_payload(payload: dict, where: str = "chunk") -> None:
"""Raise PayloadError if a required field is missing, empty or out of enum."""
for field in REQUIRED_FIELDS + PROVENANCE_FIELDS:
if field == "id" and payload.get("id") is None:
raise PayloadError(f"{where}: payload field 'id' was not set")
if field not in payload:
raise PayloadError(f"{where}: payload is missing required field {field!r}")
area = payload["area"]
if area not in AREAS:
raise PayloadError(f"{where}: area {area!r} is not one of {list(AREAS)}")
if payload["level"] not in LEVELS:
raise PayloadError(f"{where}: level {payload['level']!r} is not one of {list(LEVELS)}")
if payload["source_type"] not in SOURCE_TYPES:
raise PayloadError(
f"{where}: source_type {payload['source_type']!r} is not one of {list(SOURCE_TYPES)}"
)
if payload["confidence"] not in CONFIDENCE:
raise PayloadError(
f"{where}: confidence {payload['confidence']!r} is not one of {list(CONFIDENCE)}"
)
for field in ("period", "title", "claim", "url", "source_path", "source_id", "embedding_model"):
value = payload[field]
if not isinstance(value, str) or not value.strip():
raise PayloadError(f"{where}: field {field!r} must be a non-empty string")
text = payload["text"]
if not isinstance(text, str) or not text.strip():
raise PayloadError(f"{where}: field 'text' must be non-empty")
if not isinstance(payload["artefacts"], list):
raise PayloadError(f"{where}: field 'artefacts' must be a list (empty is allowed)")
_check_date(payload["last_verified"], where)
if not isinstance(payload["embedding_dim"], int) or payload["embedding_dim"] <= 0:
raise PayloadError(f"{where}: field 'embedding_dim' must be a positive integer")
if not isinstance(payload["chunk_index"], int) or payload["chunk_index"] < 0:
raise PayloadError(f"{where}: field 'chunk_index' must be a non-negative integer")
if not isinstance(payload["content_hash"], str) or len(payload["content_hash"]) != 64:
raise PayloadError(f"{where}: field 'content_hash' must be a sha256 hex digest")
def _check_date(value: str, where: str) -> None:
try:
datetime.date.fromisoformat(str(value))
except ValueError as exc:
raise PayloadError(
f"{where}: field 'last_verified' must be an ISO date (YYYY-MM-DD), got {value!r}"
) from exc