Sam Rolfe
f9dc574172
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>
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""Prefect entry points for the corpus pipeline.
|
|
|
|
`docs/phase-4-corpus.md` §4.4: *"Prefect deployment on the existing Prefect
|
|
server; worker only, do not deploy a second server. Trigger: git push to
|
|
content/, plus a nightly audit."*
|
|
|
|
Prefect is optional at runtime. If it is not installed, these functions are
|
|
plain callables and the same code runs straight from the CLI
|
|
(`python3 -m corpus.cli sync`), which is what `make corpus` does. Nothing about
|
|
the pipeline depends on the orchestrator, and the orchestrator depends on
|
|
nothing but the pipeline.
|
|
|
|
The deployment commands (run on `.13`, which already has the Prefect server) are
|
|
in `docs/phase-4-notes.md` §6. This file is only the work definition.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
try: # pragma: no cover - exercised only where Prefect is installed
|
|
from prefect import flow as _flow
|
|
from prefect import get_run_logger as _get_run_logger
|
|
|
|
PREFECT_AVAILABLE = True
|
|
except ImportError: # pragma: no cover
|
|
PREFECT_AVAILABLE = False
|
|
|
|
def _flow(*_args, **_kwargs):
|
|
def decorator(func):
|
|
return func
|
|
|
|
return decorator
|
|
|
|
def _get_run_logger():
|
|
class _Logger:
|
|
def info(self, message, *args, **kwargs):
|
|
print(message, file=sys.stderr)
|
|
|
|
warning = error = info
|
|
|
|
return _Logger()
|
|
|
|
|
|
from corpus import freshness, pipeline # noqa: E402 (after the Prefect shim)
|
|
from corpus.config import Config # noqa: E402
|
|
|
|
|
|
@_flow(name="corpus-sync", log_prints=True)
|
|
def sync_corpus() -> dict:
|
|
"""Incremental refresh. Runs on every push that touches content/."""
|
|
logger = _get_run_logger()
|
|
cfg = Config.from_env()
|
|
stats = pipeline.run(cfg, mode="sync", log=logger.info)
|
|
return stats.to_dict()
|
|
|
|
|
|
@_flow(name="corpus-rebuild", log_prints=True)
|
|
def rebuild_corpus() -> dict:
|
|
"""From-scratch rebuild. The recovery path: deleting the volume is always safe."""
|
|
logger = _get_run_logger()
|
|
cfg = Config.from_env()
|
|
stats = pipeline.run(cfg, mode="rebuild", log=logger.info)
|
|
return stats.to_dict()
|
|
|
|
|
|
@_flow(name="corpus-audit", log_prints=True)
|
|
def freshness_audit(alert: bool = True) -> dict:
|
|
"""Nightly audit. Reports stale documents and index drift, then alerts."""
|
|
logger = _get_run_logger()
|
|
cfg = Config.from_env()
|
|
report = freshness.audit(cfg, log=logger.info)
|
|
logger.info(freshness.render_text(report))
|
|
if alert and not report["ok"]:
|
|
try:
|
|
freshness.alert(report, cfg.apprise_url)
|
|
except Exception as exc: # noqa: BLE001 - the alert must not fail the audit
|
|
logger.warning(f"Apprise alert failed: {exc}")
|
|
return report
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover - convenience for a manual run
|
|
command = sys.argv[1] if len(sys.argv) > 1 else "sync"
|
|
if command == "audit":
|
|
print(freshness.render_text(freshness_audit(alert=False)))
|
|
elif command == "rebuild":
|
|
print(rebuild_corpus())
|
|
else:
|
|
print(sync_corpus())
|