Spaces:
Sleeping
Sleeping
| """Gradio Space: watch a ReAct agent think. Streams every step live to a timeline.""" | |
| import time | |
| import gradio as gr | |
| from agent import config, graph, llm, render, tools | |
| EXAMPLES = [ | |
| "What's the population of the capital of the country that won the most recent FIFA World Cup, divided by 1000?", | |
| "Find when the Eiffel Tower was completed, then calculate how many years ago that was from 2026.", | |
| "What is 17% of the year the first iPhone was released?", | |
| "What is the height of Mount Everest in meters, multiplied by 3?", | |
| ] | |
| HOW_IT_WORKS = """ | |
| This agent runs a **ReAct loop**: it alternates between *reasoning* (🧠 thought), | |
| *acting* (🔧 calling a real tool), and *observing* (📄 the tool's result), until it can answer. | |
| Every step is **streamed live** as it happens — you are watching the model think, not a replay. | |
| When a tool returns a weak or irrelevant result, the agent reconsiders and re-queries; that moment | |
| is marked with **↻ revision**. Backend model is Gemini 2.5 Flash (swappable). | |
| """ | |
| def run_agent(task, runs_used): | |
| task = (task or "").strip() | |
| if not task: | |
| yield render.render_timeline([], status="Enter a task above, then press Run."), "", runs_used | |
| return | |
| allowed, message = render.check_run_allowed(runs_used) | |
| if not allowed: | |
| yield render.render_timeline([], status=message), "", runs_used | |
| return | |
| runs_used += 1 | |
| events, status, tool_calls = [], "starting...", 0 | |
| start = time.perf_counter() | |
| yield render.render_timeline(events, status=status), "", runs_used | |
| try: | |
| client = llm.get_client() | |
| stream = graph.stream_run( | |
| task, client=client, tool_fns=tools.tool_callables(), | |
| declarations=tools.declarations(), system_prompt=config.SYSTEM_PROMPT, | |
| ) | |
| for event in stream: | |
| if event["kind"] == "status": | |
| status = event["text"] | |
| else: | |
| status = None | |
| events.append(event) | |
| if event["kind"] == "tool_call": | |
| tool_calls += 1 | |
| steps = max((e.get("step", 0) for e in events), default=0) | |
| stats = render.render_stats(steps=steps, tool_calls=tool_calls, | |
| seconds=time.perf_counter() - start) | |
| yield render.render_timeline(events, status=status), stats, runs_used | |
| except llm.QuotaExhaustedError as exc: | |
| yield render.render_timeline(events, status=str(exc)), "", runs_used | |
| except Exception as exc: # never leak a stack trace to the user | |
| msg = f"Something went wrong while running the agent ({type(exc).__name__}). Please try again." | |
| yield render.render_timeline(events, status=msg), "", runs_used | |
| def build_theme(): | |
| # Gradio 6.0+ moved theme/css off the Blocks constructor onto launch(). | |
| return gr.themes.Base(primary_hue="emerald", neutral_hue="slate").set( | |
| body_background_fill="#0b0f14", block_background_fill="#121821", | |
| ) | |
| def build_demo(): | |
| with gr.Blocks(title="Observable Agent — Watch It Think") as demo: | |
| gr.Markdown("# 🧠 Observable Agent — *Watch It Think*\nGive it a task and watch the ReAct loop run live.") | |
| runs_state = gr.State(0) | |
| with gr.Row(): | |
| task = gr.Textbox(label="Task", placeholder="Ask something that needs a few steps...", scale=4) | |
| run_btn = gr.Button("Run", variant="primary", scale=1) | |
| gr.Examples(examples=EXAMPLES, inputs=task, label="Example tasks") | |
| stats = gr.Markdown("", elem_id="oa-stats") | |
| timeline = gr.HTML(render.render_timeline([], status="Idle. Pick an example or type a task.")) | |
| with gr.Accordion("How this works", open=False): | |
| gr.Markdown(HOW_IT_WORKS) | |
| run_btn.click(run_agent, inputs=[task, runs_state], outputs=[timeline, stats, runs_state]) | |
| task.submit(run_agent, inputs=[task, runs_state], outputs=[timeline, stats, runs_state]) | |
| return demo | |
| if __name__ == "__main__": | |
| build_demo().launch(theme=build_theme(), css=render.TIMELINE_CSS) | |