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>
113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
"""An in-memory stand-in for Qdrant, used by the pipeline tests.
|
|
|
|
It implements the routes the pipeline actually calls and nothing else, and it is
|
|
deliberately dumb: brute-force cosine over a dict. If the pipeline passes
|
|
against this, the pipeline's own logic is right; the real server is exercised
|
|
separately in `docs/phase-4-notes.md` (rebuild against a live container).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
|
|
from corpus.qdrant import Qdrant, QdrantError, QdrantNotFound
|
|
|
|
|
|
class FakeQdrant(Qdrant):
|
|
def __init__(self, dim: int):
|
|
super().__init__(url="fake://qdrant", timeout=1.0, attempts=1)
|
|
self.dim = dim
|
|
self.collections = {} # name -> {id: {"vector": [...], "payload": {...}}}
|
|
self.calls = []
|
|
|
|
# -- transport --------------------------------------------------------
|
|
|
|
def _request(self, method: str, path: str, body=None) -> dict:
|
|
self.calls.append((method, path, body))
|
|
query = ""
|
|
if "?" in path:
|
|
path, query = path.split("?", 1)
|
|
parts = [part for part in path.split("/") if part]
|
|
|
|
if not parts:
|
|
return {"version": "fake-1.0"}
|
|
|
|
if parts[0] != "collections":
|
|
raise QdrantNotFound(f"no route {path}")
|
|
if len(parts) == 2:
|
|
name = parts[1]
|
|
if method == "PUT":
|
|
self.collections.setdefault(name, {})
|
|
return {"result": True}
|
|
if method == "DELETE":
|
|
if name not in self.collections:
|
|
raise QdrantNotFound("collection not found")
|
|
del self.collections[name]
|
|
return {"result": True}
|
|
if name not in self.collections:
|
|
raise QdrantNotFound("collection not found")
|
|
return {
|
|
"result": {
|
|
"points_count": len(self.collections[name]),
|
|
"config": {"params": {"vectors": {"size": self.dim, "distance": "Cosine"}}},
|
|
}
|
|
}
|
|
name = parts[1]
|
|
if name not in self.collections:
|
|
raise QdrantNotFound("collection not found")
|
|
points = self.collections[name]
|
|
|
|
if parts[2:4] == ["points", "count"]:
|
|
return {"result": {"count": len(points)}}
|
|
if len(parts) == 3 and parts[2] == "points" and method == "PUT":
|
|
for point in (body or {}).get("points", []):
|
|
if len(point["vector"]) != self.dim:
|
|
raise QdrantError("wrong vector dimension")
|
|
points[point["id"]] = {"vector": point["vector"], "payload": point["payload"]}
|
|
return {"result": {"status": "completed"}}
|
|
if parts[2:4] == ["points", "delete"]:
|
|
must = (((body or {}).get("filter") or {}).get("must")) or []
|
|
for point_id in list(points):
|
|
if _matches(points[point_id]["payload"], must):
|
|
del points[point_id]
|
|
return {"result": {"status": "completed"}}
|
|
if parts[2:4] == ["points", "query"]:
|
|
vector = body["query"]
|
|
must = (((body or {}).get("filter") or {}).get("must")) or []
|
|
scored = []
|
|
for point_id, point in points.items():
|
|
if not _matches(point["payload"], must):
|
|
continue
|
|
scored.append(
|
|
{
|
|
"id": point_id,
|
|
"score": _cosine(vector, point["vector"]),
|
|
"payload": point["payload"],
|
|
}
|
|
)
|
|
scored.sort(key=lambda row: row["score"], reverse=True)
|
|
return {"result": {"points": scored[: int(body.get("limit", 10))]}}
|
|
if parts[2:4] == ["points", "scroll"]:
|
|
limit = int((body or {}).get("limit", 100))
|
|
rows = [
|
|
{"id": point_id, "payload": {"source_path": point["payload"].get("source_path")}}
|
|
for point_id, point in list(points.items())[:limit]
|
|
]
|
|
return {"result": {"points": rows, "next_page_offset": None}}
|
|
raise QdrantNotFound(f"no route {path}")
|
|
|
|
|
|
def _matches(payload: dict, conditions: list) -> bool:
|
|
for condition in conditions:
|
|
key = condition.get("key")
|
|
expected = (condition.get("match") or {}).get("value")
|
|
if payload.get(key) != expected:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _cosine(left: list, right: list) -> float:
|
|
dot = sum(a * b for a, b in zip(left, right))
|
|
norm = math.sqrt(sum(a * a for a in left)) * math.sqrt(sum(b * b for b in right))
|
|
return dot / norm if norm else 0.0
|