FlyingNunchucks/07-tool-using-agent
0
1from typing import Any, Dict, List2 3import gradio as gr4 5from src.agent import run_agent, run_agent_iter6from src.audit import read_audit_records7from src.demo_presentation import (8 APP_CSS,9 ARCHITECTURE_MARKDOWN,10 AUTHORITY_HTML,11 BUSINESS_CASE_HTML,12 HERO_HTML,13 SECURITY_MARKDOWN,14 TOOL_BELT_HTML,15 idle_activity_html,16 render_activity,17 render_summary,18 starting_activity_html,19)20from src.schemas import AgentRunEvent, AgentRunResult21 22 23def get_run_audit_records(24 call_ids: List[str],25) -> List[Dict[str, Any]]:26 """Return audit records belonging only to the current agent run."""27 28 if not call_ids:29 return []30 31 records = read_audit_records()32 33 matching_records = [34 record35 for record in records36 if record.call.call_id in call_ids37 ]38 39 return [40 record.model_dump(mode="json")41 for record in matching_records42 ]43 44 45def handle_request(user_request: str):46 """Preserve the original programmatic callback contract."""47 48 if not user_request or not user_request.strip():49 return (50 "Please enter a request.",51 [],52 [],53 [],54 )55 56 try:57 result = run_agent(user_request.strip())58 59 tool_calls = [60 call.model_dump(mode="json")61 for call in result.tool_calls62 ]63 tool_results = [64 tool_result.model_dump(mode="json")65 for tool_result in result.tool_results66 ]67 call_ids = [68 call.call_id69 for call in result.tool_calls70 ]71 72 return (73 result.final_answer,74 tool_calls,75 tool_results,76 get_run_audit_records(call_ids),77 )78 79 except Exception:80 return (81 (82 "Agent execution failed. "83 "Please review the server logs for details."84 ),85 [],86 [],87 [],88 )89 90 91def _current_calls(events: list[AgentRunEvent]) -> list[Dict[str, Any]]:92 calls: dict[str, Dict[str, Any]] = {}93 94 for event in events:95 if event.call is not None:96 calls[event.call.call_id] = event.call.model_dump(mode="json")97 98 return list(calls.values())99 100 101def _current_results(events: list[AgentRunEvent]) -> list[Dict[str, Any]]:102 results: dict[str, Dict[str, Any]] = {}103 104 for event in events:105 if event.result is not None:106 results[event.result.call_id] = event.result.model_dump(mode="json")107 108 return list(results.values())109 110 111def _stream_outputs(112 events: list[AgentRunEvent],113 final_result: AgentRunResult | None = None,114 complete: bool = False,115):116 calls = (117 [call.model_dump(mode="json") for call in final_result.tool_calls]118 if final_result is not None119 else _current_calls(events)120 )121 results = (122 [result.model_dump(mode="json") for result in final_result.tool_results]123 if final_result is not None124 else _current_results(events)125 )126 call_ids = [call["call_id"] for call in calls]127 audits = get_run_audit_records(call_ids) if complete else []128 129 answer = (130 final_result.final_answer131 if final_result is not None132 else "*The final business answer will appear after controlled execution completes.*"133 )134 135 return (136 render_activity(events, complete=complete),137 answer,138 render_summary(final_result),139 calls,140 results,141 audits,142 )143 144 145def stream_request(user_request: str):146 """Stream real model/application boundary events into the Gradio demo."""147 148 if not user_request or not user_request.strip():149 yield (150 idle_activity_html(),151 "Please enter a request.",152 render_summary(None),153 [],154 [],155 [],156 )157 return158 159 events: list[AgentRunEvent] = []160 161 yield (162 starting_activity_html(),163 "*The final business answer will appear after controlled execution completes.*",164 render_summary(None),165 [],166 [],167 [],168 )169 170 try:171 for event in run_agent_iter(user_request.strip()):172 events.append(event)173 final_result = event.final_result174 complete = final_result is not None175 176 yield _stream_outputs(177 events=events,178 final_result=final_result,179 complete=complete,180 )181 182 except Exception:183 error_event = AgentRunEvent(184 event="max_rounds_reached",185 message=(186 "The request could not complete. The application stopped the run "187 "without exposing internal server details."188 ),189 )190 events.append(error_event)191 192 yield (193 render_activity(events, complete=True),194 (195 "Agent execution failed. Please review the server logs for details."196 ),197 render_summary(None),198 _current_calls(events),199 _current_results(events),200 [],201 )202 203 204FLAGSHIP_REQUEST = (205 "We may ship equipment to Japan. Find the Electronics items currently in "206 "inventory, calculate an accessory budget of $347 per matching item, and "207 "give me Japan's capital, region, and income classification."208)209 210 211with gr.Blocks(212 title="Governed Tool-Using Agent",213 css=APP_CSS,214) as demo:215 with gr.Column(elem_classes=["agent-shell"]):216 gr.HTML(HERO_HTML)217 gr.HTML(BUSINESS_CASE_HTML)218 219 gr.Markdown("## The approved tool belt")220 gr.HTML(TOOL_BELT_HTML)221 222 gr.Markdown("## Who controls what?")223 gr.HTML(AUTHORITY_HTML)224 225 gr.Markdown("## Try a controlled business request")226 request_box = gr.Textbox(227 label="Operations request",228 value=FLAGSHIP_REQUEST,229 placeholder="Ask for work that may require one or more approved tools.",230 lines=4,231 )232 233 gr.Examples(234 examples=[235 [FLAGSHIP_REQUEST],236 ["What is 347 multiplied by 29?"],237 ["What electronics are currently in inventory?"],238 ["What is the capital, region, and income level of Japan?"],239 [240 (241 "How many Electronics items are in inventory, "242 "and what is 347 multiplied by that number?"243 )244 ],245 ],246 inputs=request_box,247 )248 249 run_button = gr.Button(250 "Run controlled request",251 variant="primary",252 )253 254 activity_output = gr.HTML(idle_activity_html())255 256 with gr.Tabs():257 with gr.Tab("Business Result"):258 gr.Markdown(259 "The answer below is produced only after requested capabilities "260 "pass through the application-controlled boundary."261 )262 final_answer = gr.Markdown(263 "*Run a request to produce a business answer.*"264 )265 summary_output = gr.HTML(render_summary(None))266 267 with gr.Tab("Engineering Audit"):268 gr.Markdown(269 "These are the structured requests, normalized results, and "270 "current-run audit records behind the business-facing view."271 )272 tool_calls_output = gr.JSON(label="Model-proposed Tool Calls")273 tool_results_output = gr.JSON(label="Controlled Tool Results")274 audit_output = gr.JSON(label="Current Run Audit Records")275 276 with gr.Tab("Security & Failure Semantics"):277 gr.Markdown(SECURITY_MARKDOWN)278 279 with gr.Tab("Architecture"):280 gr.Markdown(ARCHITECTURE_MARKDOWN)281 282 run_button.click(283 fn=stream_request,284 inputs=request_box,285 outputs=[286 activity_output,287 final_answer,288 summary_output,289 tool_calls_output,290 tool_results_output,291 audit_output,292 ],293 show_progress="hidden",294 )295 296 297if __name__ == "__main__":298 demo.launch()299