Spaces:
Sleeping
Sleeping
File size: 2,508 Bytes
5a05423 6e87ff8 96b82bc 6e87ff8 5a05423 96b82bc 6e87ff8 96b82bc 6e87ff8 96b82bc 6e87ff8 96b82bc 6e87ff8 96b82bc 6e87ff8 96b82bc 6e87ff8 96b82bc 6e87ff8 96b82bc 6e87ff8 |
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 |
from fastapi import FastAPI
from pydantic import BaseModel, conlist, confloat
from typing import Dict, Any, List, Literal
from guardrails import Guard
app = FastAPI()
class Metrics(BaseModel):
TCS: confloat(ge=0, le=1) = 0.0
HDI: float = 0.0
PDS: confloat(ge=0, le=1) = 0.0
EVI: int = 0
CBS: confloat(ge=0, le=1) = 0.0
LQS: confloat(ge=0, le=1) = 0.0
class AuditResult(BaseModel):
metrics: Metrics
decision: Literal['APPROVE','REWORK']
deltas: conlist(str, min_items=0)
# Guardrails schema (you can add validators later)
AUDIT_SCHEMA = {
"type": "object",
"properties": {
"decision": {"type":"string","enum":["APPROVE","REWORK"]},
"metrics": {"type":"object"},
"deltas": {"type":"array","items":{"type":"string"}}
},
"required": ["decision","metrics","deltas"]
}
guard = Guard().configure(output_schema=AUDIT_SCHEMA)
@app.get("/health")
def health():
return {"ok": True}
@app.post("/audit", response_model=AuditResult)
def audit(payload: Dict[str, Any]):
bundle = payload.get("bundle", {}) or {}
m_in = payload.get("metrics", {}) or {}
m = {
"TCS": float(m_in.get("TCS", 0.0)),
"HDI": float(m_in.get("HDI", 0.0)),
"PDS": float(m_in.get("PDS", 0.0)),
"EVI": int(m_in.get("EVI", 0)),
"CBS": float(m_in.get("CBS", 0.0)),
"LQS": float(m_in.get("LQS", 0.0)),
}
edges = len(bundle.get("network",{}).get("data",[]) or [])
regions = len((bundle.get("regions",{}).get("data") or {}).get("regions_ranked",[]) or [])
lit_ok = bool(bundle.get("literature"))
m["TCS"] = min(1.0, (1 if edges>0 else 0) + (1 if regions>0 else 0) + (1 if lit_ok else 0)) * 0.9
m["HDI"] = 1.2 if regions>8 else (1.0 if regions>4 else 0.8)
m["EVI"] = (1 if edges>0 else 0) + (1 if regions>0 else 0) + (1 if lit_ok else 0)
m["CBS"] = max(0.2, m["CBS"])
m["LQS"] = max(0.6, m["LQS"])
decision: Literal['APPROVE','REWORK'] = 'APPROVE'
deltas: List[str] = []
if (m["HDI"] < 1.0) or (m["EVI"] < 3) or (m["CBS"] > 0.6):
decision = 'REWORK'
if m["HDI"] < 1.0: deltas.append("Increase hypothesis diversity (broaden regions).")
if m["EVI"] < 3: deltas.append("Add at least one more evidence type (review/trials).")
if m["CBS"] > 0.6: deltas.append("Surface/address contrary findings explicitly.")
out_raw = {"metrics": m, "decision": decision, "deltas": deltas}
_, validated, *_ = guard.parse(out_raw)
return AuditResult(**validated)
|