{escape(guidance['description'])}
" f"Run the exact same episode loop used in training, inspect each workspace mutation in real time, and compare the deterministic baseline against the OpenRouter-guided RL checkpoint without losing the structure of the task.
Pick a scenario, choose baseline or the OpenRouter-guided trained RL JSON checkpoint, and run a stepwise episode against the same environment used by training and evaluation.
""" ) task = gr.Radio( choices=[ "easy_deadline_extraction", "medium_triage_and_negotiation", "hard_rag_reply", ], value="easy_deadline_extraction", label="Scenario", elem_classes=["control-field"], ) provider = gr.Radio( choices=["baseline", "rl"], value="baseline", label="Policy", elem_classes=["control-field"], ) max_steps = gr.Number( value=12, precision=0, label="Max Steps", elem_classes=["control-field"], ) with gr.Accordion("Provider Settings", open=False): checkpoint_path = gr.Textbox( value=_default_rl_checkpoint(), label="RL Checkpoint Path", elem_classes=["control-field"], ) with gr.Row(): reset = gr.Button("Reset Scenario", variant="secondary") run_episode_btn = gr.Button("Run Episode", variant="primary") gr.HTML( """The RL policy loads the trained JSON checkpoint as guidance, then asks OpenRouter Gemma through the OpenAI client to generate the runtime action. If the model call fails, it falls back to the checkpoint action.
""" ) with gr.Column(scale=5): scenario_brief = gr.HTML(render_scenario_brief("easy_deadline_extraction")) status_card = gr.HTML( build_initial_status( "easy_deadline_extraction", "baseline", _default_rl_checkpoint(), ) ) with gr.Group(elem_classes=["surface-card", "workspace-grid"]): with gr.Tabs(): with gr.Tab("Live Workspace"): with gr.Row(): observation = gr.Code(label="Observation", language="json") summary = gr.Code(label="Run Summary", language="json") with gr.Row(): emails = gr.Dataframe(headers=EMAIL_COLUMNS, label="Unread Emails") todos = gr.Dataframe(headers=TODO_COLUMNS, label="Todos") with gr.Row(): files = gr.Dataframe(headers=FILE_COLUMNS, label="Search Results") action_log = gr.Dataframe(headers=ACTION_LOG_COLUMNS, label="Action Log") with gr.Tab("Episode Trace"): trace_table = gr.Dataframe(headers=TRACE_COLUMNS, label="Episode Trace") reset.click( fn=build_snapshot, inputs=[task], outputs=[observation, emails, todos, files, action_log], ) reset.click( fn=render_scenario_brief, inputs=[task], outputs=[scenario_brief], ) reset.click( fn=build_initial_status, inputs=[task, provider, checkpoint_path], outputs=[status_card], ) provider.change( fn=configure_checkpoint_input, inputs=[provider], outputs=[checkpoint_path], ) provider.change( fn=build_initial_status, inputs=[task, provider, checkpoint_path], outputs=[status_card], ) task.change( fn=render_scenario_brief, inputs=[task], outputs=[scenario_brief], ) task.change( fn=build_initial_status, inputs=[task, provider, checkpoint_path], outputs=[status_card], ) run_episode_btn.click( fn=run_live_episode, inputs=[task, provider, max_steps, checkpoint_path], outputs=[observation, status_card, emails, todos, files, action_log, trace_table, summary], ) demo.load( fn=build_snapshot, inputs=[task], outputs=[observation, emails, todos, files, action_log], ) demo.load( fn=configure_checkpoint_input, inputs=[provider], outputs=[checkpoint_path], ) demo.load( fn=render_scenario_brief, inputs=[task], outputs=[scenario_brief], ) demo.load( fn=build_initial_status, inputs=[task, provider, checkpoint_path], outputs=[status_card], ) api = FastAPI(title="EmailMaestro OpenEnv API") def _task_from_payload(payload: dict | None) -> str: if not payload: return DEFAULT_API_TASK return str( payload.get("task_name") or payload.get("task") or payload.get("task_id") or DEFAULT_API_TASK ) def _reset_openenv(payload: dict | None = None) -> dict[str, object]: global api_env, api_observation task_name = _task_from_payload(payload) api_env = ExecutiveAssistantEnv(task_name=task_name) api_observation = api_env.reset() return { "observation": api_observation.model_dump(), "state": api_env.state(), "info": { "task_name": task_name, "status": api_observation.last_action_status, }, } def _state_openenv() -> dict[str, object]: return { "observation": api_observation.model_dump(), "state": api_env.state(), "info": { "task_name": api_env.task_name, "status": api_observation.last_action_status, }, } def _step_openenv(payload: dict | None = None) -> dict[str, object]: global api_observation action_payload = (payload or {}).get("action", payload or {}) action = AssistantAction.model_validate(action_payload) api_observation, reward, done, info = api_env.step(action) return { "observation": api_observation.model_dump(), "reward": reward.model_dump(), "done": done, "info": info, } @api.get("/health") def health() -> dict[str, str]: return {"status": "ok"} @api.post("/reset") def reset_endpoint(payload: dict | None = None) -> dict[str, object]: return _reset_openenv(payload) @api.post("/openenv/reset") def openenv_reset_endpoint(payload: dict | None = None) -> dict[str, object]: return _reset_openenv(payload) @api.get("/state") def state_endpoint() -> dict[str, object]: return _state_openenv() @api.get("/openenv/state") def openenv_state_endpoint() -> dict[str, object]: return _state_openenv() @api.post("/step") def step_endpoint(payload: dict | None = None) -> dict[str, object]: return _step_openenv(payload) @api.post("/openenv/step") def openenv_step_endpoint(payload: dict | None = None) -> dict[str, object]: return _step_openenv(payload) app = gr.mount_gradio_app(api, demo, path="/", css=APP_CSS) if __name__ == "__main__": uvicorn.run( app, host=APP_RUNTIME.host, port=APP_RUNTIME.port, )