""" HallucinationGuard-Env v4.2 — Production FastAPI Server with Stunning 3D Documentation Features: - Animated 3D particle background - Floating geometric objects - Glassmorphism UI elements - Gradient text and buttons - Interactive playground with live testing - Smooth animations and transitions Endpoints: Standard : POST /reset POST /step GET /state GET /health Session : POST /session/reset POST /session/step DELETE /session Leaderboard: GET /leaderboard POST /leaderboard/submit OpenEnv : GET /tasks POST /grader POST /baseline """ import sys, os, uuid, logging, dataclasses, enum, time, threading sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from fastapi import FastAPI, HTTPException, Header, Request from fastapi.responses import JSONResponse, RedirectResponse, HTMLResponse from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager from typing import Dict, Any, Optional, List from models import HallucinationAction, HallucinationObservation, HallucinationState from environment import HallucinationEnvironment from metrics import get_tracker from tasks import ( ALL_TASKS, get_task, task_id_for_difficulty, compute_task_score, ACTION_SCHEMA, ) logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) # ═══════════════════════════════════════════════════════════════════════════════ # STUNNING 3D ANIMATED DOCUMENTATION # ═══════════════════════════════════════════════════════════════════════════════ STUNNING_DOCS_HTML = """ HallucinationGuard-Env | Production RL Environment
v4.2.0 • OpenEnv Compatible • Production Ready

Train AI to Stop
Hallucinating

The production-grade RL environment for training and evaluating LLMs on hallucination avoidance. Built on 1M+ real-world examples across 38 benchmark datasets.

⚡ Try Interactive Demo 📖 Full API Reference
0
Examples
0
Datasets
0
Reward Components
0
Task Levels

Why HallucinationGuard?

Research-grade evaluation for grounded AI systems

🎯

Factual Grounding

Rewards answers derived strictly from provided context

🔬

9-Component Reward

Factual correctness, grounding, calibration, NLI, BERTScore...

📊

Real-World Datasets

SQuAD, HotpotQA, HaluEval, TruthfulQA, FEVER, and 33 more

⚡

Fast API

RESTful endpoints with OpenEnv compliance

🧠

NLI-Powered

Detects entailment and contradiction semantically

🏆

Leaderboard

Compare model performance across tasks

Three Difficulty Levels

Progressive curriculum from basic to adversarial

🟢

Task 1: Factual Grounding

Answer straightforward factual questions from a short context passage. Single-hop retrieval with unambiguous ground truth. Perfect for initial training.

Beginner
Datasets: SQuAD, BoolQ, ARC, OpenBookQA
🟡

Task 2: Multi-Hop Synthesis

Synthesize evidence from multiple sentences. Connect disparate facts without fabricating bridging information. Requires reasoning chains.

Intermediate
Datasets: HotpotQA, CoQA, NQ-Open, MS-MARCO
🔴

Task 3: Adversarial Resistance

Resist adversarial prompts designed to elicit hallucinations. Many questions are unanswerable — confident refusals are rewarded.

Advanced
Datasets: HaluEval, TruthfulQA, FEVER, AdversarialQA

Interactive Playground

Test the API directly in your browser

🔄 Reset Episode
📝 Submit Answer
📦 Batch Evaluate
🤖 Run Baseline
REQUEST BODY
RESPONSE
// Response will appear here... // // Click "Send Request" to test the API

All Endpoints

Complete API reference at a glance

POST /reset Start a new episode with optional difficulty and seed
POST /step Submit an answer with confidence and source citation
GET /state Get current episode state, accuracy, and skill rating
GET /tasks List all 3 tasks with complete action schema
POST /grader Score a completed episode (returns 0.0–1.0)
POST /baseline Run built-in heuristic baseline agent
POST /batch/evaluate Evaluate multiple Q&A pairs in one request
GET /leaderboard View ranked model performance
GET /health Service health check
GET /datasets Dataset statistics and distribution
""" # ═══════════════════════════════════════════════════════════════════════════════ # FASTAPI APP # ═══════════════════════════════════════════════════════════════════════════════ _default_env: Optional[HallucinationEnvironment] = None _env_loading = False _env_lock = threading.Lock() def _get_default_env() -> HallucinationEnvironment: global _default_env, _env_loading if _default_env is not None: return _default_env with _env_lock: if _default_env is not None: return _default_env _env_loading = True try: logger.info("Creating HallucinationEnvironment...") _default_env = HallucinationEnvironment() logger.info(f"Environment ready — {_default_env.dataset_loader.get_total_examples():,} examples loaded.") return _default_env except Exception as e: logger.error(f"Failed to create environment: {e}") # Minimal fallback environment from dataset_loader import DatasetLoader class MinimalEnv: def __init__(self): self.dataset_loader = DatasetLoader() self.dataset_loader.examples = [] def reset(self, **kwargs): return type('Obs', (), {'question': 'Placeholder', 'context': 'Context', 'reward': 0.0, 'done': False, 'info': {}})() def step(self, action): return type('Obs', (), {'reward': 0.0, 'done': False, 'is_hallucination': False, 'info': {}})() def state(self): return {} def close(self): pass _default_env = MinimalEnv() return _default_env finally: _env_loading = False @asynccontextmanager async def lifespan(app: FastAPI): global _default_env def preload_models(): try: logger.info("Preloading ML models...") from sentence_transformers import SentenceTransformer, CrossEncoder SentenceTransformer('all-MiniLM-L6-v2') CrossEncoder('cross-encoder/nli-deberta-v3-small') from rouge_score import rouge_scorer rouge_scorer.RougeScorer(["rouge1", "rouge2", "rougeL"], use_stemmer=True) try: from bert_score import BERTScorer BERTScorer(model_type='microsoft/deberta-v3-base', lang='en', device='cpu') except: pass logger.info("All ML models preloaded!") except Exception as e: logger.error(f"Model preload failed: {e}") threading.Thread(target=preload_models, daemon=True).start() def background_load(): try: logger.info("Background dataset loading...") env = _get_default_env() logger.info(f"Loaded {env.dataset_loader.get_total_examples():,} examples.") except Exception as e: logger.error(f"Background loading failed: {e}") threading.Thread(target=background_load, daemon=True).start() yield if _default_env: try: _default_env.close() except: pass app = FastAPI( lifespan=lifespan, title="HallucinationGuard-Env", version="4.2.0", docs_url="/swagger", redoc_url="/redoc", ) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) _sessions: Dict[str, HallucinationEnvironment] = {} import json as _json _LEADERBOARD_FILE = "/tmp/hallucination_guard_leaderboard.json" def _load_leaderboard(): if os.path.exists(_LEADERBOARD_FILE): try: return _json.load(open(_LEADERBOARD_FILE)) except: pass return {} def _save_leaderboard(lb): try: _json.dump(lb, open(_LEADERBOARD_FILE, "w"), indent=2) except: pass _leaderboard: Dict[str, Dict[str, Any]] = _load_leaderboard() def _safe_dict(obj): if hasattr(obj, 'model_dump'): return _safe_dict(obj.model_dump()) if hasattr(obj, 'dict'): return _safe_dict(obj.dict()) if dataclasses.is_dataclass(obj): return {f.name: _safe_dict(getattr(obj, f.name)) for f in dataclasses.fields(obj)} if isinstance(obj, enum.Enum): return obj.value if isinstance(obj, dict): return {k: _safe_dict(v) for k, v in obj.items()} if isinstance(obj, list): return [_safe_dict(i) for i in obj] if isinstance(obj, (str, int, float, bool, type(None))): return obj return str(obj) # ═══════════════════════════════════════════════════════════════════════════════ # ROUTES # ═══════════════════════════════════════════════════════════════════════════════ @app.get("/", include_in_schema=False, response_class=HTMLResponse) async def root(): return STUNNING_DOCS_HTML @app.get("/docs", include_in_schema=False, response_class=HTMLResponse) async def docs(): return STUNNING_DOCS_HTML @app.post("/reset", tags=["Environment"]) async def reset(body: Dict[str, Any] = {}): try: env = _get_default_env() obs = env.reset(**{k: v for k, v in body.items() if k in ("seed", "episode_id", "difficulty")}) return JSONResponse(content=_safe_dict(obs)) except Exception as e: import traceback logger.error(f"Reset error: {e}\n{traceback.format_exc()}") raise HTTPException(500, str(e)) @app.post("/step", tags=["Environment"]) async def step(action_data: Dict[str, Any]): try: env = _get_default_env() valid = set(HallucinationAction.model_fields.keys()) if hasattr(HallucinationAction, 'model_fields') else set(HallucinationAction.__fields__.keys()) action = HallucinationAction(**{k: v for k, v in action_data.items() if k in valid}) return JSONResponse(content=_safe_dict(env.step(action))) except Exception as e: raise HTTPException(500, str(e)) @app.get("/state", tags=["Environment"]) async def get_state(): try: return JSONResponse(content=_safe_dict(_get_default_env().state())) except Exception as e: raise HTTPException(500, str(e)) @app.get("/tasks", tags=["OpenEnv"]) async def list_tasks(): ordered = ["task_1_factual_grounding", "task_2_multi_hop_synthesis", "task_3_adversarial_resistance"] return {"tasks": [ALL_TASKS[t].to_dict() for t in ordered if t in ALL_TASKS], "action_schema": ACTION_SCHEMA} @app.post("/grader", tags=["OpenEnv"]) async def grade_episode(body: Dict[str, Any]): task_id = body.get("task_id") if not task_id: raise HTTPException(422, "'task_id' required") task = get_task(task_id) if not task: raise HTTPException(404, f"task_id '{task_id}' not found") rewards, infos = body.get("step_rewards", []), body.get("step_infos", []) if not infos and rewards: return {"task_id": task_id, "score": round(sum(rewards)/len(rewards), 4)} return compute_task_score(task, rewards, infos) @app.post("/baseline", tags=["OpenEnv"]) async def run_baseline(body: Dict[str, Any] = {}): steps = max(3, min(10, int(body.get("steps_per_task", 5)))) seed = int(body.get("seed", 42)) results = [] for task_id, diff in [("task_1_factual_grounding","beginner"),("task_2_multi_hop_synthesis","intermediate"),("task_3_adversarial_resistance","advanced")]: task = get_task(task_id) if not task: continue sid = f"bl_{task_id}_{seed}" if sid in _sessions: _sessions[sid].close() _sessions[sid] = HallucinationEnvironment(session_id=sid) obs = _safe_dict(_sessions[sid].reset(seed=seed, difficulty=diff)) rewards, infos = [], [] for _ in range(steps): if obs.get("done"): break ctx = obs.get("context", "") action = HallucinationAction(answer=ctx[:100], confidence=0.6, source_quote=ctx[:80]) obs = _safe_dict(_sessions[sid].step(action)) rewards.append(float(obs.get("reward") or 0)) infos.append({"correctness": obs.get("grounding_score", 0), "is_hallucination": obs.get("is_hallucination", False)}) results.append(compute_task_score(task, rewards, infos)) try: _sessions[sid].close(); del _sessions[sid] except: pass return {"tasks": results, "summary": {"overall_score": round(sum(r["score"] for r in results)/max(len(results),1), 4)}} @app.post("/batch/evaluate", tags=["Evaluation"]) async def batch_evaluate(body: Dict[str, Any]): items = body.get("items", []) if not items: raise HTTPException(422, "'items' required") from server.grader import calculate_reward results = [] for i, item in enumerate(items): r, info = calculate_reward(item.get("answer",""), item.get("confidence",0.5), item.get("source_quote",""), item.get("context",""), item.get("ground_truth","")) results.append({"index": i, "reward": round(r,4), "is_hallucination": info.get("is_hallucination", False)}) return {"total_items": len(results), "results": results} @app.get("/leaderboard", tags=["Leaderboard"]) async def leaderboard(): if not _leaderboard: return {"leaderboard": [], "message": "No submissions"} ranked = sorted(_leaderboard.values(), key=lambda x: x.get("avg_reward",0), reverse=True) for i, e in enumerate(ranked): e["rank"] = i+1 return {"leaderboard": ranked} @app.post("/leaderboard/submit", tags=["Leaderboard"]) async def submit_leaderboard(data: Dict[str, Any]): required = ["model_name", "avg_reward", "avg_accuracy", "hallucination_rate", "total_episodes", "total_steps"] if missing := [f for f in required if f not in data]: raise HTTPException(422, f"Missing: {missing}") _leaderboard[data["model_name"]] = {**data, "submitted_at": time.time()} _save_leaderboard(_leaderboard) return {"status": "submitted", "model_name": data["model_name"]} @app.get("/health", tags=["Info"]) async def health(): return {"status": "healthy", "version": "4.2.0"} @app.get("/metadata", tags=["OpenEnv"]) async def metadata(): return {"name": "hallucination-guard-env", "version": "4.2.0", "license": "MIT"} @app.get("/schema", tags=["OpenEnv"]) async def schema(): return {"action": {"type": "object", "required": ["answer"]}, "observation": {"type": "object"}} @app.get("/datasets", tags=["Info"]) async def datasets(): try: return {"total_examples": _get_default_env().dataset_loader.get_total_examples()} except: return {"total_examples": 0} @app.post("/mcp", tags=["OpenEnv"]) async def mcp(body: Dict[str, Any]): if body.get("method") == "tools/list": return {"jsonrpc": "2.0", "id": body.get("id",1), "result": {"tools": [{"name": "reset", "inputSchema": {"type": "object"}}, {"name": "step", "inputSchema": {"type": "object"}}]}} return {"jsonrpc": "2.0", "id": body.get("id",1), "result": {"name": "hallucination-guard-env", "version": "4.2.0"}} @app.middleware("http") async def log_req(request, call_next): resp = await call_next(request) logger.info(f"{request.method} {request.url.path} → {resp.status_code}") return resp def main(): import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860) if __name__ == "__main__": main()