Enable anonymous app access and OpenAI query embeddings

README.md

@@ -4,6 +4,6 @@ sdk: docker
 app_port: 7860
 ---
 
-Token-protected classification and GPT review through OpenAI, plus streaming access to a private terminology dataset.
+Public classification, GPT review and EDF embedding through OpenAI, plus terminology streaming.
 
-Configure OPENAI_API_KEY, PIPELINE_API_TOKEN and HF_TOKEN as Space secrets. Configure PIPELINE_HF_DATASET and PIPELINE_ALLOWED_ORIGINS as variables. No terminology vectors are stored in the Space.
+Configure OPENAI_API_KEY and HF_TOKEN as Space secrets. Set PIPELINE_PUBLIC_ACCESS=true for anonymous app access. Configure PIPELINE_HF_DATASET and PIPELINE_ALLOWED_ORIGINS as variables. No terminology vectors are stored in the Space.

backend/app.py

@@ -5,17 +5,20 @@ 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
+from fastapi import Depends, FastAPI, Header, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import StreamingResponse
 from starlette.background import BackgroundTask
@@ -23,7 +26,7 @@ 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
+from src.llm.client import LLMCallError, default_model_for, _get_openai_client
 
 load_dotenv()
 logger = logging.getLogger(__name__)
@@ -34,10 +37,13 @@ app.add_middleware(
         "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.")
@@ -46,6 +52,35 @@ def authorize(authorization: Annotated[str | None, Header()] = None):
         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)
@@ -53,11 +88,9 @@ class Candidate(BaseModel):
     description: str = Field(min_length=1, max_length=4096)
 
 
-class Batch(BaseModel):
+class EventRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
     edf: dict
-    candidates: list[Candidate] = Field(min_length=1, max_length=20)
-    use_possible_category: bool = False
 
     @model_validator(mode="after")
     def validate_edf(self):
@@ -78,6 +111,15 @@ class Batch(BaseModel):
                 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
+
+    @model_validator(mode="after")
+    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
@@ -85,18 +127,18 @@ class Batch(BaseModel):
 
 @app.get("/health")
 def health():
-    return {"status": "ok", "retrieval": "browser-minilm"}
+    return {"status": "ok", "retrieval": ["browser-minilm", "browser-openai"]}
 
 
 _ASSET_PATH = re.compile(
-    r"(?:manifest\.json|(?:ICD10CM|ICD9CM|ICPC|MDR|RCD2|SNOMEDCT_US)-[0-9a-f]{20}/"
+    r"(?:openai/)?(?:manifest\.json|(?:ICD10CM|ICD9CM|ICPC|MDR|RCD2|SNOMEDCT_US)-[0-9a-f]{20}/"
     r"(?:lexical\.json\.gz|[0-9]{7}\.(?:json\.gz|f32)))\Z"
 )
 
 
 @app.get("/api/assets/{asset_path:path}", dependencies=[Depends(authorize)])
 def terminology_asset(asset_path: str):
-    """Stream private HF dataset assets to authorized browsers without storing them."""
+    """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", "")
@@ -134,10 +176,11 @@ def terminology_asset(asset_path: str):
         finally:
             close()
 
-    # Private terminology data must not be cached by shared HTTP proxies.
-    # The authorized browser maintains its own content-addressed IndexedDB cache.
-    return StreamingResponse(chunks(), media_type="application/json" if asset_path == "manifest.json" else "application/octet-stream",
-                             headers={"Cache-Control": "private, no-store", "X-Content-Type-Options": "nosniff"},
+    # Public mode permits shared caching; authenticated deployments stay private.
+    public = os.getenv("PIPELINE_PUBLIC_ACCESS", "").lower() == "true"
+    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))
 
 
@@ -168,13 +211,32 @@ def classify_batch(batch: Batch, review: bool):
     return {"decisions": [asdict(row) for row in records], "provider": provider, "model": model}
 
 
-@app.post("/api/classify", dependencies=[Depends(authorize)])
+@app.post("/api/classify", dependencies=[Depends(authorize), Depends(model_rate_limit)])
 def classify(batch: Batch):
     return classify_batch(batch, review=False)
 
 
-@app.post("/api/review", dependencies=[Depends(authorize)])
+@app.post("/api/review", dependencies=[Depends(authorize), Depends(model_rate_limit)])
 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)
+
+
+@app.post("/api/embed", dependencies=[Depends(authorize), Depends(model_rate_limit)])
+def embed(event: EventRequest):
+    """Encode one EDF query; terminology vectors are always precomputed."""
+    query = EventDefinitionForm(name="browser-edf", **event.edf).to_embedding_query()
+    try:
+        response = _get_openai_client().embeddings.create(
+            model="text-embedding-3-large", input=query, dimensions=3072, 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) != 3072 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": 3072}