Celes Renata
a72f336ad1
Multi-stage evidence-grounded inference architecture replacing the monolithic 9B model extraction pipeline. CPU-first specialist services handle routine extraction while the 9B vLLM model is preserved for semantic adjudication of ambiguous cases. Key components: - Capability-aware inference gateway (OpenAI-compatible + Ollama) - Endpoint registry with DB migrations and REST API - Sentence-aware document segmenter (property tests) - Deterministic financial parsing with offset integrity - Symbol resolution with ambiguity detection - Specialist service (GLiNER2, dynamic batching, K8s deployment) - Company-specific sentiment (FinBERT, calibration) - Retrieval-based novelty and duplicate detection - Confidence calibration pipeline - Deterministic routing engine (property tests) - 9B adjudication layer with VRAM gating - Stock-specific impact model (features, labels, baseline, trained) - Pipeline orchestrator (state machine, queues, leases, feature flags) - Bounded parallelism (async workers, semaphore, load shedding) - Observability (tracing, metrics, alerts) - Compatibility adapter (v3→v2 golden mapping tests) - Shadow/canary promotion framework - Active learning and fine-tuning pipeline Test results: 1,161 tests pass, ruff lint clean. All 282 spec tasks completed.
152 lines
4.9 KiB
Python
152 lines
4.9 KiB
Python
"""API router for specialist inference endpoints.
|
|
|
|
All endpoints live under /api/specialist/ prefix for consistency with
|
|
the broader Intelligence Pipeline v3 routing conventions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
|
|
from services.specialist.models import (
|
|
BatchResponse,
|
|
ClassificationRequest,
|
|
ExtractionRequest,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/specialist", tags=["specialist"])
|
|
|
|
# Engine is injected at app startup via app.state
|
|
_engine = None
|
|
|
|
|
|
def set_engine(engine) -> None: # noqa: ANN001
|
|
"""Set the engine reference used by all route handlers."""
|
|
global _engine
|
|
_engine = engine
|
|
|
|
|
|
def _get_engine():
|
|
"""Get the current engine or raise 503."""
|
|
if _engine is None or not _engine.is_ready:
|
|
raise HTTPException(status_code=503, detail="Specialist engine not ready")
|
|
return _engine
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entity extraction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/entities", response_model=BatchResponse)
|
|
async def extract_entities(request: ExtractionRequest) -> BatchResponse:
|
|
"""Batch entity extraction — returns spans with character offsets and scores."""
|
|
engine = _get_engine()
|
|
|
|
if len(request.texts) > engine.max_batch_size:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Batch size {len(request.texts)} exceeds maximum {engine.max_batch_size}",
|
|
)
|
|
|
|
start = time.perf_counter()
|
|
results = engine.extract_entities(request.texts, request.schema_labels)
|
|
elapsed_ms = (time.perf_counter() - start) * 1000
|
|
|
|
return BatchResponse(
|
|
results=results,
|
|
model_version=engine.model_name,
|
|
schema_version="specialist-v1",
|
|
processing_time_ms=round(elapsed_ms, 2),
|
|
batch_id=request.batch_id,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Event classification
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/classify", response_model=BatchResponse)
|
|
async def classify_events(request: ClassificationRequest) -> BatchResponse:
|
|
"""Batch event classification — returns labels with confidence scores."""
|
|
engine = _get_engine()
|
|
|
|
if len(request.texts) > engine.max_batch_size:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Batch size {len(request.texts)} exceeds maximum {engine.max_batch_size}",
|
|
)
|
|
|
|
start = time.perf_counter()
|
|
results = engine.classify_texts(request.texts, request.schema_labels)
|
|
elapsed_ms = (time.perf_counter() - start) * 1000
|
|
|
|
return BatchResponse(
|
|
results=results,
|
|
model_version=engine.model_name,
|
|
schema_version="specialist-v1",
|
|
processing_time_ms=round(elapsed_ms, 2),
|
|
batch_id=request.batch_id,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Relation extraction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/relations", response_model=BatchResponse)
|
|
async def extract_relations(request: ExtractionRequest) -> BatchResponse:
|
|
"""Batch relation extraction — returns subject-relation-object triples with spans."""
|
|
engine = _get_engine()
|
|
|
|
if len(request.texts) > engine.max_batch_size:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Batch size {len(request.texts)} exceeds maximum {engine.max_batch_size}",
|
|
)
|
|
|
|
start = time.perf_counter()
|
|
results = engine.extract_relations(request.texts, request.schema_labels)
|
|
elapsed_ms = (time.perf_counter() - start) * 1000
|
|
|
|
return BatchResponse(
|
|
results=results,
|
|
model_version=engine.model_name,
|
|
schema_version="specialist-v1",
|
|
processing_time_ms=round(elapsed_ms, 2),
|
|
batch_id=request.batch_id,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Structured extraction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/extract", response_model=BatchResponse)
|
|
async def extract_structured(request: ExtractionRequest) -> BatchResponse:
|
|
"""Structured fact extraction — returns key-value pairs with spans."""
|
|
engine = _get_engine()
|
|
|
|
if len(request.texts) > engine.max_batch_size:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Batch size {len(request.texts)} exceeds maximum {engine.max_batch_size}",
|
|
)
|
|
|
|
start = time.perf_counter()
|
|
results = engine.extract_structured(request.texts, request.schema_labels)
|
|
elapsed_ms = (time.perf_counter() - start) * 1000
|
|
|
|
return BatchResponse(
|
|
results=results,
|
|
model_version=engine.model_name,
|
|
schema_version="specialist-v1",
|
|
processing_time_ms=round(elapsed_ms, 2),
|
|
batch_id=request.batch_id,
|
|
)
|