Spaces:
Sleeping
Sleeping
| """ | |
| FastAPI backend for Hugging Face Spaces compatibility. | |
| Mirrors the Flask routes in `multiq/api.py`. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| from typing import Optional | |
| from fastapi import FastAPI, HTTPException | |
| from pydantic import BaseModel, Field | |
| from .bank import QuestionBank | |
| from .config import load_settings | |
| from .generator import QuestionGenerator | |
| _SMOKE = os.getenv("MQG_SMOKE", "false").lower() in ("1", "true", "yes") | |
| app = FastAPI(title="MultiQGen API", version="0.1.0") | |
| _GENERATOR: "QuestionGenerator | None" = None | |
| _BANK: "QuestionBank | None" = None | |
| _POINTS = {"easy": 10, "medium": 25, "hard": 50} | |
| def _get_generator(): | |
| global _GENERATOR | |
| if _GENERATOR is None: | |
| settings = load_settings() | |
| _GENERATOR = QuestionGenerator(settings, smoke=_SMOKE) | |
| return _GENERATOR | |
| def _get_bank(): | |
| global _BANK | |
| if _BANK is None: | |
| settings = load_settings() | |
| _BANK = QuestionBank(settings.bank_path) | |
| return _BANK | |
| def _game_meta(q: dict) -> dict: | |
| return { | |
| "points": _POINTS.get(q.get("difficulty", "easy"), 10), | |
| "tags": [q.get("lang"), q.get("type", q.get("qtype")), q.get("difficulty")], | |
| } | |
| class QuestionRequest(BaseModel): | |
| topic: str = "science" | |
| lang: str = "en" | |
| difficulty: Optional[str] = None | |
| type: str = Field("multiple_choice", alias="type") | |
| distractors: int = 3 | |
| temperature: Optional[float] = None | |
| class Config: | |
| populate_by_name = True | |
| class BatchRequest(BaseModel): | |
| topics: list[str] | |
| langs: list[str] = ["en", "ar"] | |
| difficulties: list[str] = ["easy", "medium", "hard"] | |
| class NextRequest(BaseModel): | |
| lang: Optional[str] = None | |
| difficulty: Optional[str] = None | |
| type: Optional[str] = None | |
| topic: Optional[str] = None | |
| class PrefillRequest(BaseModel): | |
| topics: list[str] | |
| langs: list[str] = ["en", "ar"] | |
| difficulties: list[str] = ["easy", "medium", "hard"] | |
| types: list[str] = ["multiple_choice"] | |
| per_cell: int = 3 | |
| use_bank: bool = True | |
| async def root(): | |
| return {"service": "MultiQGen", "version": "0.1.0", "status": "ok"} | |
| async def health(): | |
| return {"status": "healthy", "bank": _get_bank().stats()} | |
| async def generate(req: QuestionRequest): | |
| gen = _get_generator() | |
| try: | |
| q = gen.generate( | |
| topic=req.topic, | |
| lang=req.lang, | |
| difficulty=req.difficulty, | |
| qtype=req.type, | |
| n_distractors=req.distractors, | |
| temperature=req.temperature, | |
| ) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| return { | |
| "question": q.to_dict(), | |
| "game_meta": _game_meta(q.to_dict()), | |
| } | |
| async def next_question(req: NextRequest): | |
| """Game endpoint: draw a ready-made question from the bank (no LLM, no rate limits).""" | |
| bank = _get_bank() | |
| q = bank.draw(lang=req.lang, difficulty=req.difficulty, qtype=req.type, topic=req.topic) | |
| if q is None: | |
| raise HTTPException(status_code=404, detail="No pre-generated question matches. Run /api/v1/prefill first.") | |
| return {"question": q, "game_meta": _game_meta(q)} | |
| async def prefill(req: PrefillRequest): | |
| """Generate a batch of questions ahead of time and store them in the bank.""" | |
| gen = _get_generator() | |
| bank = _get_bank() | |
| generated = 0 | |
| failed = [] | |
| for topic in req.topics: | |
| for lang in req.langs: | |
| for diff in req.difficulties: | |
| for qtype in req.types: | |
| made = 0 | |
| attempts = 0 | |
| while made < req.per_cell and attempts < req.per_cell * 4: | |
| attempts += 1 | |
| try: | |
| q = gen.generate(topic=topic, lang=lang, difficulty=diff, qtype=qtype) | |
| except Exception as e: | |
| failed.append({"topic": topic, "lang": lang, "difficulty": diff, "error": str(e)[:200]}) | |
| break | |
| if bank.add(q.to_dict()): | |
| made += 1 | |
| generated += 1 | |
| else: | |
| attempts -= 1 # duplicate id: retry without burning attempts | |
| if made < req.per_cell: | |
| failed.append( | |
| {"topic": topic, "lang": lang, "difficulty": diff, "made": made, "target": req.per_cell} | |
| ) | |
| return {"generated": generated, "failed": failed, "bank": bank.stats()} | |
| async def bank_stats(): | |
| bank = _get_bank() | |
| return {"stats": bank.stats(), "topics": bank.topics()} | |
| async def batch(req: BatchRequest): | |
| gen = _get_generator() | |
| results = [] | |
| failed = [] | |
| for t in req.topics: | |
| for lang in req.langs: | |
| for diff in req.difficulties: | |
| try: | |
| q = gen.generate(topic=t, lang=lang, difficulty=diff) | |
| results.append(q.to_dict()) | |
| except Exception as e: | |
| failed.append({"topic": t, "lang": lang, "difficulty": diff, "error": str(e)}) | |
| return {"generated": results, "failed": failed} | |