Spaces:
Running on Zero
Running on Zero
File size: 3,274 Bytes
31f2a28 | 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 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | import asyncio
from fastapi import FastAPI, HTTPException, Security, Depends
from fastapi.security.api_key import APIKeyHeader
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
from starlette.status import HTTP_403_FORBIDDEN, HTTP_429_TOO_MANY_REQUESTS
from evaluator import evaluate_trajectories_batch
from report_generator import generate_ai_report
from database import verify_api_key_db, save_evaluation_to_db
app = FastAPI(
title="Limina AI API",
description="High-performance infrastructure for multi-turn AI agent evaluation",
version="1.0.0"
)
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
async def get_api_key_context(header: str = Security(api_key_header)) -> Dict[str, Any]:
if not header:
raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Missing API Key (X-API-Key).")
auth_context = verify_api_key_db(header)
if not auth_context:
raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Access Denied. Invalid API Key.")
if auth_context.get("quota_exceeded"):
raise HTTPException(
status_code=HTTP_429_TOO_MANY_REQUESTS,
detail=f"Monthly Free Quota Exceeded ({auth_context['usage']}/{auth_context['limit']}). Upgrade to Limina Pro."
)
return auth_context
cpu_semaphore = asyncio.Semaphore(4)
class NodeInput(BaseModel):
id: str
type: str
text: str
expected_keys: Optional[List[str]] = None
runs: Optional[List[str]] = None
execution_time_ms: Optional[float] = None
class EdgeInput(BaseModel):
from_node: str = Field(..., alias="from")
to_node: str = Field(..., alias="to")
class Config:
allow_population_by_alias = True
class TrajectoryInput(BaseModel):
session_id: str
description: str
nodes: List[NodeInput]
edges: List[EdgeInput]
class EvaluationReport(BaseModel):
executive_summary: dict
regression_report: dict
results: list
narrative_report: str
@app.get("/")
async def read_root():
return {
"status": "online",
"engine": "Limina AI Trajectory Engine",
"version": "1.0.0",
"cloud_sync": "active"
}
@app.post("/evaluate/trajectory", response_model=EvaluationReport)
async def evaluate_trajectory(
payload: List[TrajectoryInput],
auth_ctx: Dict[str, Any] = Depends(get_api_key_context)
):
if not payload:
raise HTTPException(status_code=400, detail="Payload is empty.")
raw_payload = [p.dict(by_alias=True) for p in payload]
async with cpu_semaphore:
report = await asyncio.to_thread(evaluate_trajectories_batch, raw_payload, "standard", False)
if not report:
raise HTTPException(status_code=500, detail="Evaluation engine failed.")
ai_markdown_report = await asyncio.to_thread(generate_ai_report, report)
report["narrative_report"] = ai_markdown_report
if auth_ctx.get("project_id"):
asyncio.create_task(
asyncio.to_thread(save_evaluation_to_db, auth_ctx["project_id"], report)
)
return report |