Create app.py

app.py

@@ -0,0 +1,205 @@
+import os
+import requests
+import gradio as gr
+import json, ast 
+
+#----------------------------------------------------------------------------------------------- 
+#-----------------------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------------------------
+#-------------------------------BACKEND COMMUNICATION SETUP-------------------------------------
+#-----------------------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------------------------
+# URL of your private backend Space API
+BACKEND_URL = "https://your-username-your-backend-space.hf.space/respond"
+
+# Hugging Face token (add this as a Secret in your frontend Space)
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+
+#----------------------------------------------------------------------------------------------- 
+#-----------------------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------------------------
+#----------------------------FORMAT DEV-MODE REPORTER BOXES-------------------------------------
+#-----------------------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------------------------
+
+def coerce_json_obj(maybe_json):
+    """
+    If maybe_json is a JSON/py-literal string, parse it.
+    Otherwise return as-is.
+    """
+    if isinstance(maybe_json, (dict, list)):
+        return maybe_json
+    if isinstance(maybe_json, str):
+        s = maybe_json.strip()
+        try:
+            return json.loads(s)   # strict JSON
+        except Exception:
+            pass
+        try:
+            return ast.literal_eval(s)  # python literal
+        except Exception:
+            pass
+    return maybe_json  # fallback
+
+
+def format_specialities_need_md(data) -> str:
+    """
+    Expected shape:
+    {
+      "selected_specialities": ["Dermatology", "Clinical Pharmacology"],
+      "reasoning": {
+        "Dermatology": "...",
+        "Clinical Pharmacology": "..."
+      }
+    }
+    """
+    data = coerce_json_obj(data)
+
+    if not isinstance(data, dict):
+        return "```json\n" + json.dumps(data, indent=2, ensure_ascii=False) + "\n```"
+
+    selected = data.get("selected_specialities") or []
+    reasoning = data.get("reasoning") or {}
+
+    if not selected:
+        return "```json\n" + json.dumps(data, indent=2, ensure_ascii=False) + "\n```"
+
+    lines = []
+    for spec in selected:
+        why = reasoning.get(spec, "").strip()
+        if not why:
+            why = "No specific reasoning provided."
+        lines.append(f"- **{spec}**: {why}")
+    return "\n".join(lines)
+
+
+def format_critical_md(eval_history, aura_report) -> str:
+    """
+    Pretty-print eval history + aura report.
+    """
+    eval_history = coerce_json_obj(eval_history)
+    aura_report  = coerce_json_obj(aura_report)
+
+    if not isinstance(eval_history, list) or not isinstance(aura_report, dict):
+        payload = {"aura_report": aura_report, "eval_history": eval_history}
+        return "```json\n" + json.dumps(payload, indent=2, ensure_ascii=False) + "\n```"
+
+    lines = []
+    final_status = (aura_report.get("final_status") or "").upper()
+    attempts = aura_report.get("attempts")
+    summary_line = f"**Final status:** {final_status or 'N/A'}"
+    if attempts is not None:
+        summary_line += f"  •  **Attempts:** {attempts}"
+    lines.append(summary_line)
+    lines.append("")
+
+    if not eval_history:
+        lines.append("_No critical evaluations available._")
+    else:
+        for i, item in enumerate(eval_history, start=1):
+            att_no = item.get("attempt") or i
+            ce = item.get("critical_eval")
+            ce0 = ce[0] if isinstance(ce, list) and ce else (ce if isinstance(ce, dict) else {})
+            verdict = (ce0.get("verdict") or "").upper() if isinstance(ce0, dict) else ""
+            lines.append(f"### Attempt {att_no} — {verdict or 'UNKNOWN'}")
+
+            rationale = ce0.get("rationale") or ce0.get("reason") or ""
+            notes = ce0.get("notes") or ""
+            suggestions = ce0.get("improvement_suggestions") or []
+
+            if rationale:
+                lines.append(f"- **Rationale:** {rationale}")
+            if notes:
+                lines.append(f"- **Notes:** {notes}")
+            if isinstance(suggestions, list) and suggestions:
+                lines.append("- **Improvement suggestions:**")
+                for s in suggestions:
+                    lines.append(f"  - {s}")
+            lines.append("")
+    return "\n".join(lines).strip()
+
+
+#-----------------------------------------------------------------------------------------------
+#----------------------------------------------------------------------------------------------- 
+#-----------------------------------------------------------------------------------------------
+#--------------------------------------GRADIO CHAT INTERFACE------------------------------------
+#-----------------------------------------------------------------------------------------------
+#-----------------------------------------------------------------------------------------------
+#----------------------------------------------------------------------------------------------- 
+
+def respond(message, history):
+    try:
+        r = requests.post(
+            BACKEND_URL,
+            headers={"Authorization": f"Bearer {HF_TOKEN}"},
+            json={"message": message},
+            timeout=60
+        )
+        r.raise_for_status()
+        data = r.json()
+
+        final_answer = data.get("final_answer", "⚠️ No answer returned")
+        specialities_need = data.get("specialities_need", {})
+        eval_history = data.get("eval_history", [])
+        aura_report = data.get("aura_report", {})
+    except Exception as e:
+        final_answer = f"⚠️ Error contacting backend: {str(e)}"
+        specialities_need = {"error": str(e)}
+        eval_history = []
+        aura_report = {"final_status": "ERROR", "details": str(e)}
+
+    # Chatbot history (still using tuple format, you could upgrade to messages type)
+    history = (history or []) + [(message, final_answer)]
+
+    # Format side panels
+    specialities_md = format_specialities_need_md(specialities_need)
+    critical_md     = format_critical_md(eval_history, aura_report)
+
+    return history, specialities_md, critical_md
+
+def clear_all():
+    return [], {}, {}
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Project Nightingale - Due Diligence Explorer (Dev mode)")
+    gr.Markdown("""
+    Ask a question — we will:
+    1. Select sources intelligently with our internal routing agent  
+    2. Search the indexed guideline data using our **G-PRISM** retrieval pipeline  
+    3. Internally verify responses with the **AURA** adversarial critical evaluator mechanism
+    """)
+    with gr.Row():
+        with gr.Column(scale=3):
+            chatbot = gr.Chatbot(label="Clinical Knowledge Chat (G-PRISM Retrieval)", height=500, render_markdown=True, sanitize_html=False)
+            with gr.Row():
+                user_in = gr.Textbox(placeholder="Type your question…", scale=4)
+                send_btn = gr.Button("Send", scale=1, variant="primary")
+            clear_btn = gr.Button("Clear Chat")
+
+        with gr.Column(scale=2):
+            with gr.Group():
+                gr.Markdown("## Intelligent source selection (Internal Routing)")
+                spec_box = gr.Markdown(value="")    # Markdown instead of JSON
+
+            with gr.Group():
+                gr.Markdown("## Critical Appraisal (AURA)")
+                critical_box = gr.Markdown(value="") # Markdown instead of JSON
+
+
+    # wire events
+    user_in.submit(respond, inputs=[user_in, chatbot], outputs=[chatbot, spec_box, critical_box])
+    send_btn.click(respond, inputs=[user_in, chatbot], outputs=[chatbot, spec_box, critical_box])
+
+    # clear input after sending
+    user_in.submit(lambda: "", None, user_in)
+    send_btn.click(lambda: "", None, user_in)
+
+    # clear everything
+    clear_btn.click(clear_all, outputs=[chatbot, spec_box, critical_box])
+
+if __name__ == "__main__":
+    demo.launch()
+