from __future__ import annotations
"""
AFLHR Lite - FastAPI Backend
REST API wrapping AFLHREngine for the React frontend.
Supports both v1 (baseline) and v2 (decomposition + windowed NLI + BGE embeddings).
AI Disclosure: Development of this module was assisted by AI tools
for code structuring, debugging, and refactoring. The API design and endpoint
logic are the author's own work.
"""
import os
os.environ["PYTORCH_ENABLE_MPS_FALLBACK"] = "1"
os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["OMP_NUM_THREADS"] = "1"
os.environ["MKL_NUM_THREADS"] = "1"
import traceback
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
import threading
from pydantic import BaseModel, Field
from engine import AFLHREngine
from config import (
DEFAULT_PIVOT,
DEFAULT_STRICT_THRESHOLD,
DEFAULT_LENIENT_THRESHOLD,
GROQ_API_KEY,
KNOWLEDGE_BASE,
)
app = FastAPI(title="AFLHR Lite API", version="2.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/", response_class=HTMLResponse)
def root():
return """
AFLHR Lite API
AFLHR Lite API v2
The backend is running. Open the frontend at
http://localhost:5173
Endpoints:
GET /api/health — Health checkPOST /api/verify — Verify a query (supports v2 mode)GET /api/knowledge-base — Knowledge base infoGET /docs — Interactive API docs (Swagger)
"""
# Engine instances — v1 loaded at startup, v2 loaded on first v2 request
engine_v1: AFLHREngine | None = None
engine_v2: AFLHREngine | None = None
_v2_lock = threading.Lock()
_startup_error: str | None = None
_v2_load_error: str | None = None
@app.on_event("startup")
def startup():
global engine_v1, _startup_error
print("Loading AFLHR Engine (v1)...")
try:
engine_v1 = AFLHREngine()
print("Engine v1 ready.")
except Exception as e:
_startup_error = str(e)
print(f"ERROR: Engine v1 failed to load: {e}")
def get_v2_engine():
"""Lazy-load v2 engine on first v2 request (thread-safe).
Raises HTTPException(503) with a structured detail if load fails.
Caches the failure so repeat requests return fast instead of re-downloading.
"""
global engine_v2, _v2_load_error
if engine_v2 is not None:
return engine_v2
if _v2_load_error is not None:
raise HTTPException(
status_code=503,
detail=f"v2 engine unavailable: {_v2_load_error}. Uncheck 'v2 mode' to use the v1 baseline.",
)
with _v2_lock:
if engine_v2 is not None:
return engine_v2
if _v2_load_error is not None:
raise HTTPException(
status_code=503,
detail=f"v2 engine unavailable: {_v2_load_error}. Uncheck 'v2 mode' to use the v1 baseline.",
)
print("Loading AFLHR Engine (v2: windowed + decomposition + BGE)...")
try:
engine_v2 = AFLHREngine(
use_windowed_nli=True,
use_decomposition=True,
use_calibration=False, # T=10 at boundary — calibration hurts
use_bge_embeddings=True,
)
print("Engine v2 ready.")
return engine_v2
except Exception as e:
traceback.print_exc()
_v2_load_error = str(e)
raise HTTPException(
status_code=503,
detail=f"v2 engine failed to load: {e}. Uncheck 'v2 mode' to use the v1 baseline.",
)
class VerifyRequest(BaseModel):
query: str
pivot: float = DEFAULT_PIVOT
strict_threshold: float = DEFAULT_STRICT_THRESHOLD
lenient_threshold: float = DEFAULT_LENIENT_THRESHOLD
offline_mode: bool = False
v2_mode: bool = False
class ClaimScore(BaseModel):
claim: str
score: float
class VerifyResponse(BaseModel):
query: str
retrieval: dict
generation: str
nli_score: float
verdict: dict
# v2 fields
version: str = "v1"
nli_method: str = "whole"
n_claims: int = 1
per_claim: list[ClaimScore] = Field(default_factory=list)
@app.post("/api/verify", response_model=VerifyResponse)
def verify(req: VerifyRequest):
engine = get_v2_engine() if req.v2_mode else engine_v1
if engine is None:
msg = f"Engine not loaded: {_startup_error}" if _startup_error else "Engine not loaded"
raise HTTPException(status_code=503, detail=msg)
# Run the standard pipeline (retrieve + generate + single-pass verify + verdict)
try:
result = engine.run_pipeline(
query=req.query,
pivot=req.pivot,
strict_threshold=req.strict_threshold,
lenient_threshold=req.lenient_threshold,
offline_mode=req.offline_mode if GROQ_API_KEY else True,
)
except Exception as e:
traceback.print_exc()
raise HTTPException(status_code=500, detail=str(e))
# v2: also run decomposed verification for the per-claim breakdown
nli_score = result["nli_score"]
nli_method = "whole"
n_claims = 1
per_claim = []
if req.v2_mode:
try:
premise = result["retrieval"]["context"]
hypothesis = result["generation"]
decomp = engine.verify_decomposed(premise=premise, hypothesis=hypothesis)
nli_score = decomp["score"]
nli_method = "decomposed"
n_claims = decomp["n_claims"]
per_claim = [
ClaimScore(claim=c["claim"], score=round(c["score"], 4))
for c in decomp["per_claim"]
]
# Recompute verdict with decomposed score
result["verdict"] = engine.calculate_verdict(
retrieval_score=result["retrieval"]["retrieval_score"],
nli_score=nli_score,
pivot=req.pivot,
strict_threshold=req.strict_threshold,
lenient_threshold=req.lenient_threshold,
)
except Exception as e:
traceback.print_exc()
raise HTTPException(status_code=500, detail=str(e))
return VerifyResponse(
query=result["query"],
retrieval=result["retrieval"],
generation=result["generation"],
nli_score=nli_score,
verdict=result["verdict"],
version="v2" if req.v2_mode else "v1",
nli_method=nli_method,
n_claims=n_claims,
per_claim=per_claim,
)
@app.get("/api/health")
def health():
return {
"status": "ok",
"engine_v1_loaded": engine_v1 is not None,
"engine_v2_loaded": engine_v2 is not None,
"has_api_key": bool(GROQ_API_KEY),
}
@app.get("/api/knowledge-base")
def get_knowledge_base():
return {
"topics": [
{"name": "University of Westminster", "passages": 3},
{"name": "AI Hallucinations", "passages": 2},
{"name": "Climate of Sri Lanka (Distractor)", "passages": 1},
],
"total_passages": len(KNOWLEDGE_BASE),
}