Spaces:
Running
Running
SooryaKiranKakkarayil
Day 5: Report Agent, Gradio UI, CLI wrapper, Dockerfile, .dockerignore, fix pywin32 marker
e3518f9 | """ | |
| ScamSentry — Gradio UI | |
| Thin presentation layer over the same orchestrator engine cli.py uses — | |
| both call agents.orchestrator.run_pipeline() directly, no duplicated | |
| pipeline logic. The "Agent Trajectory" panel just dumps the same state | |
| dict run_pipeline() already returns (intake_result, reputation_signals, | |
| linguistic_signal, injection_guard_result, draft_verdict, | |
| semantic_check_result, final_verdict, report_text) — no new | |
| instrumentation needed. | |
| Note on types: some agents (url_reputation_agent) have no output_schema, | |
| so their state value is the model's raw JSON text, not a parsed dict — | |
| same reason risk_synthesis_gate.py defensively does | |
| `json.loads(x) if isinstance(x, str) else x` before touching intake_result | |
| or semantic_check_result. This file does the same before rendering. | |
| """ | |
| import json | |
| import mimetypes | |
| import gradio as gr | |
| from agents.orchestrator import run_pipeline | |
| TRAJECTORY_KEYS = [ | |
| "intake_result", | |
| "reputation_signals", | |
| "linguistic_signal", | |
| "injection_guard_result", | |
| "draft_verdict", | |
| "semantic_check_result", | |
| "final_verdict", | |
| ] | |
| def _maybe_parse(value): | |
| """Parse a value as JSON if it's a string; return it unchanged | |
| otherwise (or if it's a string but not valid JSON, e.g. report_text, | |
| which is deliberately plain prose).""" | |
| if isinstance(value, str): | |
| try: | |
| return json.loads(value) | |
| except (json.JSONDecodeError, TypeError): | |
| return value | |
| return value | |
| def format_verdict_panel(result: dict) -> str: | |
| verdict = _maybe_parse(result.get("final_verdict")) or {} | |
| report = result.get("report_text") or "(no report generated)" | |
| risk = str(verdict.get("risk_level", "unknown")).upper() | |
| confidence = verdict.get("confidence", "unknown") | |
| injection = verdict.get("injection_attempt_detected", False) | |
| lines = [f"### Risk level: {risk}", f"**Confidence:** {confidence}"] | |
| if injection: | |
| lines.append("**⚠️ Prompt-injection attempt detected in the analyzed content.**") | |
| lines.append("") | |
| lines.append(report) | |
| return "\n".join(lines) | |
| def format_trajectory(result: dict) -> str: | |
| trajectory = {k: _maybe_parse(result.get(k)) for k in TRAJECTORY_KEYS if k in result} | |
| return json.dumps(trajectory, indent=2, default=str) | |
| async def scan(raw_text, image_path): | |
| if not raw_text and not image_path: | |
| return "Please paste a message or upload a screenshot.", "{}" | |
| image_bytes = None | |
| image_mime = "image/png" | |
| if image_path: | |
| with open(image_path, "rb") as f: | |
| image_bytes = f.read() | |
| guessed, _ = mimetypes.guess_type(image_path) | |
| if guessed: | |
| image_mime = guessed | |
| try: | |
| result = await run_pipeline( | |
| text=raw_text or None, | |
| image_bytes=image_bytes, | |
| image_mime=image_mime, | |
| verbose=False, | |
| ) | |
| except Exception as e: | |
| return f"⚠️ Scan failed: {e}", "{}" | |
| if not result: | |
| return "⚠️ Pipeline returned no result — check the logs.", "{}" | |
| return format_verdict_panel(result), format_trajectory(result) | |
| with gr.Blocks(title="ScamSentry") as demo: | |
| gr.Markdown( | |
| "# ScamSentry\n" | |
| "Paste a suspicious message or upload a screenshot to check it for scam/phishing signals." | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| text_input = gr.Textbox( | |
| label="Message text", | |
| placeholder="Paste the email, SMS, or WhatsApp message here...", | |
| lines=8, | |
| ) | |
| image_input = gr.Image(label="Or upload a screenshot", type="filepath") | |
| scan_button = gr.Button("Scan", variant="primary") | |
| with gr.Column(): | |
| verdict_output = gr.Markdown(label="Verdict") | |
| with gr.Accordion("Agent Trajectory (debug view)", open=False): | |
| trajectory_output = gr.Code(label="Raw agent state", language="json") | |
| scan_button.click( | |
| fn=scan, | |
| inputs=[text_input, image_input], | |
| outputs=[verdict_output, trajectory_output], | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=7860) |