feat: add local on-device extraction (/json/feature-extract)

.env.example

@@ -46,6 +46,14 @@ EMBEDDING_DIMENSION=384
 DEFAULT_TOP_K=10
 DATA_DIR=./data
 
+# --- Local on-device extraction (/json/feature-extract) ---
+# All keys are optional; unset values fall back to the defaults shown.
+# GLINER_MODEL=fastino/gliner2-base-v1
+# GLINER_ENABLED=true
+# GLINER_DEVICE=cpu
+# GLINER_MAX_CONCURRENT=2
+# GLINER_MAX_CONTENT_LENGTH=100000
+
 # --- Supabase ---
 SUPABASE_URL=https://your-project.supabase.co
 SUPABASE_ANON_KEY=

Dockerfile

@@ -95,6 +95,11 @@ RUN chmod +x /app/whatsapp-service/server
 
 RUN mkdir -p /app/models && python3 -c "from huggingface_hub import snapshot_download; snapshot_download(repo_id='ibm-granite/granite-embedding-small-english-r2', local_dir='/app/models/bge-384')" && chown -R appuser:appuser /app/models
 
+# Pre-download the local on-device extraction model (used by /json/feature-extract)
+# so first boot does not hit Hugging Face. The GLINER_MODEL env below points the
+# service at this local copy.
+RUN mkdir -p /app/models && python3 -c "from huggingface_hub import snapshot_download; snapshot_download(repo_id='fastino/gliner2-base-v1', local_dir='/app/models/gliner2-base-v1')" && chown -R appuser:appuser /app/models
+
 RUN mkdir -p /app/data /app/logs && \
     chown -R appuser:appuser /app/data /app/logs
 
@@ -105,6 +110,9 @@ USER appuser
 ENV PYTHONPATH=/app
 ENV PYTHONUNBUFFERED=1
 
+# Local copy of the on-device extraction model baked in at build time.
+ENV GLINER_MODEL=/app/models/gliner2-base-v1
+
 # Path to the embedded WhatsApp service binary (started by start.sh as a
 # sibling process). Set to an empty value to disable the WhatsApp gateway.
 # The Go service reads all runtime settings (SUPABASE_URL, SUPABASE_DB_URL,

app/api/server.py

@@ -30,6 +30,7 @@ from app.core.logger import get_logger
 from app.core.redis_client import close_redis, create_redis_client
 from app.core.scripts import load_scripts
 from app.services.embeddings_service import EmbeddingService
+from app.services.gliner_service import gliner_service
 from app.services.scheduler_service import SchedulerService
 from app.services.vector_store_service import VectorStoreService
 from app.utils.http_utils import SharedAsyncClient
@@ -143,6 +144,18 @@ async def lifespan(app: FastAPI):
     except Exception as exc:
         _logger.warning("OCR engine warm-up failed (will lazy-init on first use): %s", exc)
 
+    # Warm the local GLiNER2 model so the /json/no-ai route never pays the
+    # one-time ~5-15s load cost on first request. Loads on the shared thread
+    # pool; failure is non-fatal (the route lazy-loads on first use).
+    if _settings.gliner_enabled and is_service_enabled("json_extract"):
+        try:
+            await run_in_executor(gliner_service.load_model)
+            _logger.info("GLiNER2 model loaded and cached at startup (no-ai JSON extractor)")
+        except Exception as exc:
+            _logger.warning("GLiNER2 model warm-up failed (will lazy-load on first use): %s", exc)
+    else:
+        _logger.info("GLiNER2 no-ai JSON extractor is disabled, skipping model warm-up")
+
     redis = create_redis_client(_settings.redis_url) if _settings.redis_url else None
     scripts = await load_scripts(redis) if redis else {}
     app.state.redis = redis
@@ -170,6 +183,7 @@ async def lifespan(app: FastAPI):
     yield
     _logger.info("Shutting down...")
     await _scheduler_service.shutdown()
+    gliner_service.unload()
     await close_redis(redis)
     await _vector_store_service.close_all()
     await pool_manager.close_all()

app/api/v1/json_extract.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
+import asyncio
 import time
-from typing import Any, Optional
+from typing import Annotated, Any, Dict, List, Optional
 
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
 
+from app.config import get_settings
 from app.core.logger import get_logger
 from app.core.thread_pool import run_in_executor
+from app.services.gliner_service import gliner_service
 from app.services.json_service import extract_json
 
 logger = get_logger(__name__)
@@ -119,3 +122,205 @@ async def extract_json_endpoint(
         count=result.total_extracted,
         error_message=None,
     )
+
+
+class NoAiExtractRequest(BaseModel):
+    content: str = Field(
+        ...,
+        description=(
+            "Raw text to extract from (invoice text, OCR output, emails, etc.). "
+            "No external AI/LLM API is called."
+        ),
+        min_length=1,
+    )
+    mode: str = Field(
+        default="json",
+        pattern=r"^(json|entities)$",
+        description=(
+            "'json' extracts structured fields using a GLiNER2 `structure` schema; "
+            "'entities' extracts zero-shot entities using a `labels` list."
+        ),
+    )
+    structure: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description=(
+            "Required for mode='json'. GLiNER2 structure schema mapping a parent "
+            "key to field specs, e.g. "
+            '{"invoice": ["number::str::Invoice number", "total::str::Total amount"]}. '
+            "Field spec: name::dtype::choices::description."
+        ),
+    )
+    labels: Optional[List[str]] = Field(
+        default=None,
+        description="Required for mode='entities'. Entity types to detect, e.g. ['person', 'company', 'location'].",
+    )
+    threshold: float = Field(
+        default=0.5,
+        ge=0.0,
+        le=1.0,
+        description="Confidence threshold (0.0-1.0). Lower includes more candidates.",
+    )
+
+
+class NoAiExtractResponse(BaseModel):
+    success: bool
+    time_ms: float
+    mode: str
+    data: Any = None
+    count: int = 0
+    error_message: Optional[str] = None
+
+
+def _count_extracted(result: Any) -> int:
+    if not isinstance(result, dict):
+        return 0
+    total = 0
+    for parent, items in result.items():
+        if isinstance(items, list):
+            total += len(items)
+        elif isinstance(items, dict):
+            total += 1
+    return total
+
+
+@router.post(
+    "/json/feature-extract",
+    response_model=List[NoAiExtractResponse],
+    summary="Batch-extract structured JSON / entities with a local on-device model (no external AI)",
+    description=(
+        "Send up to 5 requests as a JSON array (1-5 items). All items are "
+        "processed concurrently on the shared thread pool, bounded by the local "
+        "model's concurrency limit. No external AI/LLM API is contacted -- ideal "
+        "for private or invoice/OCR data. mode='json' uses a structure schema to "
+        "pull named fields; mode='entities' detects a flat list of entity types. "
+        "Each item reports its own success/error. "
+        "Returns HTTP 503 if the model failed to load or is disabled."
+    ),
+)
+async def no_ai_extract_endpoint(
+    body: Annotated[
+        List[NoAiExtractRequest],
+        Field(
+            min_length=1,
+            max_length=5,
+            description="Array of up to 5 extraction requests, processed concurrently.",
+        ),
+    ],
+) -> List[NoAiExtractResponse]:
+    settings = get_settings()
+    total_start = time.perf_counter()
+
+    if not settings.gliner_enabled:
+        raise HTTPException(
+            status_code=503,
+            detail=NoAiExtractResponse(
+                success=False,
+                time_ms=0.0,
+                mode="",
+                data=None,
+                count=0,
+                error_message="The local extraction service is disabled.",
+            ).model_dump(),
+        )
+
+    if not gliner_service.is_loaded():
+        try:
+            await run_in_executor(gliner_service.load_model)
+        except Exception:
+            logger.exception("Lazy GLiNER2 model load failed on request")
+            raise HTTPException(
+                status_code=503,
+                detail=NoAiExtractResponse(
+                    success=False,
+                    time_ms=round((time.perf_counter() - total_start) * 1000, 3),
+                    mode="",
+                    data=None,
+                    count=0,
+                    error_message="The local extraction model is unavailable. Please try again later.",
+                ).model_dump(),
+            )
+
+    async def _process_one(index: int, item: NoAiExtractRequest) -> NoAiExtractResponse:
+        start = time.perf_counter()
+
+        if item.mode == "json" and not item.structure:
+            return NoAiExtractResponse(
+                success=False,
+                time_ms=round((time.perf_counter() - start) * 1000, 3),
+                mode=item.mode,
+                data=None,
+                count=0,
+                error_message="mode='json' requires a non-empty 'structure' schema.",
+            )
+        if item.mode == "entities" and not item.labels:
+            return NoAiExtractResponse(
+                success=False,
+                time_ms=round((time.perf_counter() - start) * 1000, 3),
+                mode=item.mode,
+                data=None,
+                count=0,
+                error_message="mode='entities' requires a non-empty 'labels' list.",
+            )
+        if len(item.content) > gliner_service.max_content_length:
+            return NoAiExtractResponse(
+                success=False,
+                time_ms=round((time.perf_counter() - start) * 1000, 3),
+                mode=item.mode,
+                data=None,
+                count=0,
+                error_message=(
+                    f"Content exceeds maximum length of {gliner_service.max_content_length:,} characters."
+                ),
+            )
+
+        try:
+            if item.mode == "json":
+                result = await run_in_executor(
+                    gliner_service.extract_json, item.content, item.structure, item.threshold
+                )
+            else:
+                result = await run_in_executor(
+                    gliner_service.extract_entities, item.content, item.labels, item.threshold
+                )
+        except Exception:
+            logger.exception("GLiNER2 inference failed for item %s", index)
+            return NoAiExtractResponse(
+                success=False,
+                time_ms=round((time.perf_counter() - start) * 1000, 3),
+                mode=item.mode,
+                data=None,
+                count=0,
+                error_message="Extraction failed. Please try again later.",
+            )
+
+        elapsed = round((time.perf_counter() - start) * 1000, 3)
+        logger.info(
+            "Feature-extract item extracted",
+            extra={
+                "index": index,
+                "mode": item.mode,
+                "count": _count_extracted(result),
+                "time_ms": elapsed,
+            },
+        )
+        return NoAiExtractResponse(
+            success=True,
+            time_ms=elapsed,
+            mode=item.mode,
+            data=result,
+            count=_count_extracted(result),
+            error_message=None,
+        )
+
+    results = await asyncio.gather(
+        *[_process_one(index, item) for index, item in enumerate(body)]
+    )
+
+    logger.info(
+        "Feature-extract batch processed",
+        extra={
+            "items": len(body),
+            "time_ms": round((time.perf_counter() - total_start) * 1000, 3),
+        },
+    )
+    return list(results)

app/config.py

@@ -219,6 +219,16 @@ class Settings(BaseSettings):
     scheduler_misfire_grace_time: int = Field(default=300, alias="SCHEDULER_MISFIRE_GRACE_TIME")
     scheduler_coordinator_prefix: str = Field(default="scheduler:", alias="SCHEDULER_COORDINATOR_PREFIX")
 
+    # Local GLiNER2 model for the /json/feature-extract extractor. Runs fully
+    # on-device (no external AI/LLM API). The model is loaded once and cached at
+    # startup; disable to skip loading and return 503 from the route. Every key
+    # is optional -- if not set, the default value below is used.
+    gliner_model: str = Field(default="fastino/gliner2-base-v1", alias="GLINER_MODEL")
+    gliner_enabled: bool = Field(default=True, alias="GLINER_ENABLED")
+    gliner_device: str = Field(default="cpu", alias="GLINER_DEVICE")
+    gliner_max_concurrent: int = Field(default=2, alias="GLINER_MAX_CONCURRENT")
+    gliner_max_content_length: int = Field(default=100_000, alias="GLINER_MAX_CONTENT_LENGTH")
+
     @property
     def max_upload_mb(self) -> int:
         return self.max_upload_bytes // (1024 * 1024)

app/services/gliner_service.py

@@ -0,0 +1,136 @@
+"""Local GLiNER2 extraction service for the ``/json/no-ai`` route.
+
+GLiNER2 performs zero-shot entity / structured-JSON extraction entirely
+on-device -- no external AI/LLM API is called, hence the "no-ai" route name.
+
+The model is loaded ONCE and cached for the life of the process (warmed at
+application startup via the lifespan hook in ``app/api/server.py`` and lazily
+loaded on first use if startup warm-up failed). Inference is CPU-bound and is
+dispatched on the shared thread pool by the route layer; a bounded semaphore
+serializes concurrent forwards on the shared model instance so a burst of
+requests cannot oversubscribe the CPU or the model's memory buffers.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import threading
+import time
+from typing import Any, Dict, List, Optional
+
+from app.config import get_settings
+from app.core.logger import get_logger
+
+logger = get_logger(__name__)
+
+DEFAULT_MODEL_ID = "fastino/gliner2-base-v1"
+DEFAULT_DEVICE = "cpu"
+DEFAULT_MAX_CONCURRENT = 2
+DEFAULT_MAX_CONTENT_LENGTH = 100_000
+
+
+class GLiNERService:
+    """Cached, concurrency-bounded wrapper around a GLiNER2 model instance."""
+
+    def __init__(
+        self,
+        model_id: str = DEFAULT_MODEL_ID,
+        device: str = DEFAULT_DEVICE,
+        max_concurrent: int = DEFAULT_MAX_CONCURRENT,
+        max_content_length: int = DEFAULT_MAX_CONTENT_LENGTH,
+    ) -> None:
+        self._model_id = model_id
+        self._device = device
+        self._max_content_length = max_content_length
+        self._max_concurrent = max(1, max_concurrent)
+        self._model: Any = None
+        self._lock = threading.Lock()
+        self._semaphore = threading.BoundedSemaphore(self._max_concurrent)
+        self._load_error: Optional[str] = None
+
+    # ------------------------------------------------------------------ #
+    # Lifecycle
+    # ------------------------------------------------------------------ #
+
+    def load_model(self) -> None:
+        """Load and cache the GLiNER2 model. Idempotent and thread-safe."""
+        if self.is_loaded():
+            return
+        with self._lock:
+            if self.is_loaded():
+                return
+            t0 = time.perf_counter()
+            try:
+                from gliner2 import GLiNER2
+
+                # GLiNER2 prints an emoji config banner to stdout on load, which
+                # crashes consoles with a non-UTF-8 encoding (e.g. Windows cp1252).
+                # Swallow it so loading works everywhere.
+                with contextlib.redirect_stdout(io.StringIO()):
+                    self._model = GLiNER2.from_pretrained(self._model_id, map_location=self._device)
+                self._load_error = None
+                logger.info(
+                    "GLiNER2 model loaded and cached (%s, device=%s) in %.2fs",
+                    self._model_id,
+                    self._device,
+                    time.perf_counter() - t0,
+                )
+            except Exception as exc:  # noqa: BLE001
+                self._load_error = str(exc)
+                logger.exception("GLiNER2 model load failed")
+                raise
+
+    def unload(self) -> None:
+        """Release the cached model (frees ~1.3 GB). Used at shutdown."""
+        with self._lock:
+            self._model = None
+            self._load_error = None
+
+    def is_loaded(self) -> bool:
+        return self._model is not None
+
+    def load_error(self) -> Optional[str]:
+        return self._load_error
+
+    @property
+    def model_id(self) -> str:
+        return self._model_id
+
+    @property
+    def device(self) -> str:
+        return self._device
+
+    @property
+    def max_content_length(self) -> int:
+        return self._max_content_length
+
+    # ------------------------------------------------------------------ #
+    # Inference (called on the shared thread pool)
+    # ------------------------------------------------------------------ #
+
+    def extract_json(self, text: str, structure: Dict[str, Any], threshold: float = 0.5) -> Dict[str, Any]:
+        """Structured JSON extraction with the cached local model."""
+        self._ensure_loaded()
+        with self._semaphore:
+            return self._model.extract_json(text, structure, threshold=threshold)
+
+    def extract_entities(self, text: str, labels: List[str], threshold: float = 0.5) -> Dict[str, Any]:
+        """Zero-shot entity extraction with the cached local model."""
+        self._ensure_loaded()
+        with self._semaphore:
+            return self._model.extract_entities(text, labels, threshold=threshold)
+
+    def _ensure_loaded(self) -> None:
+        if not self.is_loaded():
+            self.load_model()
+
+
+_settings = get_settings()
+
+gliner_service = GLiNERService(
+    model_id=_settings.gliner_model,
+    device=_settings.gliner_device,
+    max_concurrent=_settings.gliner_max_concurrent,
+    max_content_length=_settings.gliner_max_content_length,
+)

requirements.txt

@@ -42,6 +42,10 @@ xlrd==2.0.1
 
 # --- ML / embeddings / OCR models ---
 zvec==0.4.0
+# Semantic routing (aurelio-labs) used by the /semantic-router/route endpoint
+semantic-router==0.1.16
+# Local on-device model powering the /json/feature-extract route (no external AI).
+gliner2==1.3.2
 sentence-transformers==3.4.1
 transformers==4.50.2
 torch==2.5.1