mba/chunking_strategies_evaluation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(admin): add embedding model role and threshold APIs

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Mahdi Bazrafshan
2026-08-10 14:13:07 +03:30
parent 16c918538b
commit aa5838fadc
2 changed files with 275 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
src/admin/routes.py
View File

@@ -1,12 +1,30 @@
"""Admin API routes — system health, Qdrant management, chunk preview, questions, cost."""
import os
from fastapi import APIRouter, UploadFile, File
from fastapi import APIRouter, HTTPException, UploadFile, File
from pydantic import BaseModel, Field
from src.admin import service
router = APIRouter(prefix="/admin")
class ActiveEmbeddingModelBody(BaseModel):
id: str = Field(..., description="Registry Embedding Model id")
class RoleEmbeddingModelBody(BaseModel):
id: str = Field(..., description="Registry Embedding Model id for Boundary or Corpus role")
class SemanticThresholdBody(BaseModel):
semantic_threshold: float = Field(
...,
gt=0.0,
le=1.0,
description="Cosine similarity cutoff for Semantic Boundary Detection",
)
# ── Health ──────────────────────────────────────────────────
@router.get("/health")
@@ -15,6 +33,50 @@ async def health_check():
return service.get_health()
# ── Embedding Models ────────────────────────────────────────
@router.get("/embedding-models")
async def list_embedding_models():
"""List Embedding Model Registry and Boundary/Corpus defaults."""
return service.list_embedding_models()
@router.put("/embedding-models/active")
async def set_active_embedding_model(body: ActiveEmbeddingModelBody):
"""Legacy: set Default Corpus Embedding Model."""
result = service.set_corpus_embedding_model(body.id)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return {**result, "active_id": result.get("corpus_id")}
@router.put("/embedding-models/corpus")
async def set_corpus_embedding_model(body: RoleEmbeddingModelBody):
"""Set Default Corpus Embedding Model (storage + query)."""
result = service.set_corpus_embedding_model(body.id)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return result
@router.put("/embedding-models/boundary")
async def set_boundary_embedding_model(body: RoleEmbeddingModelBody):
"""Set Default Boundary Embedding Model (semantic cuts)."""
result = service.set_boundary_embedding_model(body.id)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return result
@router.put("/embedding-models/{model_id}/semantic-threshold")
async def update_semantic_threshold(model_id: str, body: SemanticThresholdBody):
"""Set per-Embedding-Model semantic_threshold (Admin override in SQLite)."""
result = service.update_semantic_threshold(model_id, body.semantic_threshold)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return result
# ── Qdrant Collections ─────────────────────────────────────
@router.get("/qdrant/collections")

234
src/admin/service.py
View File

@@ -24,7 +24,12 @@ QUESTIONS_DIR = PROJECT_ROOT / "files"
# ── Health ──────────────────────────────────────────────────
def get_health() -> dict[str, Any]:
"""Check server, Qdrant, and SQLite status."""
"""Check server, Qdrant, SQLite, OpenAI, and Ollama status."""
from src.chunking.embedding import (
get_boundary_embedding_model,
get_corpus_embedding_model,
)
result: dict[str, Any] = {"status": "ok"}
# Check Qdrant
@@ -51,57 +56,228 @@ def get_health() -> dict[str, Any]:
# Check OpenAI
try:
client = get_openai_client()
# Just check the client exists; don't make a real API call
get_openai_client()
result["openai_configured"] = bool(settings.openai_api_key)
except Exception:
result["openai_configured"] = False
try:
corpus = get_corpus_embedding_model()
boundary = get_boundary_embedding_model()
result["corpus_embedding_model"] = {
"id": corpus.id,
"provider": corpus.provider.value,
"model_name": corpus.model_name,
"dimension": corpus.dimension,
"display_name": corpus.display_name,
}
result["boundary_embedding_model"] = {
"id": boundary.id,
"provider": boundary.provider.value,
"model_name": boundary.model_name,
"dimension": boundary.dimension,
"display_name": boundary.display_name,
}
# Legacy alias for older Dashboard code
result["active_embedding_model"] = result["corpus_embedding_model"]
except Exception as exc:
result["corpus_embedding_model"] = None
result["boundary_embedding_model"] = None
result["active_embedding_model"] = None
result["active_embedding_error"] = str(exc)
result["ollama_base_url"] = settings.ollama_base_url
try:
import urllib.request
url = settings.ollama_base_url.rstrip("/") + "/api/tags"
with urllib.request.urlopen(url, timeout=2) as resp:
result["ollama_reachable"] = resp.status == 200
except Exception as exc:
result["ollama_reachable"] = False
result["ollama_error"] = str(exc)
return result
# ── Embedding Models ────────────────────────────────────────
def list_embedding_models() -> dict[str, Any]:
"""List registry entries and Boundary/Corpus defaults."""
from src.chunking.embedding import (
get_boundary_embedding_model,
get_corpus_embedding_model,
)
from src.chunking.embedding_models import get_semantic_threshold, list_models
corpus = get_corpus_embedding_model()
boundary = get_boundary_embedding_model()
models = []
for m in list_models():
models.append({
"id": m.id,
"provider": m.provider.value,
"model_name": m.model_name,
"dimension": m.dimension,
"display_name": m.display_name,
"task_prefixes": m.task_prefixes,
"default_semantic_threshold": m.default_semantic_threshold,
"semantic_threshold": get_semantic_threshold(m.id),
"is_corpus_default": m.id == corpus.id,
"is_boundary_default": m.id == boundary.id,
"is_active": m.id == corpus.id, # legacy
})
return {
"corpus_id": corpus.id,
"boundary_id": boundary.id,
"active_id": corpus.id, # legacy
"models": models,
"ollama_base_url": settings.ollama_base_url,
}
def set_active_embedding_model(model_id: str) -> dict[str, Any]:
"""Legacy: set Corpus default."""
return set_corpus_embedding_model(model_id)
def set_corpus_embedding_model(model_id: str) -> dict[str, Any]:
"""Switch the Default Corpus Embedding Model."""
from src.chunking.embedding import set_corpus_embedding_model as set_corpus
try:
model = set_corpus(model_id)
except KeyError as exc:
return {"error": str(exc)}
return {
"corpus_id": model.id,
"model": {
"id": model.id,
"provider": model.provider.value,
"model_name": model.model_name,
"dimension": model.dimension,
"display_name": model.display_name,
},
}
def set_boundary_embedding_model(model_id: str) -> dict[str, Any]:
"""Switch the Default Boundary Embedding Model."""
from src.chunking.embedding import set_boundary_embedding_model as set_boundary
try:
model = set_boundary(model_id)
except KeyError as exc:
return {"error": str(exc)}
return {
"boundary_id": model.id,
"model": {
"id": model.id,
"provider": model.provider.value,
"model_name": model.model_name,
"dimension": model.dimension,
"display_name": model.display_name,
},
}
def update_semantic_threshold(model_id: str, threshold: float) -> dict[str, Any]:
"""Persist Admin override of semantic_threshold for a registry Embedding Model."""
from src.chunking.embedding_models import (
get_model,
get_semantic_threshold,
set_semantic_threshold,
)
try:
model = get_model(model_id)
value = set_semantic_threshold(model_id, threshold)
except KeyError as exc:
return {"error": str(exc)}
except ValueError as exc:
return {"error": str(exc)}
return {
"id": model.id,
"semantic_threshold": value,
"default_semantic_threshold": model.default_semantic_threshold,
"effective": get_semantic_threshold(model.id),
}
# ── Qdrant Collections ─────────────────────────────────────
def list_qdrant_collections() -> dict[str, Any]:
"""List all Qdrant collections with their point counts."""
"""List all Qdrant collections with point counts and Embedding Model labels."""
from src.chunking.embedding import get_active_embedding_model
client = get_qdrant_client()
collections_data = client.get_collections().collections
active = get_active_embedding_model()
result = []
for col in collections_data:
meta = qdrant_store.parse_collection_meta(col.name)
try:
info = client.get_collection(collection_name=col.name)
result.append({
"name": col.name,
"points_count": info.points_count or 0,
})
points = info.points_count or 0
except Exception as exc:
result.append({
"name": col.name,
"points_count": None,
"error": str(exc),
})
points = None
err = str(exc)
else:
err = None
return {"collections": result}
entry = {
"name": col.name,
"points_count": points,
"strategy": meta.get("strategy"),
"embedding_model_id": meta.get("embedding_model_id"),
"is_legacy": meta.get("is_legacy", False),
"is_active_corpus": meta.get("embedding_model_id") == active.id,
}
if err:
entry["error"] = err
result.append(entry)
# Active Model Corpus first
result.sort(key=lambda c: (not c.get("is_active_corpus", False), c["name"]))
return {
"collections": result,
"active_embedding_model_id": active.id,
}
def create_qdrant_collection(collection_name: str) -> dict[str, Any]:
"""Create a new Qdrant collection."""
"""Create a new Qdrant collection using Active Embedding Model dimension."""
from src.chunking.embedding import get_active_embedding_model
client = get_qdrant_client()
active = get_active_embedding_model()
existing = [c.name for c in client.get_collections().collections]
if collection_name in existing:
return {"created": False, "message": f"Collection '{collection_name}' already exists"}
meta = qdrant_store.parse_collection_meta(collection_name)
dimension = active.dimension
if meta.get("embedding_model_id"):
try:
from src.chunking.embedding_models import get_model
dimension = get_model(meta["embedding_model_id"]).dimension
except KeyError:
pass
client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=qdrant_store.VECTOR_DIMENSION,
size=dimension,
distance=Distance.COSINE,
),
)
logger.info("Created Qdrant collection: %s", collection_name)
return {"created": True, "collection": collection_name}
logger.info("Created Qdrant collection: %s (dim=%d)", collection_name, dimension)
return {"created": True, "collection": collection_name, "dimension": dimension}
def delete_qdrant_collection(collection_name: str) -> dict[str, Any]:
@@ -134,8 +310,11 @@ def wipe_qdrant_collection_points(collection_name: str) -> dict[str, Any]:
# ── Chunk Preview ───────────────────────────────────────────
def preview_chunks(doc_id: str, strategy: str | None = None) -> dict[str, Any]:
"""Preview chunks for a document. Uses Qdrant scroll to fetch chunks with payload."""
"""Preview chunks for a document from the Active Embedding Model's corpus."""
from src.chunking.embedding import get_active_embedding_model
client = get_qdrant_client()
active = get_active_embedding_model()
from qdrant_client.models import Filter, FieldCondition, MatchValue
@@ -155,12 +334,12 @@ def preview_chunks(doc_id: str, strategy: str | None = None) -> dict[str, Any]:
results = {}
for strat_name in strategies_to_search:
col_name = f"{strat_name}_collection"
col_name = qdrant_store.collection_name(strat_name, active.id)
try:
existing = [c.name for c in client.get_collections().collections]
if col_name not in existing:
results[strat_name] = {"chunks": [], "count": 0}
results[strat_name] = {"chunks": [], "count": 0, "collection": col_name}
continue
scroll_filter = Filter(
@@ -188,12 +367,21 @@ def preview_chunks(doc_id: str, strategy: str | None = None) -> dict[str, Any]:
# Sort by chunk_index
chunks.sort(key=lambda c: c.get("chunk_index") or 0)
results[strat_name] = {"chunks": chunks, "count": len(chunks)}
results[strat_name] = {
"chunks": chunks,
"count": len(chunks),
"collection": col_name,
}
except Exception as exc:
results[strat_name] = {"error": str(exc), "chunks": [], "count": 0}
return {"document_id": doc_id, "filename": doc_name, "strategies": results}
return {
"document_id": doc_id,
"filename": doc_name,
"embedding_model_id": active.id,
"strategies": results,
}
# ── Questions Dataset ───────────────────────────────────────