import os import time from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from pydantic import ValidationError from graph.pipeline import run_pipeline from core.response_schema import AccessoryIQResponse, QueryRequest, EvidenceSource from retrieval.faiss_store import index_size from retrieval.ingest import ingest_sample_data from observability.logger import log_pipeline_run, log_retrieval_failure from observability.metrics import metrics INDEX_PATH = os.path.join("data", "faiss_index", "accessoryiq.index") # ────────────────────────────────────────────── # App setup # ────────────────────────────────────────────── app = FastAPI( title="AccessoryIQ v2", description=( "Compatibility intelligence API for hardware accessories. " "Combines FAISS retrieval, self-healing RAG, and trust-ranked evidence " "to answer accessory compatibility questions with grounded citations." ), version="2.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ────────────────────────────────────────────── # 1. Root # ────────────────────────────────────────────── @app.get("/") def root(): return {"status": "ok", "service": "AccessoryIQ v2", "version": "2.0.0"} # ────────────────────────────────────────────── # 2. Health check # ────────────────────────────────────────────── @app.get("/health") def health(): return {"status": "healthy", "index_size": index_size()} # ────────────────────────────────────────────── # 3. Main query endpoint # ────────────────────────────────────────────── _rate_limits = {} @app.post("/query", response_model=AccessoryIQResponse) def query(req: Request, request: QueryRequest): ip = req.client.host if req.client else "unknown" now = time.time() if ip not in _rate_limits: _rate_limits[ip] = [] _rate_limits[ip] = [t for t in _rate_limits[ip] if now - t < 60] if len(_rate_limits[ip]) >= 10: raise HTTPException(status_code=429, detail="Rate limit exceeded") _rate_limits[ip].append(now) if len(request.query) > 500: raise HTTPException(status_code=422, detail="Query too long") t_start = time.time() run_id = "" try: state = run_pipeline(request.query) run_id = state.get("run_id", "") latency_ms = (time.time() - t_start) * 1000 # Log outcome log_pipeline_run( run_id=run_id, query=request.query, result=state.get("result", ""), confidence=state.get("confidence", 0.0), latency_ms=latency_ms, ) # Record metrics for this query metrics.record_query( confidence=state.get("confidence", 0.0), latency_ms=latency_ms, faiss_hit=state.get("retrieval_passed", False), fallback=state.get("fallback_triggered", False), refused=state.get("result", "") == "REFUSED", ) if not state.get("retrieval_passed") and state.get("retrieval_attempts", 0) > 0: log_retrieval_failure( run_id=run_id, attempts=state.get("retrieval_attempts", 0), reason=state.get("refusal_reason", "retrieval did not pass critic"), ) # Build EvidenceSource list from state["sources"] evidence_sources = [] for src in state.get("sources", []): try: evidence_sources.append(EvidenceSource( title=src.get("title", src.get("doc_title", "")), url=src.get("url", src.get("source_url", "")), trust_tier=src.get("trust_tier", "tier_3"), trust_label=src.get("trust_label", "Unknown"), snippet=src.get("snippet", src.get("text", ""))[:300], score=src.get("score", src.get("relevance_score", 0.0)), )) except Exception: continue response = AccessoryIQResponse( result=state.get("result", "INSUFFICIENT_EVIDENCE"), accessory=state.get("accessory_type", ""), device=state.get("device_model", ""), explanation=state.get("explanation", ""), confidence=state.get("confidence", 0.0), trust_level=_derive_trust_level(state.get("confidence", 0.0)), evidence_sources=evidence_sources, warnings=state.get("warnings", []), reasoning_trace=state.get("reasoning_trace", []), refusal_reason=state.get("refusal_reason") or None, run_id=run_id, ) return response except ValidationError as exc: raise HTTPException(status_code=422, detail=str(exc)) except Exception as exc: latency_ms = (time.time() - t_start) * 1000 log_pipeline_run( run_id=run_id, query=request.query, result="ERROR", confidence=0.0, latency_ms=latency_ms, ) raise HTTPException(status_code=500, detail=str(exc)) # ────────────────────────────────────────────── # 4. Index info # ────────────────────────────────────────────── @app.get("/index-info") def index_info(): total = index_size() status = "ready" if total > 0 else "empty" return { "total_vectors": total, "index_path": INDEX_PATH, "status": status, } # ────────────────────────────────────────────── # 5. Ingest sample data # ────────────────────────────────────────────── @app.post("/ingest-sample") def ingest_sample(): try: ingest_sample_data() return {"message": "Sample data ingested", "index_size": index_size()} except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) @app.post("/rebuild-index") async def rebuild_index(): from retrieval.ingest import ingest_all_dataset_files from retrieval.faiss_store import index_size as get_index_size try: result = ingest_all_dataset_files() return { "message": "Index rebuilt successfully", "chunks_added": result, "total_vectors": get_index_size() } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # ────────────────────────────────────────────── # 6. Metrics # ────────────────────────────────────────────── @app.get("/metrics") def get_metrics(): return metrics.get_summary() # ────────────────────────────────────────────── # Helper # ────────────────────────────────────────────── def _derive_trust_level(confidence: float) -> str: from core.confidence import get_trust_level return get_trust_level(confidence) # ────────────────────────────────────────────── # __main__ # ────────────────────────────────────────────── if __name__ == "__main__": import uvicorn uvicorn.run("api.main:app", host="0.0.0.0", port=8000, reload=True)