Spaces:
Sleeping
Sleeping
File size: 2,213 Bytes
2f27a28 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | """FastAPI service exposing the Python Q&A RAG pipeline."""
import logging
import time
from collections import OrderedDict
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse
from groq import GroqError
from app import config
from app.rag import RAGPipeline, Retriever
from app.schemas import AskRequest, AskResponse, HealthResponse
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# question -> response dict, evicted FIFO once full.
_answer_cache: OrderedDict[str, dict] = OrderedDict()
@asynccontextmanager
async def lifespan(app: FastAPI):
retriever = Retriever()
app.state.pipeline = RAGPipeline(retriever)
logger.info("Index loaded: %d documents", retriever.count())
yield
app = FastAPI(
title="Python Programming Q&A Assistant",
description=(
"RAG-powered Q&A over the Stack Overflow Python dataset "
"(Kaggle: stackoverflow/pythonquestions), answered by Llama on Groq."
),
version="1.0.0",
lifespan=lifespan,
)
@app.get("/", include_in_schema=False)
async def root():
return RedirectResponse(url="/docs")
@app.get("/health", response_model=HealthResponse)
async def health():
return HealthResponse(
status="ok",
index_size=app.state.pipeline.retriever.count(),
model=config.GROQ_MODEL,
)
@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest):
cache_key = f"{req.question.strip().lower()}|{req.top_k}"
if cache_key in _answer_cache:
cached = _answer_cache[cache_key]
return AskResponse(**{**cached, "cached": True, "latency_ms": 0})
start = time.perf_counter()
try:
result = await app.state.pipeline.ask(req.question, top_k=req.top_k)
except GroqError as e:
logger.exception("LLM call failed")
raise HTTPException(status_code=502, detail=f"LLM provider error: {e}") from e
result["latency_ms"] = int((time.perf_counter() - start) * 1000)
_answer_cache[cache_key] = result
if len(_answer_cache) > config.ANSWER_CACHE_SIZE:
_answer_cache.popitem(last=False)
return AskResponse(**result)
|