Spaces:
Sleeping
Sleeping
| """Small classification API for the static browser app. No embedding dependencies. | |
| Run: python -m uvicorn backend.app:app --host 127.0.0.1 --port 8000 | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import asdict, fields | |
| from collections import deque | |
| import hmac | |
| import json | |
| import logging | |
| import math | |
| import os | |
| import re | |
| import threading | |
| import time | |
| from typing import Annotated, Literal | |
| import httpx | |
| from dotenv import load_dotenv | |
| from fastapi import Depends, FastAPI, Header, HTTPException, Request | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import RedirectResponse, StreamingResponse | |
| from starlette.background import BackgroundTask | |
| from pydantic import BaseModel, ConfigDict, Field, model_validator | |
| from src.data.loaders import EventDefinitionForm | |
| from src.llm.classify import llm_classify | |
| from src.llm.client import LLMCallError, default_model_for, _get_openai_client | |
| load_dotenv() | |
| logger = logging.getLogger(__name__) | |
| app = FastAPI(title="Semantic Harmonization classification API", docs_url=None, redoc_url=None) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=[origin.strip() for origin in os.getenv( | |
| "PIPELINE_ALLOWED_ORIGINS", "http://127.0.0.1:5173,http://localhost:5173,https://ruurdkuiper.github.io" | |
| ).split(",") if origin.strip()], | |
| allow_methods=["GET", "POST"], allow_headers=["Authorization", "Content-Type"], | |
| expose_headers=["Retry-After"], | |
| ) | |
| def authorize(authorization: Annotated[str | None, Header()] = None): | |
| if os.getenv("PIPELINE_PUBLIC_ACCESS", "").lower() == "true": | |
| return | |
| token = os.getenv("PIPELINE_API_TOKEN", "") | |
| if not token: | |
| raise HTTPException(503, "The classification service needs an access token configured by its administrator.") | |
| supplied = (authorization or "").removeprefix("Bearer ") | |
| if not hmac.compare_digest(supplied.encode(), token.encode()): | |
| raise HTTPException(401, "Enter a valid classification service access token.") | |
| _model_windows = {} | |
| _model_usage_lock = threading.Lock() | |
| _daily_model_usage = {"day": None, "count": 0} | |
| def model_rate_limit(request: Request): | |
| """Bound anonymous paid requests; settings are server-side only.""" | |
| if os.getenv("PIPELINE_PUBLIC_ACCESS", "").lower() != "true": | |
| return | |
| now = time.time() | |
| client = request.client.host if request.client else "unknown" | |
| with _model_usage_lock: | |
| if _daily_model_usage["day"] != int(now // 86400): | |
| _daily_model_usage.update(day=int(now // 86400), count=0) | |
| if _daily_model_usage["count"] >= int(os.getenv("PIPELINE_DAILY_MODEL_REQUESTS", "10000")): | |
| raise HTTPException(429, "The shared daily model allowance has been reached. Please try again tomorrow.", headers={"Retry-After": "3600"}) | |
| window = _model_windows.setdefault(client, deque()) | |
| while window and window[0] <= now - 60: | |
| window.popleft() | |
| if len(window) >= int(os.getenv("PIPELINE_MODEL_REQUESTS_PER_MINUTE", "60")): | |
| raise HTTPException(429, "The model service is busy. Please retry shortly.", headers={"Retry-After": "60"}) | |
| window.append(now) | |
| _daily_model_usage["count"] += 1 | |
| if len(_model_windows) > 2048: | |
| for address in list(_model_windows): | |
| if not _model_windows[address] or _model_windows[address][-1] <= now - 60: | |
| del _model_windows[address] | |
| class Candidate(BaseModel): | |
| model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) | |
| code: str = Field(min_length=1, max_length=128) | |
| vocabulary: Literal["ICD10CM", "ICD9CM", "ICPC", "MDR", "RCD2", "SNOMEDCT_US"] | |
| description: str = Field(min_length=1, max_length=4096) | |
| class EventRequest(BaseModel): | |
| model_config = ConfigDict(extra="forbid") | |
| edf: dict | |
| def validate_edf(self): | |
| allowed = {field.name: field for field in fields(EventDefinitionForm) if field.name != "name"} | |
| if not set(self.edf) <= set(allowed): | |
| raise ValueError("Unknown EDF field") | |
| if len(json.dumps(self.edf)) > 65_536: | |
| raise ValueError("EDF is too large") | |
| for key, value in self.edf.items(): | |
| # Accept cleared YAML fields, matching the browser's normalization. | |
| is_list = key in {"inclusion_criteria", "exclusion_criteria", "narrow_decision_rules", "possible_decision_rules", | |
| "exclude_decision_rules", "synonyms", "candidate_anchor_codes", "ambiguities", "references"} | |
| if value is None: | |
| self.edf[key] = [] if is_list else "" | |
| elif is_list and (not isinstance(value, list) or any(not isinstance(item, str) for item in value)): | |
| raise ValueError(f"{key} must be a list of text values") | |
| elif not is_list and not isinstance(value, str): | |
| raise ValueError(f"{key} must be text") | |
| if not self.edf.get("preferred_name", "").strip(): | |
| raise ValueError("preferred_name is required") | |
| return self | |
| class Batch(EventRequest): | |
| candidates: list[Candidate] = Field(min_length=1, max_length=20) | |
| use_possible_category: bool = False | |
| def validate_candidates(self): | |
| if len({(row.code, row.vocabulary) for row in self.candidates}) != len(self.candidates): | |
| raise ValueError("Duplicate candidates in batch") | |
| return self | |
| class EmbeddingRequest(EventRequest): | |
| # Preserve the original API default for older clients and full-size packs. | |
| dimensions: Literal[256, 3072] = 3072 | |
| def health(): | |
| return {"status": "ok", "retrieval": ["browser-minilm", "browser-openai"]} | |
| _ASSET_PATH = re.compile( | |
| r"(?:(?:openai|openai-256-int8|minilm-int8)/)?(?:manifest\.json|(?:ICD10CM|ICD9CM|ICPC|MDR|RCD2|SNOMEDCT_US)-[0-9a-f]{20}/" | |
| r"(?:lexical\.json\.gz|[0-9]{7}\.(?:json\.gz|f32|i8)))\Z" | |
| ) | |
| def terminology_asset(asset_path: str): | |
| """Stream HF dataset assets without exposing the server's dataset credential.""" | |
| if not _ASSET_PATH.fullmatch(asset_path): | |
| raise HTTPException(404, "Unknown terminology asset.") | |
| repo = os.getenv("PIPELINE_HF_DATASET", "") | |
| revision = os.getenv("PIPELINE_HF_REVISION", "main") | |
| if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo) or not re.fullmatch(r"[A-Za-z0-9_-]+", revision): | |
| raise HTTPException(503, "The Hugging Face terminology dataset is not configured.") | |
| if not os.getenv("HF_TOKEN"): | |
| raise HTTPException(503, "Private terminology downloads need a read-only HF_TOKEN configured in the Space secrets.") | |
| headers = {"Accept-Encoding": "identity"} | |
| if os.getenv("HF_TOKEN"): | |
| headers["Authorization"] = "Bearer " + os.environ["HF_TOKEN"] | |
| # HTTPX strips Authorization on cross-origin CDN redirects. Neither the HF | |
| # token nor provider keys are forwarded to the browser. | |
| client = httpx.Client(timeout=60, follow_redirects=True) | |
| url = f"https://huggingface.co/datasets/{repo}/resolve/{revision}/{asset_path}" | |
| public = os.getenv("PIPELINE_PUBLIC_ACCESS", "").lower() == "true" | |
| if public and asset_path.endswith((".f32", ".i8")): | |
| # Let the CDN serve large vectors directly. Only the temporary link to | |
| # this file reaches the browser; the HF access token stays server-side. | |
| try: | |
| resolved = client.head(url, headers=headers, follow_redirects=False) | |
| location = resolved.headers.get("location", "") | |
| target = httpx.URL(location) | |
| trusted = target.scheme == "https" and any( | |
| target.host == domain or target.host.endswith("." + domain) | |
| for domain in ("hf.co", "huggingface.co")) | |
| if resolved.is_redirect and trusted: | |
| client.close() | |
| # Signed links expire: cache the verified bytes in the browser, | |
| # never the redirect itself. | |
| return RedirectResponse(location, status_code=307, headers={"Cache-Control": "no-store"}) | |
| except (httpx.HTTPError, httpx.InvalidURL): | |
| pass # Retain streaming as a fallback when resolution is unavailable. | |
| try: | |
| response = client.send(client.build_request("GET", | |
| url, headers=headers), stream=True) | |
| except httpx.HTTPError: | |
| client.close() | |
| raise HTTPException(502, "The terminology download could not be reached. Please retry.") from None | |
| if response.status_code != 200: | |
| status = response.status_code | |
| response.close() | |
| client.close() | |
| raise HTTPException(404 if status == 404 else 502, | |
| "The terminology asset is unavailable. The administrator should check the dataset access settings.") | |
| def close(): | |
| response.close() | |
| client.close() | |
| def chunks(): | |
| try: | |
| yield from response.iter_raw(chunk_size=65536) | |
| finally: | |
| close() | |
| # Public mode permits shared caching; authenticated deployments stay private. | |
| cache_control = ("public, max-age=60" if asset_path.endswith("manifest.json") else "public, max-age=31536000, immutable") if public else "private, no-store" | |
| return StreamingResponse(chunks(), media_type="application/json" if asset_path.endswith("manifest.json") else "application/octet-stream", | |
| headers={"Cache-Control": cache_control, "X-Content-Type-Options": "nosniff"}, | |
| background=BackgroundTask(close)) | |
| def classify_batch(batch: Batch, review: bool): | |
| provider = "openai" if review else os.getenv("PIPELINE_CLASSIFIER_PROVIDER", "jev") | |
| if provider not in {"jev", "openai", "anthropic", "google"}: | |
| raise HTTPException(503, "The classification provider is not configured correctly.") | |
| model = os.getenv("PIPELINE_REVIEW_MODEL" if review else "PIPELINE_CLASSIFIER_MODEL") or default_model_for(provider) | |
| try: | |
| records = llm_classify( | |
| batch.candidates, EventDefinitionForm(name="browser-edf", **batch.edf), | |
| provider=provider, model=model, max_retries=3, batch_size=10 if review else 20, | |
| use_possible_category=batch.use_possible_category, reasoning_effort="medium", adaptive_stopping=None, | |
| ) | |
| except (LLMCallError, ValueError, RuntimeError): | |
| logger.exception("Classification request failed") | |
| # Provider responses may contain sensitive diagnostic text. Keep it server-side. | |
| raise HTTPException(502, "The model request failed. Retry the batch or ask the service administrator to check its configuration.") from None | |
| expected = {(row.code, row.vocabulary) for row in batch.candidates} | |
| labels = {"Narrow", "Exclude", "Possible"} if batch.use_possible_category else {"Narrow", "Exclude"} | |
| keys = [(row.code, row.vocabulary) for row in records] | |
| if len(keys) != len(expected) or set(keys) != expected or any( | |
| row.label not in labels or not math.isfinite(row.confidence) or not 0 <= row.confidence <= 1 | |
| or row.explanation == "No classification returned by LLM; routed for manual review." | |
| for row in records | |
| ): | |
| raise HTTPException(502, "The model returned an incomplete or invalid batch. Retry to continue.") | |
| return {"decisions": [asdict(row) for row in records], "provider": provider, "model": model} | |
| def classify(batch: Batch): | |
| return classify_batch(batch, review=False) | |
| def review(batch: Batch): | |
| if len(batch.candidates) > 10: | |
| raise HTTPException(422, "Review accepts at most ten candidates per batch.") | |
| return classify_batch(batch, review=True) | |
| def embed(event: EmbeddingRequest): | |
| """Encode one EDF query; terminology vectors are always precomputed.""" | |
| query = EventDefinitionForm(name="browser-edf", **event.edf).to_embedding_query(include_labels=False) | |
| try: | |
| response = _get_openai_client().embeddings.create( | |
| model="text-embedding-3-large", input=query, dimensions=event.dimensions, encoding_format="float") | |
| except Exception as exc: | |
| logger.warning("Embedding request failed (%s)", type(exc).__name__) | |
| raise HTTPException(502, "The event definition could not be embedded. Please retry; very long definitions may need shortening.") from None | |
| vector = response.data[0].embedding | |
| if len(vector) != event.dimensions or any(not math.isfinite(value) for value in vector): | |
| raise HTTPException(502, "The embedding service returned an invalid vector.") | |
| norm = math.sqrt(sum(value * value for value in vector)) | |
| if norm < 1e-12: | |
| raise HTTPException(502, "The embedding service returned an empty vector.") | |
| return {"vector": [value / norm for value in vector], "model": "text-embedding-3-large", "dimensions": event.dimensions} | |