from dotenv import load_dotenv load_dotenv() from fastapi import FastAPI from fastapi.responses import JSONResponse, HTMLResponse from pydantic import BaseModel from typing import Optional, List import uvicorn import os from environment.env import VoiceAuthenticityEnv app = FastAPI( title="Voice Authenticity OpenEnv", description="Multi-step agentic environment for detecting synthetic speech", version="2.0.0" ) TASKS = [ "clean_detection", "compressed_detection", "adversarial_detection", "streaming_detection", "phonecall_detection", ] envs = {task: VoiceAuthenticityEnv(task) for task in TASKS} current_task = "clean_detection" class ActionRequest(BaseModel): action_type: str = "final_classify" label: int = 0 confidence: float = 0.5 reasoning: str = "" focus: List[str] = [] task_name: Optional[str] = None @app.get("/web", response_class=HTMLResponse) def web_interface(): return """ Voice Authenticity OpenEnv
Live โ€” 5 tasks available

๐ŸŽ™๏ธ Voice Authenticity OpenEnv

Multi-step agentic environment for detecting synthetic (AI-generated) speech across real-world degradation and adversarial conditions.

5+
Tasks
5
Steps per episode
48
Feature dimensions

Tasks

clean_detection easy Classify real vs synthetic speech from clean, unmodified audio features
compressed_detection medium Classify speech under codec compression degradation
adversarial_detection hard Adversarially crafted synthetic speech with overlapping feature distributions
streaming_detection medium-hard Step-dependent noise soft-gating โ€” earlier steps noisier, later steps cleaner
phonecall_detection extreme Heavy codec compression and narrowband degradation simulating phone calls

5-Step Agent Protocol

1. request_temporal_features
Reveals jitter, shimmer, and HNR โ€” the core discriminating signals
2. request_spectral_features
Reveals 20 MFCC means, 20 MFCC stds, ZCR, spectral centroid
3. request_comparison
Compares sample to real/fake reference centroids via cosine similarity
4. analyze_evidence
Synthesizes all gathered signals into a structured evidence summary
5. final_classify
Submits final verdict: label (0=real, 1=synthetic) + confidence + reasoning. Terminates episode.

API Endpoints

POST /reset Reset episode, optionally set task_name
POST /step Submit action, receive observation + reward
GET /state Current environment state
GET /health Health check
GET /docs Interactive API documentation (Swagger UI)

Tags

openenv speech fraud-detection audio partial-observability multi-step confidence-calibration adversarial
""" @app.post("/reset") def reset(request: dict = {}): global current_task task = request.get("task_name", current_task) if request else current_task if task not in envs: task = "clean_detection" current_task = task obs = envs[current_task].reset() return JSONResponse({ "observation": obs.dict(), "done": False, "reward": 0.0, "info": {} }) @app.post("/step") def step(action: ActionRequest): global current_task task = action.task_name or current_task if task not in envs: task = current_task action_dict = { "action_type": action.action_type, "label": action.label, "confidence": action.confidence, "reasoning": action.reasoning, "focus": action.focus, } obs, reward, done, info = envs[task].step(action_dict) return JSONResponse({ "observation": obs.dict(), "reward": reward, "done": done, "info": info }) @app.get("/state") def state(): return JSONResponse(envs[current_task].state()) @app.get("/health") def health(): return {"status": "healthy", "service": "voice-authenticity-openenv"} @app.get("/") def root(): return { "name": "voice-authenticity-openenv", "version": "2.0.0", "status": "running", "tasks": TASKS, "web": "/web", "docs": "/docs" } def main(): uvicorn.run(app, host="0.0.0.0", port=7860) if __name__ == "__main__": main()