Update app.py

app.py

@@ -1,97 +1,61 @@
 from fastapi import FastAPI
-from pydantic import BaseModel, Field, confloat
+from pydantic import BaseModel, Field
 from typing import Dict, Any, List, Literal
 from datetime import datetime
-import os
-
-# Optional tracing: initialize ONLY if key exists (won't crash if missing)
-TRACING = False
-try:
-    if os.getenv("TRACELOOP_API_KEY"):
-        from traceloop.sdk import Traceloop
-        from traceloop.sdk.decorators import workflow
-        Traceloop.init(app_name="ssra-auditor")
-        TRACING = True
-except Exception:
-    TRACING = False
-
-# Guardrails: schema guard only (semantic validators optional)
-from guardrails import Guard
 
 app = FastAPI()
 
 class Metrics(BaseModel):
-    TCS: confloat(ge=0, le=1) = 0.0
+    TCS: float = Field(ge=0.0, le=1.0, default=0.0)
     HDI: float = 0.0
-    PDS: confloat(ge=0, le=1) = 0.0
+    PDS: float = Field(ge=0.0, le=1.0, default=0.0)
     EVI: int = 0
-    CBS: confloat(ge=0, le=1) = 0.0
-    LQS: confloat(ge=0, le=1) = 0.0
+    CBS: float = Field(ge=0.0, le=1.0, default=0.0)
+    LQS: float = Field(ge=0.0, le=1.0, default=0.0)
 
 class AuditResult(BaseModel):
     metrics: Metrics
     decision: Literal['APPROVE','REWORK']
-    # Pydantic v2 uses min_length/max_length instead of v1's conlist(min_items)
-    deltas: List[str] = Field(default_factory=list, min_length=0)
-
-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)
-# Pydantic v2 migration: min_items → min_length etc. :contentReference[oaicite:1]{index=1}
+    deltas: List[str] = Field(default_factory=list)
 
 @app.get("/health")
 def health():
-    return {"ok": True, "service": "auditor", "ts": datetime.utcnow().isoformat(), "tracing": TRACING}
-
-def _audit_logic(payload: Dict[str, Any]) -> AuditResult:
-    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)),
-    }
+    return {"ok": True, "service": "auditor", "ts": datetime.utcnow().isoformat()}
+
+@app.post("/audit", response_model=AuditResult)
+def audit(payload: Dict[str, Any]):
+    bundle = payload.get("bundle", {})
+    m_in   = payload.get("metrics", {})
+
+    # seed metrics from input
+    m = Metrics(**{
+        "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"])
+    # heuristic patches
+    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
+    if regions > 8: m.HDI = max(m.HDI, 1.2)
+    elif regions > 4: m.HDI = max(m.HDI, 1.0)
+    else: m.HDI = max(m.HDI, 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, float(m.CBS))
 
-    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 one more evidence type (review/trials).")
-        if m["CBS"] > 0.6: deltas.append("Surface/address contrary findings explicitly.")
+    decision: Literal['APPROVE','REWORK'] = 'APPROVE'
 
-    out_raw = {"metrics": m, "decision": decision, "deltas": deltas}
-    _, validated, *_ = guard.parse(out_raw)
-    return AuditResult(**validated)
+    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 search.")
+        if m.EVI < 3:   deltas.append("Add at least one more evidence type (e.g., trials or review).")
+        if m.CBS > 0.6: deltas.append("Surface and address contrary findings explicitly.")
 
-if TRACING:
-    from traceloop.sdk.decorators import workflow
-    @workflow(name="audit_workflow")
-    @app.post("/audit", response_model=AuditResult)
-    def audit(payload: Dict[str, Any]):
-        return _audit_logic(payload)
-else:
-    @app.post("/audit", response_model=AuditResult)
-    def audit(payload: Dict[str, Any]):
-        return _audit_logic(payload)
+    return AuditResult(metrics=m, decision=decision, deltas=deltas)