Levandor

Knowledgebase vectorization (phase 1) — implementation plan

11 min read

KB Vectorization (Phase 1) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax.

Goal: Embed every OCR’d manual section into a Qdrant vector DB with per-chunk metadata (manual, page URL, images) and provide semantic search.

Architecture: A ~/kb-vectors/ Python project (CPU-torch + sentence-transformers bge-large-en-v1.5 + qdrant-client) reads the knowledgebase’s per-section markdown, sub-chunks it, embeds it, and upserts into a Qdrant docker container. Re-runnable/idempotent per manual.

Tech Stack: Python venv (CPU torch, sentence-transformers, qdrant-client), Qdrant (docker), the knowledgebase docs at ~/knowledgebase/manuals-src/docs/.

Design: 2026-07-24-kb-vectorize-design

Global Constraints

  • All code under /home/levander/kb-vectors/ on telep-mainframe. Runs as levander (Qdrant via docker; levander is in the docker group).
  • No comments/docstrings/annotations in any code (user’s global rule). Self-explanatory naming only.
  • Do not run git commit. Each task ends in verification.
  • CPU torch (install from the cpu index) — sidesteps the CUDA-13-vs-driver-550 trap we hit with marker AND avoids GPU contention with the OCR queue + Frigate. Embedding a few thousand chunks on the 12900K is fine.
  • Embeddings: BAAI/bge-large-en-v1.5, 1024-dim, cosine. Passages: no prefix. Queries: prefix "Represent this sentence for searching relevant passages: ". Normalize embeddings.
  • Qdrant: container bound to 127.0.0.1:6333 (REST) + 127.0.0.1:6334 (gRPC), storage volume ~/kb-vectors/qdrant-storage. Collection manuals.
  • Deterministic point IDs (uuid5 of folder/manual/page/chunk_index) so re-index replaces, never duplicates.
  • Read-only over ~/knowledgebase/manuals-src/docs/ — must not modify the live KB.
  • Run commands on the box: /usr/bin/ssh levander@telep-mainframe. On this Mac bare ssh is broken — always /usr/bin/ssh.

Test command: cd /home/levander/kb-vectors && venv/bin/python -m unittest <module> -v

Task 1: Project setup — venv, deps, Qdrant container

Files:

  • Create: /home/levander/kb-vectors/docker-compose.yml

  • Step 1: Create project + venv + CPU torch + deps

/usr/bin/ssh levander@telep-mainframe 'mkdir -p /home/levander/kb-vectors/qdrant-storage && cd /home/levander/kb-vectors && python3 -m venv venv && venv/bin/pip install -q --upgrade pip && venv/bin/pip install -q torch --index-url https://download.pytorch.org/whl/cpu && venv/bin/pip install -q sentence-transformers qdrant-client && venv/bin/python -c "import torch, sentence_transformers, qdrant_client; print(\"torch\", torch.__version__, \"| cuda\", torch.cuda.is_available(), \"| st ok | qc ok\")"'

Expected: prints torch version, cuda False (CPU build — intended), and no import errors. (Large download; torch CPU ~200MB, sentence-transformers deps.)

  • Step 2: Start Qdrant as a docker container
/usr/bin/ssh levander@telep-mainframe 'cat > /home/levander/kb-vectors/docker-compose.yml <<'"'"'YML'"'"'
services:
  qdrant:
    image: qdrant/qdrant:latest
    container_name: kb-qdrant
    restart: unless-stopped
    ports:
      - "127.0.0.1:6333:6333"
      - "127.0.0.1:6334:6334"
    volumes:
      - /home/levander/kb-vectors/qdrant-storage:/qdrant/storage
YML
cd /home/levander/kb-vectors && docker compose up -d && sleep 5 && curl -s http://127.0.0.1:6333/ | head -c 200; echo; docker ps --filter name=kb-qdrant --format "{{.Names}} {{.Status}}"'

Expected: a Qdrant JSON banner (version info) and kb-qdrant Up.

  • Step 3: Verify the Python client reaches Qdrant
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -c "
from qdrant_client import QdrantClient
c = QdrantClient(host=\"127.0.0.1\", port=6333)
print(\"collections:\", c.get_collections())
"'

Expected: prints an (empty) collections list, no connection error.

Task 2: chunker.py — read a manual dir into token-bounded chunks

Files:

  • Create: /home/levander/kb-vectors/chunker.py
  • Test: /home/levander/kb-vectors/test_chunker.py

Interfaces:

  • Produces:

    • heading_of(text) -> str — first #/## heading text, else ""
    • images_of(text) -> list[str] — the ![](name) refs in order
    • chunk_text(text, max_words=350, overlap=50) -> list[str] — word windows with overlap; [] if empty; single chunk if short
    • chunk_manual(manual_dir, folder, manual) -> list[dict] — per section file, one dict per chunk with keys: folder, manual, manual_id, page, page_url, heading, images, text, chunk_index

  • Step 1: Write the failing test

Create /home/levander/kb-vectors/test_chunker.py:

import os
import tempfile
import unittest
 
from chunker import chunk_manual, chunk_text, heading_of, images_of
 
SECTION = """## Brake Bleeding
 
Loosen the bleeder screw.
 
![](_page_12_Figure_1.jpeg)
 
Tighten to spec.
 
![](_page_12_Picture_3.jpeg)
"""
 
 
class TestHelpers(unittest.TestCase):
    def test_heading(self):
        self.assertEqual(heading_of(SECTION), "Brake Bleeding")
 
    def test_heading_none(self):
        self.assertEqual(heading_of("no heading here"), "")
 
    def test_images(self):
        self.assertEqual(images_of(SECTION), ["_page_12_Figure_1.jpeg", "_page_12_Picture_3.jpeg"])
 
    def test_chunk_short_single(self):
        self.assertEqual(chunk_text("a b c"), ["a b c"])
 
    def test_chunk_empty(self):
        self.assertEqual(chunk_text("   "), [])
 
    def test_chunk_long_overlap(self):
        text = " ".join(str(i) for i in range(800))
        chunks = chunk_text(text, max_words=350, overlap=50)
        self.assertGreaterEqual(len(chunks), 2)
        first = chunks[0].split()
        second = chunks[1].split()
        self.assertEqual(first[-50:], second[:50])
 
 
class TestChunkManual(unittest.TestCase):
    def _manual(self):
        base = tempfile.mkdtemp()
        d = os.path.join(base, "suzuki-vitara", "5door-supplement")
        os.makedirs(d)
        with open(os.path.join(d, "001-intro.md"), "w", encoding="utf-8") as h:
            h.write(SECTION)
        with open(os.path.join(d, "readme.txt"), "w") as h:
            h.write("ignore me")
        return d
 
    def test_chunk_manual(self):
        d = self._manual()
        chunks = chunk_manual(d, "suzuki-vitara", "5door-supplement")
        self.assertEqual(len(chunks), 1)
        c = chunks[0]
        self.assertEqual(c["manual_id"], "suzuki-vitara/5door-supplement")
        self.assertEqual(c["page"], "001-intro")
        self.assertEqual(c["page_url"], "suzuki-vitara/5door-supplement/001-intro/")
        self.assertEqual(c["heading"], "Brake Bleeding")
        self.assertIn("_page_12_Figure_1.jpeg", c["images"])
        self.assertEqual(c["chunk_index"], 0)
 
    def test_ignores_non_md(self):
        d = self._manual()
        chunks = chunk_manual(d, "suzuki-vitara", "5door-supplement")
        self.assertTrue(all(c["page"] != "readme" for c in chunks))
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -m unittest test_chunker -v'

Expected: FAIL with ModuleNotFoundError: No module named 'chunker'

  • Step 3: Write the implementation

Create /home/levander/kb-vectors/chunker.py:

import os
import re
 
_IMG = re.compile(r"!\[\]\(([^)]+)\)")
 
 
def heading_of(text):
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("# ") or stripped.startswith("## "):
            return stripped.lstrip("#").strip()
    return ""
 
 
def images_of(text):
    return _IMG.findall(text)
 
 
def chunk_text(text, max_words=350, overlap=50):
    words = text.split()
    if not words:
        return []
    if len(words) <= max_words:
        return [text.strip()]
    step = max_words - overlap
    chunks = []
    index = 0
    while index < len(words):
        chunks.append(" ".join(words[index:index + max_words]))
        if index + max_words >= len(words):
            break
        index += step
    return chunks
 
 
def chunk_manual(manual_dir, folder, manual):
    results = []
    for name in sorted(os.listdir(manual_dir)):
        if not name.endswith(".md"):
            continue
        with open(os.path.join(manual_dir, name), encoding="utf-8") as handle:
            text = handle.read()
        page = name[:-3]
        heading = heading_of(text)
        images = images_of(text)
        page_url = folder + "/" + manual + "/" + page + "/"
        for index, chunk in enumerate(chunk_text(text)):
            if not chunk.strip():
                continue
            results.append(
                {
                    "folder": folder,
                    "manual": manual,
                    "manual_id": folder + "/" + manual,
                    "page": page,
                    "page_url": page_url,
                    "heading": heading,
                    "images": images,
                    "text": chunk,
                    "chunk_index": index,
                }
            )
    return results
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -m unittest test_chunker -v'

Expected: OK — 9 tests pass.

  • Step 5: Verify against a real manual
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -c "
from chunker import chunk_manual
c = chunk_manual(\"/home/levander/knowledgebase/manuals-src/docs/suzuki-vitara/5door-supplement\", \"suzuki-vitara\", \"5door-supplement\")
print(len(c), \"chunks\")
print(\"sample:\", {k: c[0][k] for k in [\"page_url\",\"heading\",\"chunk_index\"]}, \"images:\", len(c[0][\"images\"]))
"'

Expected: a chunk count (more than the 59 section files, since long sections split) and a sensible sample.

Task 3: embed.py — local bge embeddings

Files:

  • Create: /home/levander/kb-vectors/embed.py

Interfaces:

  • Produces:

    • embed_passages(texts) -> list[list[float]] — normalized, no prefix
    • embed_query(text) -> list[float] — normalized, with the bge query instruction prefix
    • DIM = 1024

  • Step 1: Write the implementation

Create /home/levander/kb-vectors/embed.py:

from sentence_transformers import SentenceTransformer
 
MODEL_NAME = "BAAI/bge-large-en-v1.5"
QUERY_PREFIX = "Represent this sentence for searching relevant passages: "
DIM = 1024
 
_model = None
 
 
def _get_model():
    global _model
    if _model is None:
        _model = SentenceTransformer(MODEL_NAME, device="cpu")
    return _model
 
 
def embed_passages(texts):
    vectors = _get_model().encode(texts, normalize_embeddings=True, batch_size=16)
    return [vector.tolist() for vector in vectors]
 
 
def embed_query(text):
    vector = _get_model().encode([QUERY_PREFIX + text], normalize_embeddings=True)[0]
    return vector.tolist()
  • Step 2: Verify the model loads and embeds (first run downloads ~1.3GB)
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -c "
from embed import embed_passages, embed_query, DIM
p = embed_passages([\"loosen the bleeder screw\", \"torque the axle nut\"])
q = embed_query(\"how to bleed brakes\")
print(\"passage dim:\", len(p[0]), \"count:\", len(p))
print(\"query dim:\", len(q))
import math
print(\"normalized:\", round(math.sqrt(sum(x*x for x in q)), 3))
assert len(p[0]) == DIM and len(q) == DIM
"'

Expected: dims 1024, count 2, normalized ≈ 1.0.

Task 4: index.py + search.py — Qdrant integration

Files:

  • Create: /home/levander/kb-vectors/index.py
  • Create: /home/levander/kb-vectors/search.py

Interfaces:

  • index.py: get_client(), ensure_collection(client), index_manual(client, docs_dir, folder, manual) -> int, index_all(client, docs_dir) -> dict, COLLECTION, point_count(client)

  • search.py: search(client, query, limit=8, folder=None, manual=None) -> list[dict]

  • Step 1: Write index.py

Create /home/levander/kb-vectors/index.py:

import os
import uuid
 
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, FieldCondition, Filter, MatchValue, PointStruct, VectorParams
 
from chunker import chunk_manual
from embed import DIM, embed_passages
 
COLLECTION = "manuals"
 
 
def get_client():
    return QdrantClient(host="127.0.0.1", port=6333)
 
 
def ensure_collection(client):
    if not client.collection_exists(COLLECTION):
        client.create_collection(
            COLLECTION,
            vectors_config=VectorParams(size=DIM, distance=Distance.COSINE),
        )
 
 
def _point_id(chunk):
    key = chunk["manual_id"] + "/" + chunk["page"] + "/" + str(chunk["chunk_index"])
    return str(uuid.uuid5(uuid.NAMESPACE_URL, key))
 
 
def index_manual(client, docs_dir, folder, manual):
    ensure_collection(client)
    manual_dir = os.path.join(docs_dir, folder, manual)
    chunks = chunk_manual(manual_dir, folder, manual)
    if not chunks:
        return 0
    vectors = embed_passages([chunk["text"] for chunk in chunks])
    points = [
        PointStruct(id=_point_id(chunk), vector=vector, payload=chunk)
        for chunk, vector in zip(chunks, vectors)
    ]
    client.delete(
        COLLECTION,
        points_selector=Filter(
            must=[FieldCondition(key="manual_id", match=MatchValue(value=folder + "/" + manual))]
        ),
    )
    client.upsert(COLLECTION, points)
    return len(points)
 
 
def index_all(client, docs_dir):
    counts = {}
    for folder in sorted(os.listdir(docs_dir)):
        folder_path = os.path.join(docs_dir, folder)
        if not os.path.isdir(folder_path):
            continue
        for manual in sorted(os.listdir(folder_path)):
            if os.path.isdir(os.path.join(folder_path, manual)):
                counts[folder + "/" + manual] = index_manual(client, docs_dir, folder, manual)
    return counts
 
 
def point_count(client):
    return client.count(COLLECTION).count
  • Step 2: Write search.py

Create /home/levander/kb-vectors/search.py:

from qdrant_client.models import FieldCondition, Filter, MatchValue
 
from embed import embed_query
from index import COLLECTION
 
 
def search(client, query, limit=8, folder=None, manual=None):
    conditions = []
    if folder:
        conditions.append(FieldCondition(key="folder", match=MatchValue(value=folder)))
    if manual:
        conditions.append(FieldCondition(key="manual_id", match=MatchValue(value=manual)))
    query_filter = Filter(must=conditions) if conditions else None
    response = client.query_points(
        COLLECTION,
        query=embed_query(query),
        limit=limit,
        query_filter=query_filter,
        with_payload=True,
    )
    hits = []
    for point in response.points:
        payload = dict(point.payload)
        payload["score"] = point.score
        hits.append(payload)
    return hits
  • Step 3: Index one manual and verify idempotency + count
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -c "
from index import get_client, index_manual, point_count
c = get_client()
docs = \"/home/levander/knowledgebase/manuals-src/docs\"
n1 = index_manual(c, docs, \"suzuki-vitara\", \"5door-supplement\")
total1 = point_count(c)
n2 = index_manual(c, docs, \"suzuki-vitara\", \"5door-supplement\")
total2 = point_count(c)
print(\"indexed:\", n1, \"total after 1st:\", total1, \"| after re-index:\", total2)
assert total1 == total2, \"re-index duplicated points\"
print(\"IDEMPOTENT OK\")
"'

Expected: indexed: N, equal totals, IDEMPOTENT OK.

  • Step 4: Real search — the quality proof
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python -c "
from index import get_client
from search import search
c = get_client()
for q in [\"how to bleed the brakes\", \"torque specification\", \"differential\"]:
    hits = search(c, q, limit=3, folder=\"suzuki-vitara\")
    print(\"\n Q:\", q)
    for h in hits:
        print(\"  \", round(h[\"score\"],3), h[\"manual_id\"], \"|\", h[\"heading\"][:50], \"| imgs:\", len(h[\"images\"]))
"'

Expected: relevant hits (right manual/section, plausible headings), scores descending, image counts present. This is the real proof the pipeline works.

Task 5: kbvec.py CLI + full index + end-to-end

Files:

  • Create: /home/levander/kb-vectors/kbvec.py

Interfaces:

  • CLI: index --all / index --manual <folder/manual>, search "<query>" [--folder F] [--manual F/M], stats

  • Step 1: Write kbvec.py

Create /home/levander/kb-vectors/kbvec.py:

import argparse
 
from index import get_client, index_all, index_manual, point_count
from search import search
 
DOCS = "/home/levander/knowledgebase/manuals-src/docs"
 
 
def main():
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)
    index_parser = sub.add_parser("index")
    index_parser.add_argument("--all", action="store_true")
    index_parser.add_argument("--manual")
    search_parser = sub.add_parser("search")
    search_parser.add_argument("query")
    search_parser.add_argument("--folder")
    search_parser.add_argument("--manual")
    search_parser.add_argument("--limit", type=int, default=8)
    sub.add_parser("stats")
    args = parser.parse_args()
    client = get_client()
    if args.command == "index":
        if args.manual:
            folder, manual = args.manual.split("/", 1)
            print(index_manual(client, DOCS, folder, manual), "points:", args.manual)
        else:
            for name, count in index_all(client, DOCS).items():
                print(count, name)
        print("total:", point_count(client))
    elif args.command == "search":
        for hit in search(client, args.query, limit=args.limit, folder=args.folder, manual=args.manual):
            print(round(hit["score"], 3), hit["page_url"], "|", hit["heading"][:60])
    elif args.command == "stats":
        print("total points:", point_count(client))
 
 
if __name__ == "__main__":
    main()
  • Step 2: Index ALL current manuals
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && venv/bin/python kbvec.py index --all'

Expected: a per-manual point count for every manual currently in the KB (5door-supplement, supplement-61a40, workshop-1988-1998, owners-1995, and any others landed by then), and a non-zero total. (Takes a few minutes on CPU.)

  • Step 3: End-to-end CLI search + filter
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/kb-vectors && echo "=== all manuals ===" && venv/bin/python kbvec.py search "how to replace the differential" --limit 5 && echo "=== vitara only ===" && venv/bin/python kbvec.py search "brake bleeding procedure" --folder suzuki-vitara --limit 5 && echo "=== stats ===" && venv/bin/python kbvec.py stats'

Expected: relevant results across manuals; the filtered query returns only suzuki-vitara/* page URLs; stats shows the total. Confirm the differential query surfaces the workshop manual’s differential section.

  • Step 4: Confirm nothing was disturbed
/usr/bin/ssh levander@telep-mainframe 'systemctl is-active knowledgebase.service; docker ps --filter name=kb-qdrant --format "{{.Names}} {{.Status}}"; systemctl is-active frigate 2>/dev/null || docker ps --filter name=frigate --format "{{.Names}} {{.Status}}"'

Expected: knowledgebase active, kb-qdrant Up, Frigate healthy — the vectorizer touched none of them.

Self-review notes

Spec coverage: Qdrant container (Task 1), local bge embeddings with query/passage prefixing (Task 3), chunker reusing KB sections with images+metadata, token-bounded (Task 2), index with idempotent per-manual replace + deterministic IDs (Task 4), semantic search with folder/manual filter (Task 4/5), CLI + full index + real quality proof (Task 5), read-only over the KB (verified Task 5 Step 4).

Deliberate deviations: git commit steps omitted (user’s rule). CPU torch chosen over GPU to avoid the CUDA-13/driver-550 trap and OCR/Frigate GPU contention (design allowed GPU-when-free; CPU is the robust default at this scale). Unit tests are on the pure chunker; embed/index/search are proven by real integration verification steps (they need the model + Qdrant, not mockable meaningfully).

Graph View

Backlinks