| """Gradio demo for Agent 9 — Agentic Workflow System.""" |
|
|
| from __future__ import annotations |
|
|
| import os |
| import time |
| from typing import Any |
|
|
| import gradio as gr |
|
|
| from src.demo_presentation import ( |
| APP_CSS, |
| EMPTY_ACTIVITY_HTML, |
| EMPTY_AI_HTML, |
| EMPTY_JOURNEY_HTML, |
| EMPTY_OUTCOME_HTML, |
| EMPTY_PATH_HTML, |
| EMPTY_ROUTING_HTML, |
| HERO_HTML, |
| PATTERN_HTML, |
| activity_html, |
| activity_start_html, |
| playback_delay, |
| ) |
| from src.demo_workflow import build_onboarding_registry, build_onboarding_workflow |
| from src.evaluation import run_evaluation_suite |
| from src.executor import ( |
| WorkflowExecutionError, |
| WorkflowHumanDecisionError, |
| execute_workflow, |
| submit_human_decision, |
| ) |
| from src.hf_provider import ( |
| build_hf_onboarding_model_from_env, |
| configured_model_id, |
| live_ai_enabled_from_env, |
| ) |
| from src.persistence import RunNotFoundError, SQLiteStateStore |
| from src.schemas import HumanDecision, WorkflowRun |
| from src.ui_presenters import ( |
| CASE_HEADERS, |
| EVENT_HEADERS, |
| METRIC_HEADERS, |
| NODE_HEADERS, |
| REVIEW_HEADERS, |
| ai_insight_html, |
| business_journey_html, |
| business_outcome_html, |
| evaluation_case_rows, |
| evaluation_metric_rows, |
| routing_explanation_html, |
| run_bundle, |
| workflow_path_html, |
| ) |
|
|
| WORKFLOW = build_onboarding_workflow() |
| DATABASE_PATH = os.getenv("WORKFLOW_DB_PATH", "workflow_runs.db") |
| STATE_STORE = SQLiteStateStore(DATABASE_PATH) |
| LIVE_MODEL_ID = configured_model_id() |
| LIVE_ONBOARDING_MODEL = build_hf_onboarding_model_from_env() |
|
|
| STORY_SPECS: dict[str, dict[str, Any]] = { |
| "Harbor Family — straightforward household": { |
| "household_id": "HH-HARBOR-STANDARD", |
| "household_type": "JOINT", |
| "onboarding_notes": "The fictional Harbor family is beginning a standard advisory relationship. The household has two adults, straightforward planning needs, and a complete synthetic intake.", |
| "documents_complete": True, |
| "identity_status": "VERIFIED", |
| "relationship_complexity": "STANDARD", |
| "simulation_mode": "NONE", |
| }, |
| "Harbor Family — temporary onboarding-service issue": { |
| "household_id": "HH-HARBOR-RETRY", |
| "household_type": "JOINT", |
| "onboarding_notes": "The fictional Harbor family has a straightforward synthetic intake, but the demo will simulate one temporary package-preparation service interruption.", |
| "documents_complete": True, |
| "identity_status": "VERIFIED", |
| "relationship_complexity": "STANDARD", |
| "simulation_mode": "TRANSIENT_ONCE", |
| }, |
| "Redwood Family Trust — human review required": { |
| "household_id": "HH-REDWOOD-TRUST", |
| "household_type": "TRUST", |
| "onboarding_notes": "The fictional Redwood Family Trust has multiple interested parties and a more complex planning relationship. The synthetic intake is complete but should demonstrate an exception-review path.", |
| "documents_complete": True, |
| "identity_status": "VERIFIED", |
| "relationship_complexity": "COMPLEX", |
| "simulation_mode": "NONE", |
| }, |
| "Cedar Household — permanent dependency failure": { |
| "household_id": "HH-CEDAR-FAILURE", |
| "household_type": "JOINT", |
| "onboarding_notes": "The fictional Cedar household has a standard synthetic intake, but the demo will simulate a permanent downstream package-preparation failure.", |
| "documents_complete": True, |
| "identity_status": "VERIFIED", |
| "relationship_complexity": "STANDARD", |
| "simulation_mode": "PERMANENT", |
| }, |
| } |
|
|
| RunBundle = tuple[str, str, list[list[Any]], list[list[str]], list[list[str]], dict[str, Any]] |
| AppBundle = tuple[ |
| str, |
| str, |
| list[list[Any]], |
| list[list[str]], |
| list[list[str]], |
| dict[str, Any], |
| str, |
| str, |
| str, |
| str, |
| str, |
| ] |
|
|
|
|
| def _ai_runtime_html() -> str: |
| if LIVE_ONBOARDING_MODEL is not None: |
| return ( |
| '<div class="ai-runtime ai-runtime-live"><strong>● Live AI connected</strong><br>' |
| f'The AI Intake Organizer is using Hugging Face Inference Providers with <code>{LIVE_MODEL_ID}</code>. ' |
| 'Its output still must pass the same strict Pydantic schema before the workflow can use it.</div>' |
| ) |
| if live_ai_enabled_from_env(): |
| return ( |
| '<div class="ai-runtime ai-runtime-waiting"><strong>◐ Live AI requested but not connected</strong><br>' |
| 'LIVE_AI_ENABLED is on, but no usable HF_TOKEN was found. The demo is using its explicit deterministic fallback.</div>' |
| ) |
| return ( |
| '<div class="ai-runtime"><strong>○ AI boundary ready; deterministic fallback active</strong><br>' |
| 'Live inference is deliberately opt-in. The same bounded node can use Hugging Face Inference Providers when the deployment sets <code>LIVE_AI_ENABLED=true</code> and supplies <code>HF_TOKEN</code>.</div>' |
| ) |
|
|
|
|
| def _build_registry(): |
| return build_onboarding_registry( |
| onboarding_model=LIVE_ONBOARDING_MODEL, |
| model_label=LIVE_MODEL_ID if LIVE_ONBOARDING_MODEL is not None else None, |
| ) |
|
|
|
|
| def run_request_ui( |
| household_id: str, |
| household_type: str, |
| onboarding_notes: str, |
| documents_complete: bool, |
| identity_status: str, |
| relationship_complexity: str, |
| simulation_mode: str, |
| ) -> AppBundle: |
| context = { |
| "household_id": household_id, |
| "household_type": household_type, |
| "onboarding_notes": onboarding_notes, |
| "documents_complete": documents_complete, |
| "identity_status": identity_status, |
| "relationship_complexity": relationship_complexity, |
| "simulation_mode": simulation_mode, |
| } |
| try: |
| run = execute_workflow(WORKFLOW, _build_registry(), context=context, state_store=STATE_STORE) |
| return _app_bundle(run) |
| except WorkflowExecutionError as exc: |
| return _bundle_with_notice(exc.run, "The onboarding workflow stopped after a controlled failure.") |
| except Exception as exc: |
| return _empty_bundle(f"Unable to start onboarding workflow: {type(exc).__name__}") |
|
|
|
|
| def run_story_ui(story: str) -> AppBundle: |
| spec = STORY_SPECS.get(story) |
| if spec is None: |
| return _empty_bundle("Choose one of the four onboarding stories before running the workflow.") |
| return run_request_ui(**spec) |
|
|
|
|
| def refresh_run_ui(run_id: str) -> AppBundle: |
| if not run_id or not run_id.strip(): |
| return _empty_bundle("Enter a run ID to load a persisted workflow.") |
| try: |
| return _app_bundle(STATE_STORE.load(run_id.strip())) |
| except RunNotFoundError: |
| return _empty_bundle("No persisted workflow run was found for that ID.", run_id.strip()) |
| except Exception as exc: |
| return _empty_bundle(f"Unable to load workflow: {type(exc).__name__}", run_id.strip()) |
|
|
|
|
| def submit_human_decision_ui(run_id: str, decision: str) -> AppBundle: |
| if not run_id or not run_id.strip(): |
| return _empty_bundle("Enter a run ID before submitting a human decision.") |
| normalized_run_id = run_id.strip() |
| try: |
| run = STATE_STORE.load(normalized_run_id) |
| except RunNotFoundError: |
| return _empty_bundle("No persisted workflow run was found for that ID.", normalized_run_id) |
| open_reviews = [review for review in run.human_reviews if review.decision is None] |
| if len(open_reviews) != 1: |
| return _bundle_with_notice(run, "This run does not have exactly one open human review. Reload the run and inspect the Human Review table.") |
| try: |
| updated = submit_human_decision( |
| WORKFLOW, |
| _build_registry(), |
| run_id=normalized_run_id, |
| review_id=open_reviews[0].review_id, |
| decision=HumanDecision(decision), |
| state_store=STATE_STORE, |
| ) |
| return _app_bundle(updated) |
| except (WorkflowHumanDecisionError, WorkflowExecutionError) as exc: |
| return _bundle_with_notice(exc.run, str(exc)) |
| except ValueError: |
| return _bundle_with_notice(run, "Unknown human decision.") |
| except Exception as exc: |
| return _bundle_with_notice(run, f"Unable to apply decision: {type(exc).__name__}") |
|
|
|
|
| def run_evaluation_ui() -> tuple[str, list[list[str]], list[list[str]]]: |
| report = run_evaluation_suite() |
| summary = ( |
| "### Wealth Onboarding Workflow Evaluation\n" |
| f"**Cases:** {report.passed_cases}/{report.total_cases} passed \n" |
| f"**Pass rate:** {report.case_pass_rate:.2f} \n" |
| f"**Duplicate executions:** {report.metrics['duplicate_execution_count']} \n" |
| f"**Invalid transitions:** {report.metrics['invalid_transition_count']}" |
| ) |
| return summary, evaluation_metric_rows(report), evaluation_case_rows(report) |
|
|
|
|
| def _app_bundle(run: WorkflowRun) -> AppBundle: |
| technical: RunBundle = run_bundle(run) |
| return ( |
| *technical, |
| business_outcome_html(run), |
| workflow_path_html(run), |
| ai_insight_html(run), |
| routing_explanation_html(run), |
| business_journey_html(run), |
| ) |
|
|
|
|
| def _bundle_with_notice(run: WorkflowRun, notice: str) -> AppBundle: |
| technical = run_bundle(run) |
| return ( |
| f"> {notice}\n\n{technical[0]}", |
| *technical[1:], |
| business_outcome_html(run), |
| workflow_path_html(run), |
| ai_insight_html(run), |
| routing_explanation_html(run), |
| business_journey_html(run), |
| ) |
|
|
|
|
| def _empty_bundle(message: str, run_id: str = "") -> AppBundle: |
| return ( |
| f"### Workflow Status\n{message}", run_id, [], [], [], {}, |
| EMPTY_OUTCOME_HTML, EMPTY_PATH_HTML, EMPTY_AI_HTML, EMPTY_ROUTING_HTML, EMPTY_JOURNEY_HTML, |
| ) |
|
|
|
|
| def _running_bundle(run_id: str = "") -> AppBundle: |
| return ( |
| "### Workflow running…\nThe workflow engine is controlling the approved next steps.", |
| run_id, |
| [], [], [], {}, |
| EMPTY_OUTCOME_HTML, EMPTY_PATH_HTML, EMPTY_AI_HTML, EMPTY_ROUTING_HTML, EMPTY_JOURNEY_HTML, |
| ) |
|
|
|
|
| def _stream_bundle(bundle: AppBundle): |
| run_id = bundle[1] |
| if not run_id: |
| yield ('<div class="activity-card"><div class="activity-badge">STOPPED</div><div class="activity-title">Workflow did not start</div></div>', *bundle) |
| return |
| try: |
| run = STATE_STORE.load(run_id) |
| except Exception: |
| yield ('<div class="activity-card"><div class="activity-badge">STOPPED</div><div class="activity-title">Workflow trace unavailable</div></div>', *bundle) |
| return |
| running = _running_bundle(run_id) |
| for index, event in enumerate(run.events, start=1): |
| yield (activity_html(run, index), *running) |
| time.sleep(playback_delay(event.event_type.value)) |
| yield (activity_html(run, len(run.events), complete=True), *bundle) |
|
|
|
|
| def stream_story_ui(story: str): |
| yield (activity_start_html(story), *_running_bundle()) |
| yield from _stream_bundle(run_story_ui(story)) |
|
|
|
|
| def stream_request_ui( |
| household_id: str, |
| household_type: str, |
| onboarding_notes: str, |
| documents_complete: bool, |
| identity_status: str, |
| relationship_complexity: str, |
| simulation_mode: str, |
| ): |
| yield (activity_start_html("Custom synthetic household"), *_running_bundle()) |
| bundle = run_request_ui( |
| household_id, household_type, onboarding_notes, documents_complete, |
| identity_status, relationship_complexity, simulation_mode, |
| ) |
| yield from _stream_bundle(bundle) |
|
|
|
|
| def build_app() -> gr.Blocks: |
| with gr.Blocks(title="Agent 9 — Wealth Onboarding Workflow", analytics_enabled=False) as demo: |
| gr.HTML(HERO_HTML) |
| gr.HTML(PATTERN_HTML) |
| gr.HTML(_ai_runtime_html()) |
|
|
| gr.Markdown("## Try the onboarding pattern in one click") |
| gr.Markdown("Choose a fictional household and watch the workflow control plane move it through AI assistance, deterministic rules, retry/recovery, and human escalation.") |
| story = gr.Radio(choices=list(STORY_SPECS), value="Harbor Family — straightforward household", label="Choose an onboarding story") |
| story_button = gr.Button("Run this onboarding workflow", variant="primary") |
| live_activity = gr.HTML(EMPTY_ACTIVITY_HTML) |
|
|
| business_outcome = gr.HTML(EMPTY_OUTCOME_HTML) |
| workflow_path = gr.HTML(EMPTY_PATH_HTML) |
| ai_insight = gr.HTML(EMPTY_AI_HTML) |
| routing_explanation = gr.HTML(EMPTY_ROUTING_HTML) |
| business_journey = gr.HTML(EMPTY_JOURNEY_HTML) |
|
|
| with gr.Accordion("Reload a saved run", open=False): |
| gr.Markdown("Every run receives a random ID and is checkpointed in SQLite. Reloading a completed run shows the same state without replaying completed work.") |
| with gr.Row(): |
| current_run_id = gr.Textbox(label="Run ID", placeholder="A random run ID appears here after execution.") |
| refresh_button = gr.Button("Reload Saved Run") |
|
|
| with gr.Tabs(): |
| with gr.Tab("Customize Scenario"): |
| gr.Markdown("### Build your own synthetic onboarding scenario\nThese inputs are fictional and exist only to demonstrate workflow behavior.") |
| household_id = gr.Textbox(label="Household ID", value="HH-DEMO-001") |
| household_type = gr.Dropdown(choices=["INDIVIDUAL", "JOINT", "TRUST", "ENTITY"], value="JOINT", label="Household Type") |
| documents_complete = gr.Checkbox(label="Onboarding documents complete", value=True) |
| onboarding_notes = gr.Textbox(label="Synthetic Onboarding Notes", value="Fictional household seeking a standard advisory relationship. All information in this demo is synthetic.", lines=4) |
| identity_status = gr.Dropdown(choices=["VERIFIED", "REVIEW_REQUIRED"], value="VERIFIED", label="Identity Check Status") |
| relationship_complexity = gr.Dropdown(choices=["STANDARD", "COMPLEX"], value="STANDARD", label="Relationship Complexity") |
| simulation_mode = gr.Dropdown(choices=["NONE", "TRANSIENT_ONCE", "PERMANENT"], value="NONE", label="Failure Simulation") |
| gr.Markdown("**Helpful shortcuts:** TRUST/ENTITY, COMPLEX, missing documents, or REVIEW_REQUIRED demonstrates human review. TRANSIENT_ONCE demonstrates bounded retry recovery.") |
| run_button = gr.Button("Run custom workflow", variant="primary") |
|
|
| with gr.Tab("Engineering State"): |
| gr.Markdown("### Engineering evidence\nInspect exact workflow status, node state, attempt counts, errors, and structured outputs—including the AI Intake Organizer output and its source.") |
| status = gr.Markdown("### Workflow Status\nRun a synthetic household to begin.") |
| node_table = gr.Dataframe(headers=NODE_HEADERS, interactive=False, label="Node Runtime State") |
| final_output = gr.JSON(label="Final Output") |
|
|
| with gr.Tab("Audit Trail"): |
| gr.Markdown("### Why the workflow is auditable\nEvery meaningful control transition is recorded. The audit timeline contains control metadata only; onboarding context is not copied into event details.") |
| event_table = gr.Dataframe(headers=EVENT_HEADERS, interactive=False, label="Append-Only Workflow Events") |
|
|
| with gr.Tab("Human Review"): |
| gr.Markdown("### The human remains in control\nA fictional exception case pauses here. The workflow cannot mark the package ready until a person explicitly decides what should happen.") |
| review_table = gr.Dataframe(headers=REVIEW_HEADERS, interactive=False, label="Human Review History") |
| human_decision = gr.Dropdown(choices=["APPROVE", "REJECT", "RETRY"], value="APPROVE", label="What should the reviewer do?") |
| decision_button = gr.Button("Submit Human Decision", variant="primary") |
| gr.Markdown("**APPROVE** continues the approved exception path. **REJECT** ends the fictional process as `REJECTED`, not `FAILED`. **RETRY** opens a fresh review without replaying completed upstream work. This is illustrative, not any firm's actual regulatory procedure.") |
|
|
| with gr.Tab("Evaluation"): |
| gr.Markdown("### Reliability evidence\nThe same onboarding engine is exercised across success, retry, human review, persistence, failure, and adversarial-control scenarios.") |
| evaluation_summary = gr.Markdown("Run the deterministic evaluation harness to inspect system-level metrics.") |
| evaluation_button = gr.Button("Run 10-Case Evaluation") |
| metric_table = gr.Dataframe(headers=METRIC_HEADERS, interactive=False, label="Evaluation Metrics") |
| case_table = gr.Dataframe(headers=CASE_HEADERS, interactive=False, label="Evaluation Cases") |
|
|
| run_outputs = [ |
| status, current_run_id, node_table, event_table, review_table, final_output, |
| business_outcome, workflow_path, ai_insight, routing_explanation, business_journey, |
| ] |
| stream_outputs = [live_activity, *run_outputs] |
|
|
| story_button.click(fn=stream_story_ui, inputs=[story], outputs=stream_outputs, show_progress="hidden") |
| run_button.click( |
| fn=stream_request_ui, |
| inputs=[household_id, household_type, onboarding_notes, documents_complete, identity_status, relationship_complexity, simulation_mode], |
| outputs=stream_outputs, |
| show_progress="hidden", |
| ) |
| refresh_button.click(fn=refresh_run_ui, inputs=[current_run_id], outputs=run_outputs) |
| decision_button.click(fn=submit_human_decision_ui, inputs=[current_run_id, human_decision], outputs=run_outputs) |
| evaluation_button.click(fn=run_evaluation_ui, outputs=[evaluation_summary, metric_table, case_table]) |
|
|
| return demo |
|
|
|
|
| demo = build_app() |
|
|
| if __name__ == "__main__": |
| demo.queue().launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", "7860")), css=APP_CSS) |
|
|
