mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-13 06:45:26 -04:00
* feat(router): make KNN a first-class classifier with a persisted, curated corpus
Add `classifier: knn` — similarity-weighted voting over labelled
example prompts. Unlike score/colbert it needs no classifier model:
label knowledge lives in a corpus seeded and curated through the
admin API, so routing decisions are deterministic, auditable, and
grounded in graded experience rather than a model's opinion.
Epistemic gate: corpus entries below knn.similarity_threshold cannot
vote; when none clears it the classifier activates no labels and the
router uses the fallback — a prompt unlike all labelled experience is
treated as undecidable, not guessed. Decisions record
nearest_similarity (also on fallback rows) so admins can see how far
the nearest labelled experience was; the Routing tab explains
out-of-corpus fallbacks and shows per-label corpus counts.
Persistence: one JSONL file per router under
<data path>/router-corpus (text, labels, vector, embedder
fingerprint). The file is the source of truth; the local-store index
is rebuilt from it at classifier build time and stays a pure
in-memory index. Entries recorded under a different embedding model
re-embed on load. Also corrects the docs' false claim that
local-store collections persist — the embedding cache never survived
restarts (and still doesn't); the corpus does.
Corpus input is API-only by design (entries may contain example user
content): POST /api/router/{name}/corpus seeds (labels validated
against declared policies, embedded server-side, indexed
immediately), GET .../corpus/stats inspects — label counts only,
entry texts are never returned by any surface — DELETE .../corpus
wipes. Admin-gated like the sibling router endpoints, and exposed as
MCP tools (seed_router_corpus / get_router_corpus_stats /
clear_router_corpus) in both the httpapi and inproc clients with
coverage-test route mappings.
Plumbing: VectorStore gains SearchK (top-K was hardcoded to 1);
local-store gets InsertBatch/Delete as optional fast paths;
RouterConfig gains a knn block (embedding_model, k,
similarity_threshold, vote_threshold, store_name) with meta-registry
fields; the classifier dropdown now offers knn and the
previously-missing colbert; embedding_cache is ignored (with a
warning) for knn — it IS an embedding-KNN lookup; the stale
/api/instructions intelligent-routing entry is rewritten (it
described a classifier that no longer exists); swagger regenerated.
Tests: KNN vote/gate specs with hand-computed vote shares, corpus
manager suite (restart reload without re-embedding, fingerprint
re-embed, dedupe, hostile store names), middleware specs (corpus
routing, gate fallback, config validation, cache-wrap refusal),
corpus endpoint specs pinning the texts-never-returned contract, MCP
catalog + route-mapping gates, and a Playwright spec for corpus
stats and the out-of-corpus decision detail.
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* feat(router): name consulted corpus neighbours in knn decisions
Every knn decision (decision log rows and the /api/router/decide
response) now carries neighbors: the K retrieved corpus entries by
descending similarity - including ones below the epistemic gate, which
is what makes fallback decisions diagnosable - each as {id, similarity,
labels}. The id is the entry's content hash (first 8 bytes of the
SHA-256 of its text, hex): stable across reseeds and re-embeds, and
text-free, so an external platform that seeded the corpus can recompute
text->id on its own copy and bucket decisions by corpus region (per-
region reliability accounting) without corpus text ever leaving the
server. A corrupt index payload surfaces as an id-less neighbour at a
real similarity instead of disappearing.
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* refactor(router): deduplicate knn plumbing and cut corpus hot-path waste
Post-review cleanup of the knn-first-class-router branch; no behaviour
changes on the API surface.
Reuse/altitude:
- RouterKNNConfig.ResolvedStoreName is now the single source of the
router-corpus-<name> default (was hand-derived in four files).
- corpus.ResolveKNNRouter + corpus.Seed carry the shared model
resolution and seed validation; the REST endpoints and the assistant
MCP client are thin transport adapters over them, with sentinel
errors mapped to HTTP statuses at the echo boundary.
- middleware.NewClassifierDeps assembles the classifier dependency set
once for all five entry points (OpenAI, Anthropic, realtime, decide,
corpus) instead of five hand-copied literals.
- router.AllClassifiers feeds both the status endpoint and the
unknown-classifier error, ending the classifier-list drift.
- Per-classifier requirements moved out of validateRouterPolicies into
their buildClassifier arms; the knn arm owns its embedding_cache
opt-out instead of a name-check in the shared wrap tail.
- adminOnly replaces four inline copies of the admin gate in the
middleware routes.
- localVectorStore.Search delegates to SearchK (identical traces).
Efficiency:
- Manager.Add embeds outside the manager mutex and appends to the
JSONL file (O(new) instead of O(corpus) rewrite); a torn tail from a
crash mid-append is tolerated on read and repaired on next write.
- Stats memoises per store keyed on the file's stat fingerprint and no
longer takes the manager mutex, so the 5s status poll stops parsing
vector-laden JSONL and stops blocking behind seeds.
- KNN Classify decodes each neighbour payload once (was twice) and
builds refs and votes in a single pass with one fallback return.
- Corpus file writes fsync before rename/close.
- The corpus manager is built eagerly in newApplication (sync.Once
dropped); test helper dead branch removed.
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* feat(router): bind knn corpus vectors to an embedder fingerprint and fail closed on mismatch
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* chore(mcp): align corpus tool prompts and the mutating-tool safety list
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* feat(proto,backend): report embedding shape from the llama-cpp backend
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* feat(embeddings): Go-side pooling — mean/last/decayed_mean with half-life
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* feat(embeddings): accept chat messages[] and per-request pooling on /v1/embeddings
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* chore(middleware): name the failing fields when post-merge validation 400s
An intermittent post-merge validation failure surfaced as an opaque 400
during integration (pooling scheme mismatch that no client had sent).
Log the model, the request's pooling override, and the merged config's
pooling fields at the failure point so the next occurrence identifies
whether the request or the stored config carried the bad value.
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* fix(embeddings): scheme override must not inherit the config's half-life
A model config defaulting to decayed_mean pooling carries
pooling_half_life_tokens; a request overriding the scheme to mean/last
without its own half-life inherited that value, and post-merge
validation rejected the pair the server itself had assembled. Zero the
inherited half-life when the overridden scheme is not decayed_mean; a
request that explicitly pairs a half-life with a non-decayed scheme
still 400s.
Assisted-by: Claude:claude-fable-5 [Claude Code]
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* fix embedding pooling validation and router bounds
Declare backend embedding layouts and reject incompatible pooling modes. Reset local-store dimensions after a full clear, validate KNN thresholds, and add real backend and store integration coverage.
Assisted-by: Codex:gpt-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>
* ci: run local-store integration tests
Build and install the local-store backend in the Linux test job, then run the existing store integration suite so new specs are discovered automatically.
Assisted-by: Codex:gpt-5
Signed-off-by: Richard Palethorpe <io@richiejp.com>
---------
Signed-off-by: Richard Palethorpe <io@richiejp.com>
316 lines
12 KiB
Python
316 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""gRPC server for the insightface face recognition backend.
|
|
|
|
Implements Health / LoadModel / Status plus the face-specific methods:
|
|
Embedding, Detect, FaceVerify, FaceAnalyze. The heavy lifting is
|
|
delegated to engines.py — this file is just the gRPC plumbing.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import os
|
|
import signal
|
|
import sys
|
|
import time
|
|
from concurrent import futures
|
|
from io import BytesIO
|
|
|
|
import backend_pb2
|
|
import backend_pb2_grpc
|
|
import cv2
|
|
import grpc
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common"))
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "common"))
|
|
from grpc_auth import get_auth_interceptors # noqa: E402
|
|
|
|
from engines import FaceEngine, build_engine # noqa: E402
|
|
|
|
_ONE_DAY = 60 * 60 * 24
|
|
MAX_WORKERS = int(os.environ.get("PYTHON_GRPC_MAX_WORKERS", "1"))
|
|
|
|
# Default cosine-distance threshold for "same person" on buffalo_l
|
|
# ArcFace R50. Clients can override per-request; clients using SFace
|
|
# should pass threshold≈0.4 since the distance distribution is wider.
|
|
DEFAULT_VERIFY_THRESHOLD = 0.35
|
|
|
|
|
|
def _decode_image(src: str) -> np.ndarray | None:
|
|
"""Decode a base64-encoded image into an OpenCV BGR numpy array."""
|
|
if not src:
|
|
return None
|
|
try:
|
|
data = base64.b64decode(src, validate=False)
|
|
except Exception:
|
|
return None
|
|
arr = np.frombuffer(data, dtype=np.uint8)
|
|
if arr.size == 0:
|
|
return None
|
|
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
|
|
return img
|
|
|
|
|
|
def _parse_options(raw: list[str]) -> dict[str, str]:
|
|
out: dict[str, str] = {}
|
|
for entry in raw:
|
|
if ":" not in entry:
|
|
continue
|
|
k, v = entry.split(":", 1)
|
|
out[k.strip()] = v.strip()
|
|
return out
|
|
|
|
|
|
class BackendServicer(backend_pb2_grpc.BackendServicer):
|
|
def __init__(self) -> None:
|
|
self.engine: FaceEngine | None = None
|
|
self.engine_name: str = ""
|
|
self.model_name: str = ""
|
|
self.verify_threshold: float = DEFAULT_VERIFY_THRESHOLD
|
|
|
|
def Health(self, request, context):
|
|
return backend_pb2.Reply(message=bytes("OK", "utf-8"))
|
|
|
|
def LoadModel(self, request, context):
|
|
options = _parse_options(list(request.Options))
|
|
# Surface LocalAI's models directory (ModelPath) so engines can
|
|
# anchor relative paths — OnnxDirectEngine's detector_onnx /
|
|
# recognizer_onnx point at gallery-managed files that LocalAI
|
|
# dropped there, and InsightFaceEngine auto-downloads its packs
|
|
# into that same directory alongside every other managed model.
|
|
# Private key to avoid clashing with user-provided options.
|
|
if request.ModelPath:
|
|
options["_model_dir"] = request.ModelPath
|
|
|
|
engine_name = options.get("engine", "insightface")
|
|
try:
|
|
self.engine = build_engine(engine_name)
|
|
self.engine.prepare(options)
|
|
except Exception as err: # pragma: no cover - exercised via e2e
|
|
return backend_pb2.Result(success=False, message=f"Failed to load face engine: {err}")
|
|
|
|
self.engine_name = engine_name
|
|
self.model_name = request.Model or options.get("model_pack", "")
|
|
if "verify_threshold" in options:
|
|
try:
|
|
self.verify_threshold = float(options["verify_threshold"])
|
|
except ValueError:
|
|
pass
|
|
print(f"[insightface] engine={engine_name} model={self.model_name} loaded", file=sys.stderr)
|
|
return backend_pb2.Result(success=True, message="Model loaded successfully")
|
|
|
|
def Status(self, request, context):
|
|
state = (
|
|
backend_pb2.StatusResponse.READY
|
|
if self.engine is not None
|
|
else backend_pb2.StatusResponse.UNINITIALIZED
|
|
)
|
|
return backend_pb2.StatusResponse(state=state)
|
|
|
|
def Embedding(self, request, context):
|
|
if self.engine is None:
|
|
context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
|
|
context.set_details("face model not loaded")
|
|
return backend_pb2.EmbeddingResult()
|
|
if not request.Images:
|
|
context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
|
|
context.set_details("Embedding requires Images[0] to be a base64 image")
|
|
return backend_pb2.EmbeddingResult()
|
|
|
|
img = _decode_image(request.Images[0])
|
|
if img is None:
|
|
context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
|
|
context.set_details("failed to decode image")
|
|
return backend_pb2.EmbeddingResult()
|
|
|
|
vec = self.engine.embed(img)
|
|
if vec is None:
|
|
context.set_code(grpc.StatusCode.NOT_FOUND)
|
|
context.set_details("no face detected")
|
|
return backend_pb2.EmbeddingResult()
|
|
return backend_pb2.EmbeddingResult(
|
|
embeddings=[float(x) for x in vec],
|
|
layout=backend_pb2.EMBEDDING_LAYOUT_FINAL,
|
|
)
|
|
|
|
def Detect(self, request, context):
|
|
if self.engine is None:
|
|
return backend_pb2.DetectResponse()
|
|
img = _decode_image(request.src)
|
|
if img is None:
|
|
return backend_pb2.DetectResponse()
|
|
detections = []
|
|
for d in self.engine.detect(img):
|
|
x1, y1, x2, y2 = d.bbox
|
|
detections.append(
|
|
backend_pb2.Detection(
|
|
x=float(x1),
|
|
y=float(y1),
|
|
width=float(x2 - x1),
|
|
height=float(y2 - y1),
|
|
confidence=float(d.score),
|
|
class_name="face",
|
|
)
|
|
)
|
|
return backend_pb2.DetectResponse(Detections=detections)
|
|
|
|
def FaceVerify(self, request, context):
|
|
if self.engine is None:
|
|
context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
|
|
context.set_details("face model not loaded")
|
|
return backend_pb2.FaceVerifyResponse()
|
|
|
|
img1 = _decode_image(request.img1)
|
|
img2 = _decode_image(request.img2)
|
|
if img1 is None or img2 is None:
|
|
context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
|
|
context.set_details("failed to decode one or both images")
|
|
return backend_pb2.FaceVerifyResponse()
|
|
|
|
threshold = request.threshold if request.threshold > 0 else self.verify_threshold
|
|
|
|
start = time.time()
|
|
e1 = self.engine.embed(img1)
|
|
e2 = self.engine.embed(img2)
|
|
if e1 is None or e2 is None:
|
|
context.set_code(grpc.StatusCode.NOT_FOUND)
|
|
context.set_details("no face detected in one or both images")
|
|
return backend_pb2.FaceVerifyResponse()
|
|
|
|
# Both engines return L2-normalized vectors, so the dot product
|
|
# is the cosine similarity directly.
|
|
sim = float(np.dot(e1, e2))
|
|
distance = 1.0 - sim
|
|
verified = distance < threshold
|
|
confidence = max(0.0, min(100.0, (1.0 - distance / threshold) * 100.0)) if threshold > 0 else 0.0
|
|
|
|
# Detect once per image — region is needed for the response and
|
|
# potentially for the antispoof crop. Returns the highest-score face.
|
|
def _best_detection(img):
|
|
dets = self.engine.detect(img)
|
|
if not dets:
|
|
return None
|
|
return max(dets, key=lambda d: d.score)
|
|
|
|
def _region(det) -> backend_pb2.FacialArea:
|
|
if det is None:
|
|
return backend_pb2.FacialArea()
|
|
x1, y1, x2, y2 = det.bbox
|
|
return backend_pb2.FacialArea(x=x1, y=y1, w=x2 - x1, h=y2 - y1)
|
|
|
|
det1 = _best_detection(img1)
|
|
det2 = _best_detection(img2)
|
|
|
|
img1_is_real = False
|
|
img1_score = 0.0
|
|
img2_is_real = False
|
|
img2_score = 0.0
|
|
if request.anti_spoofing:
|
|
spoof1 = self.engine.antispoof(img1, det1.bbox) if det1 is not None else None
|
|
spoof2 = self.engine.antispoof(img2, det2.bbox) if det2 is not None else None
|
|
if spoof1 is None or spoof2 is None:
|
|
context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
|
|
context.set_details(
|
|
"anti_spoofing requested but no antispoof model is loaded — "
|
|
"install `silent-face-antispoofing` or pick a gallery entry "
|
|
"that bundles MiniFASNet weights"
|
|
)
|
|
return backend_pb2.FaceVerifyResponse()
|
|
img1_is_real, img1_score = spoof1.is_real, spoof1.score
|
|
img2_is_real, img2_score = spoof2.is_real, spoof2.score
|
|
# Failed liveness vetoes verification regardless of similarity.
|
|
if not (img1_is_real and img2_is_real):
|
|
verified = False
|
|
|
|
return backend_pb2.FaceVerifyResponse(
|
|
verified=verified,
|
|
distance=float(distance),
|
|
threshold=float(threshold),
|
|
confidence=float(confidence),
|
|
model=self.model_name or self.engine_name,
|
|
img1_area=_region(det1),
|
|
img2_area=_region(det2),
|
|
processing_time_ms=float((time.time() - start) * 1000.0),
|
|
img1_is_real=img1_is_real,
|
|
img1_antispoof_score=float(img1_score),
|
|
img2_is_real=img2_is_real,
|
|
img2_antispoof_score=float(img2_score),
|
|
)
|
|
|
|
def FaceAnalyze(self, request, context):
|
|
if self.engine is None:
|
|
context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
|
|
context.set_details("face model not loaded")
|
|
return backend_pb2.FaceAnalyzeResponse()
|
|
img = _decode_image(request.img)
|
|
if img is None:
|
|
context.set_code(grpc.StatusCode.INVALID_ARGUMENT)
|
|
context.set_details("failed to decode image")
|
|
return backend_pb2.FaceAnalyzeResponse()
|
|
|
|
faces = []
|
|
for attrs in self.engine.analyze(img):
|
|
x, y, w, h = attrs.region
|
|
fa = backend_pb2.FaceAnalysis(
|
|
region=backend_pb2.FacialArea(x=float(x), y=float(y), w=float(w), h=float(h)),
|
|
face_confidence=float(attrs.face_confidence),
|
|
)
|
|
if attrs.age is not None:
|
|
fa.age = float(attrs.age)
|
|
if attrs.dominant_gender:
|
|
fa.dominant_gender = attrs.dominant_gender
|
|
for k, v in attrs.gender.items():
|
|
fa.gender[k] = float(v)
|
|
if request.anti_spoofing:
|
|
bbox = (float(x), float(y), float(x + w), float(y + h))
|
|
spoof = self.engine.antispoof(img, bbox)
|
|
if spoof is None:
|
|
context.set_code(grpc.StatusCode.FAILED_PRECONDITION)
|
|
context.set_details(
|
|
"anti_spoofing requested but no antispoof model is loaded — "
|
|
"install `silent-face-antispoofing` or pick a gallery entry "
|
|
"that bundles MiniFASNet weights"
|
|
)
|
|
return backend_pb2.FaceAnalyzeResponse()
|
|
fa.is_real = spoof.is_real
|
|
fa.antispoof_score = float(spoof.score)
|
|
faces.append(fa)
|
|
return backend_pb2.FaceAnalyzeResponse(faces=faces)
|
|
|
|
|
|
def serve(address: str) -> None:
|
|
server = grpc.server(
|
|
futures.ThreadPoolExecutor(max_workers=MAX_WORKERS),
|
|
options=[
|
|
("grpc.max_message_length", 50 * 1024 * 1024),
|
|
("grpc.max_send_message_length", 50 * 1024 * 1024),
|
|
("grpc.max_receive_message_length", 50 * 1024 * 1024),
|
|
],
|
|
interceptors=get_auth_interceptors(),
|
|
)
|
|
backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
|
|
server.add_insecure_port(address)
|
|
server.start()
|
|
print("[insightface] Server started. Listening on: " + address, file=sys.stderr)
|
|
|
|
def _stop(sig, frame): # pragma: no cover
|
|
print("[insightface] shutting down")
|
|
server.stop(0)
|
|
sys.exit(0)
|
|
|
|
signal.signal(signal.SIGINT, _stop)
|
|
signal.signal(signal.SIGTERM, _stop)
|
|
|
|
try:
|
|
while True:
|
|
time.sleep(_ONE_DAY)
|
|
except KeyboardInterrupt:
|
|
server.stop(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Run the insightface gRPC server.")
|
|
parser.add_argument("--addr", default="localhost:50051", help="The address to bind the server to.")
|
|
args = parser.parse_args()
|
|
print(f"[insightface] startup: {args}", file=sys.stderr)
|
|
serve(args.addr)
|