Files
kb/engine/tests/test_ingest_data.py
T
steve 6dfc13be1d Add reranking, RRF fusion, bench harness, tag contexts, and data ingestion
Implements five of the six enhancements from docs/kb-enhancements-proposal.htm,
closing the retrieval-quality gap identified in the qmd review.

- Cross-encoder reranking: new kb/reranker.py loads an optional reranking
  model at startup (KB_RERANK_ENABLED, KB_RERANKER_MODEL,
  KB_RERANK_CANDIDATES). Search degrades gracefully to plain hybrid
  retrieval when the model is absent. Exposed via a "rerank" block in
  /status, a rerank flag on search, and --no-rerank in the CLI.
- RRF rank fusion: FTS and vector lists now merge by reciprocal rank
  fusion with a top-rank bonus, replacing the old score blend. Scores are
  comparable across queries.
- Bench harness and explain traces: kb bench runs a query fixture against
  each backend and reports precision@k, recall and MRR. --explain returns a
  per-result score breakdown.
- Tag context descriptions: tags carry an optional one-line description
  (kb tag-describe), returned as tag_contexts with search results. Adds a
  tags.description column migration.
- Structured data ingestion: .json/.yaml/.toml files ingest as text via the
  new "data" doc type, pretty-printing minified JSON before chunking.

Query expansion (proposal item 5) is deliberately left out pending bench
results. Requires engine v3.3.0.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-21 09:51:33 +01:00

65 lines
2.1 KiB
Python

"""Tests for structured-data (.json/.yaml/.toml) ingestion."""
import json
from pathlib import Path
import pytest
from kb.ingest.data import chunk_data
from kb.ingest.detector import detect_type, is_supported
def test_detector_accepts_data_extensions():
assert detect_type(Path("config.json")) == ("data", "json")
assert detect_type(Path("stack.yaml")) == ("data", "yaml")
assert detect_type(Path("stack.yml")) == ("data", "yaml")
assert detect_type(Path("pyproject.toml")) == ("data", "toml")
for name in ("a.json", "b.yaml", "c.yml", "d.toml"):
assert is_supported(Path(name))
def test_minified_json_is_pretty_printed():
minified = json.dumps({"hosts": [{"name": "web1", "ip": "10.0.0.1"}]})
assert "\n" not in minified
chunks = chunk_data(minified, "json")
assert len(chunks) == 1
text = chunks[0]["text"]
assert "\n" in text, "expected pretty-printed multi-line JSON"
assert '"name": "web1"' in text
def test_large_minified_json_multi_chunks():
big = json.dumps([{"id": i, "payload": "x" * 100} for i in range(200)])
chunks = chunk_data(big, "json", max_tokens=256)
assert len(chunks) > 1, "large JSON must split into multiple chunks"
# Pretty-printing means chunks break on lines, not mid-token blobs.
for c in chunks:
assert c["text"].strip()
assert [c["chunk_index"] for c in chunks] == list(range(len(chunks)))
def test_malformed_json_ingests_raw():
broken = '{"unterminated": [1, 2'
chunks = chunk_data(broken, "json")
assert len(chunks) == 1
assert chunks[0]["text"] == broken
def test_yaml_passes_through_unchanged():
yaml_text = "services:\n web:\n image: nginx\n"
chunks = chunk_data(yaml_text, "yaml")
assert len(chunks) == 1
assert chunks[0]["text"] == yaml_text.strip()
def test_toml_passes_through():
toml_text = '[tool.example]\nname = "kb"\n'
chunks = chunk_data(toml_text, "toml")
assert len(chunks) == 1
assert chunks[0]["text"] == toml_text.strip()
def test_empty_file_yields_no_chunks():
assert chunk_data("", "json") == []
assert chunk_data(" \n ", "yaml") == []