# User Experience UX patterns for streaming, error handling, and traceability. ## Real-time Streaming ### Agent Progress Display The UI shows agent activity in real-time using a Claude Code-inspired format: ``` ● 🎯 **Triage Agent** ├── Geocoding Address("Broadway and 42nd") │ ⎿ ✓ Manhattan, NYC ● 🔍 **Research Agent** ├── Looking Up City Records... │ ⎿ RD-MN-0042, 3 complaints ├── Checking Nearby Reports... │ ⎿ 2 reports (1 open) ├── Getting Weather... │ ⎿ 45°F, Clear ● 📋 **Report Agent** ├── Getting Department Info... │ ⎿ DOT (24-48h response) ├── Generating PDF Report... │ ⎿ 📄 FMN-20251129 (12.5KB) `8.3s` ``` **Implementation** (`app.py:222-277`): ```python def build_inline_progress(agent_states: dict, elapsed: float, thinking: str = "", tool_history: list = None) -> str: """Build multi-agent style progress display.""" lines = [] for tool in tool_history: agent = tool.get("agent") name = tool.get("display_name") result = tool.get("result_summary", "") # Agent header if agent != current_agent: lines.append(f"● {icon} **{agent_name}**") # Tool call line lines.append(f" ├── {name}...") # Result line if result: lines.append(f" │ ⎿ {result}") return "\n".join(lines) ``` ### Event-Driven Updates Events flow from agents to UI via generator: ```python # In controller.py for event_type, data in orchestrator.process(message, image_analysis): if event_type == "tool_call": # Update UI immediately yield build_outputs() elif event_type == "agent_done": # Show completion yield build_outputs() elif event_type == "complete": # Final response yield build_outputs() ``` ### Result Summarization Tool results are summarized for display (`app.py:68-198`): | Tool | Raw Output | Display Summary | |------|------------|-----------------| | `validate_address` | `{"is_valid_nyc": true, "borough": "Manhattan", ...}` | `✓ Manhattan` | | `weather_get_current` | `{"temp_f": 45, "conditions": "Clear", ...}` | `45°F, Clear` | | `get_nearby_reports` | `{"total_reports": 2, "open_reports": 1, ...}` | `2 reports (1 open)` | | `pdf_generate_report` | `{"report_id": "FMN-...", "size_kb": 12.5, ...}` | `📄 FMN-... (12.5KB)` | ## Reasoning Trace ### Visible Thinking The Reasoning Trace panel shows the agent's thought process: ``` **🧠 Autonomous Reasoning Trace** [0.0s] 💭 Received infrastructure report request [0.2s] 💭 Delegating full control to Controller agent [1.5s] 💭 Controller delegated to Triage Agent [2.3s] 🔧 Agent calling: Geocoding Address [3.1s] 👁️ Tool result: {"is_valid_nyc": true... [3.2s] ✅ Triage Agent completed (1.7s) [confidence: 0.85] [3.3s] 💭 Controller delegated to Research Agent ... [8.0s] ✅ Presenting Controller's response to user [confidence: 0.92] --- **Final Quality Score:** 8.7/10 ``` **Implementation** (`agents/reasoning.py`): ```python @dataclass class ReasoningTrace: thoughts: List[Dict[str, Any]] = field(default_factory=list) def think(self, thought: str, category: str = "reasoning") -> None: """Record a thought.""" self.thoughts.append({ "type": "thought", "category": category, "content": thought, "timestamp": time.time() - self.start_time, }) def to_display(self) -> str: """Format for UI display.""" lines = [] for t in self.thoughts: icon = {"thought": "💭", "decision": "✅", "observation": "👁️"} lines.append(f"[{t['timestamp']:.1f}s] {icon} {t['content']}") return "\n".join(lines) ``` ### Self-Evaluation Display Quality scores shown after completion: ```python # In controller.py quality_score = self._evaluate_response_quality(response, agent_states, trace) # In app.py if quality_score is not None: current_reasoning += f"\n\n**Final Quality Score:** {quality_score * 10:.1f}/10" ``` ## Error Handling ### Friendly Error Messages Technical errors are mapped to user-friendly messages: | Error Type | User Sees | |------------|-----------| | Rate limit | "You're sending requests too quickly. Please wait 30 seconds." | | API timeout | "We're having trouble connecting. Please try again." | | Invalid input | "Please provide a valid NYC address." | | MCP server down | "The infrastructure tools server is temporarily unavailable." | **Implementation** (`security/error_handler.py`): ```python class ErrorHandler: def handle(self, error: Exception, context: str = "") -> FriendlyError: if isinstance(error, RateLimitExceeded): return FriendlyError( message="You're sending requests too quickly.", suggestion=f"Please wait {error.wait_seconds} seconds.", is_recoverable=True, ) ``` ### Graceful Degradation If services are unavailable: ```python # MCP server check mcp_available, mcp_error = self._check_mcp_server() if not mcp_available: yield ("needs_info", { "message": "⚠️ **Service Temporarily Unavailable**\n\n" "The NYC infrastructure tools server is offline.\n\n" "Please wait a moment and try again." }) return ``` ### Input Validation Feedback Clear feedback for invalid input: ```python if not validation_result.is_valid: error_msg = validation_result.get_friendly_message() # "Please provide a description of the infrastructure issue." # "Your message is too long. Please limit to 5000 characters." ``` ## Multi-turn Conversation ### Context Awareness The Controller sees full conversation history: ```python # In prompts.py CONVERSATION HISTORY (you have full context): User: pothole on my street Agent: I'd be happy to help! To file a report, I need the street address... User: Broadway and 42nd St ``` ### Follow-up Questions Controller asks for missing information naturally: ``` User: "There's a pothole" Agent: "I'd be happy to help you report that pothole! To file an accurate report with NYC DOT, I need a bit more information: **What's the street address?** (e.g., "Broadway and 42nd St" or "123 Main Street, Brooklyn") Once you provide the location, I can look up the city records and generate an official report." ``` ## Map Integration ### Real-time Location Updates Map updates as soon as location is validated: ```python elif event_type == "location_update": lat = data.get("lat") lon = data.get("lon") if lat and lon: current_map = create_issue_map( lat=lat, lon=lon, borough=data.get("borough"), address=data.get("address") ) yield build_outputs() ``` ### Folium Map Display Interactive map with issue marker: ```python # In ui/mapping.py def create_issue_map(lat, lon, borough, address): m = folium.Map(location=[lat, lon], zoom_start=15) folium.Marker( [lat, lon], popup=f"{address}
{borough}", icon=folium.Icon(color="red", icon="exclamation-sign") ).add_to(m) folium.Circle( [lat, lon], radius=50, color="red", fill=True, ).add_to(m) return m._repr_html_() ``` ## Agent Pipeline Display ### Status Cards Each agent has a status card showing: ``` 🔍 **Research Agent** ✅ 3.2s *Classifies issue & gathers location data* - ✅ Looking Up City Records (1.2s) - ✅ Checking Nearby Reports (0.8s) - ✅ Getting Weather (1.1s) ``` ### Completion Summary Final response includes execution summary: ``` **🤖 Agent Execution** ● 🎯 **Triage Agent** └─ Geocoding Address → ✓ Broadway & 42nd, Manhattan ● 🔍 **Research Agent** └─ Looking Up City Records → RD-MN-0042, 3 complaints └─ Checking Nearby Reports → 2 reports (1 open) └─ Getting Weather → 45°F, Clear ● 📋 **Report Agent** └─ Getting Department Info → DOT (24-48h response) └─ Generating PDF Report → 📄 FMN-20251129 *Completed in 8.3s* --- ## Infrastructure Report **Report ID:** FMN-20251129-001 **Issue Type:** Pothole **Location:** Broadway & 42nd St, Manhattan ... ``` ## Accessibility ### Clear Visual Hierarchy - Agent names in **bold** - Tool calls with tree-style indentation - Results with checkmarks/icons - Timing in subtle `code` format ### Status Icons | Icon | Meaning | |------|---------| | ⌛ | Pending | | ⏳ | Running | | ✅ | Completed | | ❌ | Error | | 📄 | PDF generated | | ✉️ | Email sent |