Search Business Procedures Locally With FastEmbed

anyio==4.15.1
certifi==2026.7.22
charset-normalizer==3.5.2
click==8.5.0
colorama==0.4.6
fastembed==0.8.1
filelock==4.0.9
flatbuffers==25.12.19
fsspec==2026.9.0
h11==0.16.0
hf-xet==1.6.0
httpcore==1.0.9
httpx==0.28.1
huggingface_hub==1.33.0
idna==3.20
loguru==0.7.3
mmh3==5.3.1
numpy==2.5.3
onnxruntime==1.30.0
packaging==26.3
pillow==12.3.0
protobuf==7.36.2
py_rust_stemmers==0.1.8
PyYAML==6.0.3
requests==2.34.2
tokenizers==0.23.2
tqdm==4.70.1
typing_extensions==4.16.0
urllib3==2.8.0
win32_setctime==1.2.0

"""Download the exact public ONNX model snapshot used by the article."""
import hashlib
import json
import os
from pathlib import Path

os.environ["HF_HUB_DISABLE_IMPLICIT_TOKEN"] = "1"
os.environ["HF_HUB_DISABLE_XET"] = "1"
from huggingface_hub import snapshot_download

REPOSITORY = "Qdrant/bge-small-en-v1.5-onnx-Q"
REVISION = "aa8f8b060edb00e03bfdd08813a2949946c8ba55"
FILES = ["README.md", "config.json", "model_optimized.onnx", "ort_config.json",
         "special_tokens_map.json", "tokenizer.json", "tokenizer_config.json", "vocab.txt"]


if __name__ == "__main__":
    folder = Path("model")
    snapshot_download(repo_id=REPOSITORY, revision=REVISION,
                      local_dir=folder, allow_patterns=FILES, token=False, max_workers=1)
    manifest = {"repository": REPOSITORY, "revision": REVISION, "files": {}}
    for name in FILES:
        data = (folder / name).read_bytes()
        manifest["files"][name] = {"bytes": len(data),
                                   "sha256": hashlib.sha256(data).hexdigest()}
    Path("model-manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    print(json.dumps({"revision": REVISION, "files": len(FILES),
                      "bytes": sum(f["bytes"] for f in manifest["files"].values())}))

[
  {"id":"kickoff","title":"Start a website project","text":"At kickoff, collect the client's goals, intended audience and required pages. Assign one client contact to consolidate feedback. Record scope and decision owners in the project brief."},
  {"id":"copy","title":"Collect page copy","text":"Ask the client to supply final page text in the shared content folder. Mark missing sections clearly. Placeholder copy must be replaced or explicitly approved before launch."},
  {"id":"scope","title":"Request work outside the agreed scope","text":"Log requests for additional pages or features as a change request. The project lead estimates the work and obtains written client approval before changing the delivery plan."},
  {"id":"logo","title":"Request usable brand assets","text":"Request the client's logo as SVG or another original vector file, along with brand colors and fonts. A small screenshot of the logo is not a suitable production source."},
  {"id":"revisions","title":"Consolidate design feedback","text":"Collect comments into one numbered revision list from the designated client contact. Resolve contradictory requests before the designer starts the next revision round."},
  {"id":"launch","title":"Approve public launch","text":"Do not publish the website until the named client approver has given written signoff on the launch checklist. A preview link or a casual positive comment does not replace that approval."},
  {"id":"access","title":"Share domain access","text":"Ask the domain owner to invite the project lead with the minimum required permissions. Do not ask the client to email their registrar password. Confirm the intended domain before any DNS change."},
  {"id":"invoice","title":"Handle an invoice status question","text":"Check the invoice number against the billing system and confirm whether payment is recorded. Route discrepancies to the office contact. Do not infer payment from a customer's message alone."},
  {"id":"accessibility","title":"Review keyboard navigation","text":"Before launch, navigate links, buttons and forms using the keyboard. Check visible focus and descriptive field labels. Record issues for repair; this check alone is not a complete accessibility audit."},
  {"id":"backup","title":"Prepare a recovery copy","text":"Before a site update, save a current backup of the website files and database. Record its location and the restoration steps. A successful download alone does not demonstrate that restoration works."},
  {"id":"meeting","title":"Record a client meeting","text":"After a client meeting, save decisions, named owners and due dates in the project notes. Send the agreed action list through the existing client communication channel."},
  {"id":"images","title":"Approve supplied campaign photos","text":"Confirm the client has permission to use supplied photos in the intended campaign. Record the source and usage restrictions. Finding an image on a public website does not establish reuse permission."}
]

"""Rank a small fictional procedure library with a local embedding model."""
import argparse
import json
from pathlib import Path

import numpy as np
from fastembed import TextEmbedding
from tokenizers import Tokenizer

MODEL = "BAAI/bge-small-en-v1.5"
QUERY_PREFIX = "Represent this sentence for searching relevant passages: "


class ProcedureSearch:
    def __init__(self, documents, model_dir="model"):
        if not documents or len({d["id"] for d in documents}) != len(documents):
            raise ValueError("Provide nonempty documents with unique IDs")
        self.documents = documents
        self.tokenizer = Tokenizer.from_file(str(Path(model_dir) / "tokenizer.json"))
        self.tokenizer.no_truncation()
        self.tokenizer.no_padding()
        self.model = TextEmbedding(MODEL, specific_model_path=str(model_dir),
                                   local_files_only=True,
                                   providers=["CPUExecutionProvider"], threads=2)
        texts = [d["title"] + "\n" + d["text"] for d in documents]
        self.check_lengths(texts)
        self.vectors = np.array(list(self.model.embed(texts)), dtype=np.float32)

    def check_lengths(self, texts):
        for text in texts:
            if len(self.tokenizer.encode(text).ids) > 512:
                raise ValueError("Text exceeds 512 tokens; split it before embedding")

    def search(self, query, k=3):
        if not query.strip():
            raise ValueError("Query must not be blank")
        if not 1 <= k <= len(self.documents):
            raise ValueError("k must be between 1 and the document count")
        text = QUERY_PREFIX + query.strip()
        self.check_lengths([text])
        vector = np.array(next(self.model.embed([text])), dtype=np.float32)
        # This model's FastEmbed implementation normalizes both vectors.
        scores = self.vectors @ vector
        order = sorted(range(len(scores)),
                       key=lambda i: (-float(scores[i]), self.documents[i]["id"]))[:k]
        return [{**self.documents[i], "similarity": round(float(scores[i]), 6)}
                for i in order]


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("query")
    parser.add_argument("--documents", default="procedures.json")
    parser.add_argument("--model-dir", default="model")
    parser.add_argument("--top", type=int, default=3)
    args = parser.parse_args()
    documents = json.loads(Path(args.documents).read_text("utf-8"))
    engine = ProcedureSearch(documents, args.model_dir)
    print(json.dumps({"notice": "Ranked references, not an answer or approval.",
                      "results": engine.search(args.query, args.top)}, indent=2))

[
  {"query":"The customer wants three extra pages. What is the process?","expected":"scope"},
  {"query":"The logo looks blurry when enlarged. What file should I request?","expected":"logo"},
  {"query":"Can I make the site public without written signoff?","expected":"launch"},
  {"query":"How should the customer give us control of their domain?","expected":"access"},
  {"query":"Two stakeholders asked for opposite design changes. What now?","expected":"revisions"},
  {"query":"How do we recover if a website update goes wrong?","expected":"backup"},
  {"query":"A user cannot use a mouse. What should we check?","expected":"accessibility"},
  {"query":"The client says they already paid. Can we mark the bill settled?","expected":"invoice"},
  {"query":"When are staff wages paid?","expected":null},
  {"query":"How long is the hardware warranty?","expected":null}
]

"""Exercise the actual downloaded model and record all fixture rankings."""
import hashlib
import json
import os
import platform
import socket
from importlib.metadata import version
from pathlib import Path
from unittest.mock import patch

os.environ["HF_HUB_OFFLINE"] = "1"
from search_sops import ProcedureSearch, QUERY_PREFIX
import numpy as np


def must_reject(callback):
    try:
        callback()
    except ValueError:
        return
    raise AssertionError("Expected input rejection")


if __name__ == "__main__":
    documents = json.loads(Path("procedures.json").read_text("utf-8"))
    queries = json.loads(Path("queries.json").read_text("utf-8"))
    source_hash = hashlib.sha256(Path("procedures.json").read_bytes()).hexdigest()
    manifest = json.loads(Path("model-manifest.json").read_text("utf-8"))
    for name, info in manifest["files"].items():
        assert hashlib.sha256((Path("model") / name).read_bytes()).hexdigest() == info["sha256"]
    checks = ["downloaded model file hashes match manifest"]
    # This catches Python socket connections; it is not an OS network sandbox.
    with patch.object(socket.socket, "connect", side_effect=AssertionError("Network attempted")) as connect, \
         patch.object(socket.socket, "connect_ex", side_effect=AssertionError("Network attempted")) as connect_ex:
        engine = ProcedureSearch(documents)
        assert engine.model.model.model.get_providers() == ["CPUExecutionProvider"]
        checks.append("actual ONNX session uses CPUExecutionProvider")
        assert engine.vectors.shape == (12, 384) and np.isfinite(engine.vectors).all()
        checks.append("twelve finite 384-dimensional document vectors")
        assert np.allclose(np.linalg.norm(engine.vectors, axis=1), 1, atol=1e-5)
        checks.append("document vectors are unit normalized")
        query_vector = next(engine.model.embed([QUERY_PREFIX + queries[0]["query"]]))
        assert np.isclose(np.linalg.norm(query_vector), 1, atol=1e-5)
        checks.append("query vector is unit normalized")
        ranks = []
        for case in queries:
            hits = engine.search(case["query"])
            assert len(hits) == 3
            assert [h["similarity"] for h in hits] == sorted([h["similarity"] for h in hits], reverse=True)
            for hit in hits:
                source = next(d for d in documents if d["id"] == hit["id"])
                assert all(hit[k] == source[k] for k in ["id", "title", "text"])
            ranks.append({**case, "hits": hits,
                          "rank": next((i + 1 for i,h in enumerate(hits) if h["id"] == case["expected"]), None)})
        checks.append("all ten queries return three sorted source-preserving results")
        for query in ["", "   ", "word " * 600]:
            must_reject(lambda q=query: engine.search(q))
        checks.append("blank and oversized queries rejected")
        for k in [0, 13]:
            must_reject(lambda n=k: engine.search("project", n))
        checks.append("out-of-range result limits rejected")
        must_reject(lambda: engine.check_lengths(["word " * 600]))
        checks.append("oversized document rejected before truncation")
        must_reject(lambda: ProcedureSearch([]))
        must_reject(lambda: ProcedureSearch([documents[0], documents[0]]))
        checks.append("empty corpus and duplicate document IDs rejected")
        assert connect.call_count == 0 and connect_ex.call_count == 0
        checks.append("no Python socket connection during model loading and search")
    assert hashlib.sha256(Path("procedures.json").read_bytes()).hexdigest() == source_hash
    checks.append("source corpus unchanged")
    known = [r for r in ranks if r["expected"] is not None]
    evidence = {
        "python": platform.python_version(), "platform": platform.platform(),
        "versions": {p:version(p) for p in ["fastembed", "onnxruntime", "numpy", "tokenizers", "huggingface-hub"]},
        "model_repository": manifest["repository"], "model_revision": manifest["revision"],
        "corpus_sha256": source_hash, "checks_passed": len(checks), "checks": checks,
        "known_queries": len(known), "top1_hits": sum(r["rank"] == 1 for r in known),
        "top3_hits": sum(r["rank"] is not None for r in known),
        "unsupported_queries": len(ranks) - len(known), "results": ranks,
        "limits": "Small authored fictional fixture set, selected before the first ranking run; not a held-out business benchmark. No generative model, production server, privacy certification or load test. Socket guard is Python-level only."
    }
    Path("test-evidence.json").write_text(json.dumps(evidence, indent=2), encoding="utf-8")
    print(json.dumps({k:evidence[k] for k in ["checks_passed", "known_queries", "top1_hits", "top3_hits", "unsupported_queries"]}))
    for result in ranks:
        print(json.dumps({"query":result["query"], "expected":result["expected"], "rank":result["rank"],
                          "top3":[[h["id"],h["similarity"]] for h in result["hits"]]}))