Spaces:
Running
Running
| """ | |
| LifeFlow AI - Demo | |
| ✅ Agent status dynamically updated (use tool, task completed, etc.) | |
| ✅ Automatically switches to the Trip Summary Tab (full report + map) after planning is complete | |
| ✅ Perfectly compatible with Light/Dark themes | |
| """ | |
| import gradio as gr | |
| import plotly.graph_objects as go | |
| from datetime import datetime, time | |
| import json | |
| from typing import Dict, List, Optional, Tuple | |
| import time as time_module | |
| class LifeFlowInteractive: | |
| """LifeFlow AI Interactive Version v9""" | |
| def __init__(self): | |
| self.team = None | |
| self.current_step = 0 | |
| self.task_list = [] | |
| self.poi_results = [] | |
| self.chat_history = [] | |
| self.reasoning_messages = [] | |
| self.planning_completed = False | |
| # Settings | |
| self.settings = { | |
| 'google_maps_api_key': '', | |
| 'openweather_api_key': '', | |
| 'anthropic_api_key': '', | |
| 'model': 'claude-sonnet-4-20250514' | |
| } | |
| # Agent information | |
| self.agents_info = { | |
| "planner": { | |
| "name": "Planner", | |
| "avatar": "👔", | |
| "role": "Chief Planner", | |
| "color": "#4A90E2" | |
| }, | |
| "scout": { | |
| "name": "Scout", | |
| "avatar": "🕵️", | |
| "role": "POI Investigation Expert", | |
| "color": "#50C878" | |
| }, | |
| "optimizer": { | |
| "name": "Optimizer", | |
| "avatar": "🤖", | |
| "role": "Route Optimization Engine", | |
| "color": "#F5A623" | |
| }, | |
| "validator": { | |
| "name": "Validator", | |
| "avatar": "🛡️", | |
| "role": "Quality Assurance Expert", | |
| "color": "#7ED321" | |
| }, | |
| "weather": { | |
| "name": "Weather", | |
| "avatar": "🌈", | |
| "role": "Weather Analyst", | |
| "color": "#50E3C2" | |
| }, | |
| "traffic": { | |
| "name": "Traffic", | |
| "avatar": "🚗", | |
| "role": "Traffic Planner", | |
| "color": "#FF6B6B" | |
| } | |
| } | |
| def create_agent_card(self, agent_key: str, status: str = "idle", message: str = ""): | |
| """Create Agent card (theme adaptive, no progress bar)""" | |
| agent = self.agents_info[agent_key] | |
| status_icons = { | |
| "idle": "⚪", | |
| "thinking": "💭", | |
| "using_tool": "🔧", | |
| "working": "⚙️", | |
| "completed": "✅", | |
| "waiting": "⏸️", | |
| "error": "❌" | |
| } | |
| icon = status_icons.get(status, "⚪") | |
| breathing_class = "breathing" if status in ["working", "using_tool"] else "" | |
| return f""" | |
| <div class="{breathing_class}" style=" | |
| background: var(--background-fill-secondary); | |
| border-left: 4px solid {agent['color']}; | |
| border-radius: 8px; | |
| padding: 12px; | |
| margin: 8px 0; | |
| transition: all 0.3s ease; | |
| "> | |
| <div style="display: flex; align-items: center; gap: 10px;"> | |
| <span style="font-size: 1.8em;">{agent['avatar']}</span> | |
| <div style="flex: 1;"> | |
| <div style="font-size: 1.1em; font-weight: bold; color: var(--body-text-color);"> | |
| {agent['name']} | |
| </div> | |
| <div style="font-size: 0.85em; color: var(--body-text-color); opacity: 0.8; margin-top: 2px;"> | |
| {icon} {message} | |
| </div> | |
| </div> | |
| </div> | |
| </div> | |
| """ | |
| def add_reasoning_message(self, agent: str, message: str, msg_type: str = "info"): | |
| """Add reasoning message to record""" | |
| agent_info = self.agents_info.get(agent, {}) | |
| avatar = agent_info.get('avatar', '🤖') | |
| name = agent_info.get('name', agent) | |
| timestamp = datetime.now().strftime("%H:%M:%S") | |
| # Choose style based on type | |
| if msg_type == "tool": | |
| icon = "🔧" | |
| bg_color = "var(--color-accent-soft)" | |
| elif msg_type == "success": | |
| icon = "✅" | |
| bg_color = "var(--background-fill-secondary)" | |
| elif msg_type == "thinking": | |
| icon = "💭" | |
| bg_color = "var(--background-fill-secondary)" | |
| else: | |
| icon = "ℹ️" | |
| bg_color = "var(--background-fill-secondary)" | |
| msg_html = f""" | |
| <div style="margin: 8px 0; padding: 10px; background: {bg_color}; border-radius: 8px; border-left: 3px solid {agent_info.get('color', '#4A90E2')};"> | |
| <div style="display: flex; align-items: center; gap: 8px; margin-bottom: 5px;"> | |
| <span style="font-size: 1.2em;">{avatar}</span> | |
| <span style="font-weight: bold; color: var(--body-text-color);">{name}</span> | |
| <span style="opacity: 0.6; font-size: 0.85em; color: var(--body-text-color);">{timestamp}</span> | |
| <span>{icon}</span> | |
| </div> | |
| <div style="color: var(--body-text-color); padding-left: 35px;"> | |
| {message} | |
| </div> | |
| </div> | |
| """ | |
| self.reasoning_messages.append(msg_html) | |
| def get_reasoning_html(self): | |
| """Get HTML of all reasoning messages (scrollable)""" | |
| if not self.reasoning_messages: | |
| return """ | |
| <div style="text-align: center; padding: 40px; opacity: 0.6; color: var(--body-text-color);"> | |
| <p>AI conversation logs will be displayed here</p> | |
| </div> | |
| """ | |
| # Combine all messages and add auto-scroll to bottom script | |
| messages_html = "".join(self.reasoning_messages) | |
| return f""" | |
| <div id="reasoning-container" style="max-height: 600px; overflow-y: auto; padding: 10px;"> | |
| {messages_html} | |
| </div> | |
| <script> | |
| // Auto-scroll to bottom | |
| var container = document.getElementById('reasoning-container'); | |
| if (container) {{ | |
| container.scrollTop = container.scrollHeight; | |
| }} | |
| </script> | |
| """ | |
| def create_task_list_ui(self, tasks: List[Dict]): | |
| """Create task list UI (theme adaptive)""" | |
| if not tasks: | |
| return "<p style='color: var(--body-text-color); opacity: 0.6;'>No tasks yet</p>" | |
| priority_colors = { | |
| 'HIGH': '#ff4444', | |
| 'MEDIUM': '#FFA500', | |
| 'LOW': '#4A90E2' | |
| } | |
| html = f""" | |
| <div class="slide-in" style="background: var(--background-fill-secondary); border-radius: 12px; padding: 20px; margin: 15px 0;"> | |
| <h3 style="color: var(--body-text-color); margin-top: 0;">📋 Identified Task List</h3> | |
| """ | |
| for i, task in enumerate(tasks): | |
| priority = task.get('priority', 'MEDIUM') | |
| color = priority_colors.get(priority, '#4A90E2') | |
| html += f""" | |
| <div style="background: var(--background-fill-primary); border-left: 4px solid {color}; border-radius: 8px; | |
| padding: 15px; margin: 10px 0;"> | |
| <div style="font-size: 1.2em; color: var(--body-text-color); font-weight: bold;"> | |
| {i + 1}. {task['description']} | |
| </div> | |
| <div style="color: var(--body-text-color); opacity: 0.7; margin-top: 5px;"> | |
| <span style="background: {color}; color: white; padding: 3px 10px; border-radius: 12px; | |
| font-size: 0.85em; margin-right: 10px;"> | |
| {priority} | |
| </span> | |
| <span>📁 {task.get('category', 'Other')}</span> | |
| {f" | ⏰ {task.get('time_window', 'Anytime')}" if task.get('time_window') else ""} | |
| </div> | |
| </div> | |
| """ | |
| html += f""" | |
| <div style="margin-top: 20px; padding: 15px; background: var(--color-accent-soft); border-radius: 8px;"> | |
| <div style="font-size: 1.1em; color: var(--body-text-color); font-weight: bold;">📊 Overview</div> | |
| <div style="color: var(--body-text-color); opacity: 0.9; margin-top: 5px;"> | |
| Total tasks: {len(tasks)} | | |
| High priority: {sum(1 for t in tasks if t.get('priority') == 'HIGH')} | | |
| Medium priority: {sum(1 for t in tasks if t.get('priority') == 'MEDIUM')} | | |
| Low priority: {sum(1 for t in tasks if t.get('priority') == 'LOW')} | |
| </div> | |
| </div> | |
| </div> | |
| """ | |
| return html | |
| def create_map(self, route_data: Optional[Dict] = None) -> go.Figure: | |
| """Create route map""" | |
| fig = go.Figure() | |
| if route_data and 'stops' in route_data: | |
| stops = route_data['stops'] | |
| # Extract coordinates | |
| lats = [stop['location']['lat'] for stop in stops] | |
| lons = [stop['location']['lng'] for stop in stops] | |
| names = [f"{i + 1}. {stop['poi_name']}" for i, stop in enumerate(stops)] | |
| # Add route lines | |
| fig.add_trace(go.Scattermapbox( | |
| lat=lats, | |
| lon=lons, | |
| mode='markers+lines+text', | |
| marker=dict( | |
| size=15, | |
| color=['green'] + ['blue'] * (len(stops) - 2) + ['red'] if len(stops) > 2 else ['green', 'red'], | |
| ), | |
| line=dict(width=3, color='blue'), | |
| text=names, | |
| textposition="top center", | |
| name="Route" | |
| )) | |
| # Set map center | |
| center_lat = sum(lats) / len(lats) | |
| center_lon = sum(lons) / len(lons) | |
| else: | |
| # Default center (Taipei) | |
| center_lat, center_lon = 25.0330, 121.5654 | |
| # Map style | |
| fig.update_layout( | |
| mapbox=dict( | |
| style="open-street-map", | |
| center=dict(lat=center_lat, lon=center_lon), | |
| zoom=12 | |
| ), | |
| showlegend=False, | |
| height=500, | |
| margin=dict(l=0, r=0, t=0, b=0), | |
| paper_bgcolor='rgba(0,0,0,0)', | |
| plot_bgcolor='rgba(0,0,0,0)' | |
| ) | |
| return fig | |
| def step1_analyze_tasks(self, user_input: str): | |
| """Step 1: Analyze user requirements and extract tasks""" | |
| self.reasoning_messages = [] # Clear previous messages | |
| self.task_list = [] | |
| self.planning_completed = False | |
| # Planner Agent start analysis | |
| self.add_reasoning_message("planner", "Starting to analyze user requirements...", "thinking") | |
| time_module.sleep(0.5) | |
| # Simulate Planner using tools | |
| self.add_reasoning_message("planner", "Using tool: parse_requirements()", "tool") | |
| time_module.sleep(0.3) | |
| # Extract tasks (simulated) | |
| if "hospital" in user_input.lower() or "doctor" in user_input.lower() or "醫院" in user_input or "看病" in user_input: | |
| self.task_list.append({ | |
| 'description': 'Go to hospital for medical consultation', | |
| 'category': 'Medical', | |
| 'priority': 'HIGH', | |
| 'estimated_duration': 45, | |
| 'time_window': '08:00-12:00' | |
| }) | |
| if "supermarket" in user_input.lower() or "shopping" in user_input.lower() or "grocery" in user_input.lower() or "超市" in user_input or "買菜" in user_input or "購物" in user_input: | |
| self.task_list.append({ | |
| 'description': 'Go to supermarket for shopping', | |
| 'category': 'Shopping', | |
| 'priority': 'MEDIUM', | |
| 'estimated_duration': 30, | |
| 'time_window': 'Anytime' | |
| }) | |
| if "post office" in user_input.lower() or "mail" in user_input.lower() or "郵局" in user_input or "寄包裹" in user_input: | |
| self.task_list.append({ | |
| 'description': 'Go to post office to mail package', | |
| 'category': 'Postal', | |
| 'priority': 'HIGH', | |
| 'estimated_duration': 20, | |
| 'time_window': '09:00-15:00' | |
| }) | |
| # If no tasks identified, add default | |
| if not self.task_list: | |
| self.task_list.append({ | |
| 'description': 'Complete errands', | |
| 'category': 'Other', | |
| 'priority': 'MEDIUM', | |
| 'estimated_duration': 30, | |
| 'time_window': 'Anytime' | |
| }) | |
| self.add_reasoning_message("planner", | |
| f"Analysis complete! Identified {len(self.task_list)} tasks<br>" + | |
| "<br>".join( | |
| [f"• {t['description']} (Priority: {t['priority']})" for t in self.task_list]), | |
| "success") | |
| # Generate task list and summary | |
| task_list_html = self.create_task_list_ui(self.task_list) | |
| task_summary = f""" | |
| <div style="background: var(--background-fill-secondary); border-radius: 12px; padding: 20px; margin: 15px 0;"> | |
| <h3 style="color: var(--body-text-color);">📊 Task Summary</h3> | |
| <div style="color: var(--body-text-color); opacity: 0.9; line-height: 1.8;"> | |
| <p><strong>Total tasks:</strong> {len(self.task_list)}</p> | |
| <p><strong>High priority tasks:</strong> {sum(1 for t in self.task_list if t.get('priority') == 'HIGH')}</p> | |
| <p><strong>Estimated total time:</strong> {sum(t.get('estimated_duration', 0) for t in self.task_list)} minutes</p> | |
| </div> | |
| </div> | |
| """ | |
| # Update Agent status | |
| agent_updates = [ | |
| self.create_agent_card("planner", "completed", "Task analysis complete ✓"), | |
| self.create_agent_card("scout", "idle", "Waiting for confirmation"), | |
| self.create_agent_card("optimizer", "idle", "Waiting for confirmation"), | |
| self.create_agent_card("validator", "idle", "Waiting for confirmation"), | |
| self.create_agent_card("weather", "idle", "Waiting for confirmation"), | |
| self.create_agent_card("traffic", "idle", "Waiting for confirmation") | |
| ] | |
| return ( | |
| *agent_updates, | |
| self.get_reasoning_html(), | |
| task_list_html, | |
| task_summary, | |
| gr.update(visible=True), # Show task confirmation area | |
| gr.update(visible=True), # Show chat modification area | |
| [], # Clear chat history | |
| "Task analysis complete, please confirm or modify" | |
| ) | |
| def modify_task_chat(self, user_message: str, chat_history: List): | |
| """Modify tasks through chat""" | |
| if not user_message.strip(): | |
| return chat_history, self.create_task_list_ui(self.task_list) | |
| # Add user message | |
| chat_history.append((user_message, None)) | |
| # Simulate AI response | |
| time_module.sleep(0.3) | |
| # Simple keyword matching modification logic | |
| if "priority" in user_message.lower() or "優先" in user_message: | |
| if any(str(i) in user_message for i in range(1, len(self.task_list) + 1)): | |
| task_num = next(int(str(i)) for i in range(1, len(self.task_list) + 1) if str(i) in user_message) | |
| if 0 < task_num <= len(self.task_list): | |
| if "high" in user_message.lower() or "高" in user_message: | |
| self.task_list[task_num - 1]['priority'] = 'HIGH' | |
| response = f"Understood! Changed task {task_num} to HIGH priority." | |
| elif "low" in user_message.lower() or "低" in user_message: | |
| self.task_list[task_num - 1]['priority'] = 'LOW' | |
| response = f"Understood! Changed task {task_num} to LOW priority." | |
| else: | |
| self.task_list[task_num - 1]['priority'] = 'MEDIUM' | |
| response = f"Understood! Changed task {task_num} to MEDIUM priority." | |
| else: | |
| response = "Task number not found, please check again." | |
| else: | |
| response = "Please specify which task number you want to modify." | |
| elif "delete" in user_message.lower() or "remove" in user_message.lower() or "刪除" in user_message: | |
| if any(str(i) in user_message for i in range(1, len(self.task_list) + 1)): | |
| task_num = next(int(str(i)) for i in range(1, len(self.task_list) + 1) if str(i) in user_message) | |
| if 0 < task_num <= len(self.task_list): | |
| deleted_task = self.task_list.pop(task_num - 1) | |
| response = f"Understood! Deleted task: {deleted_task['description']}" | |
| else: | |
| response = "Task number not found, please check again." | |
| else: | |
| response = "Please specify which task number you want to delete." | |
| else: | |
| response = "I understand. You can say: 'Change task 2 to high priority' or 'Delete task 3'" | |
| # Add AI response | |
| chat_history[-1] = (user_message, response) | |
| # Update task list | |
| updated_task_list = self.create_task_list_ui(self.task_list) | |
| return chat_history, updated_task_list | |
| def step2_search_pois(self): | |
| """Step 2: Search POI""" | |
| # Scout Agent start searching | |
| self.add_reasoning_message("scout", "Starting POI search...", "thinking") | |
| agent_updates = [ | |
| self.create_agent_card("planner", "completed", "Task analysis complete ✓"), | |
| self.create_agent_card("scout", "working", "Searching for POIs..."), | |
| self.create_agent_card("optimizer", "waiting", "Waiting for search completion"), | |
| self.create_agent_card("validator", "waiting", "Waiting for search completion"), | |
| self.create_agent_card("weather", "idle", "Standby"), | |
| self.create_agent_card("traffic", "idle", "Standby") | |
| ] | |
| yield ( | |
| *agent_updates, | |
| self.get_reasoning_html(), | |
| "Searching for POIs..." | |
| ) | |
| time_module.sleep(0.5) | |
| # Simulate searching for each task | |
| for i, task in enumerate(self.task_list): | |
| self.add_reasoning_message("scout", | |
| f"Using tool: search_poi(category='{task['category']}')", | |
| "tool") | |
| time_module.sleep(0.3) | |
| # Simulate finding POI | |
| poi = { | |
| 'name': f"{task['category']} - Best Choice", | |
| 'rating': 4.5 + (i * 0.1), | |
| 'distance': 800 + (i * 200), | |
| 'category': task['category'] | |
| } | |
| self.poi_results.append(poi) | |
| self.add_reasoning_message("scout", | |
| f"Found POI: {poi['name']}<br>Rating: {poi['rating']}★ | Distance: {poi['distance']}m", | |
| "success") | |
| yield ( | |
| *agent_updates, | |
| self.get_reasoning_html(), | |
| f"Found {i + 1}/{len(self.task_list)} POIs..." | |
| ) | |
| # Search complete | |
| self.add_reasoning_message("scout", | |
| f"POI search complete! Found {len(self.poi_results)} suitable locations", | |
| "success") | |
| agent_updates[1] = self.create_agent_card("scout", "completed", "POI search complete ✓") | |
| yield ( | |
| *agent_updates, | |
| self.get_reasoning_html(), | |
| "POI search complete" | |
| ) | |
| def step3_optimize_route(self): | |
| """Step 3: Optimize route""" | |
| # Optimizer Agent start optimizing | |
| self.add_reasoning_message("optimizer", "Starting route optimization...", "thinking") | |
| agent_updates = [ | |
| self.create_agent_card("planner", "completed", "Task analysis complete ✓"), | |
| self.create_agent_card("scout", "completed", "POI search complete ✓"), | |
| self.create_agent_card("optimizer", "working", "Optimizing route..."), | |
| self.create_agent_card("validator", "waiting", "Waiting for optimization"), | |
| self.create_agent_card("weather", "idle", "Standby"), | |
| self.create_agent_card("traffic", "idle", "Standby") | |
| ] | |
| yield ( | |
| *agent_updates, | |
| self.get_reasoning_html(), | |
| "", # trip_summary_display | |
| None, # map_output | |
| gr.update(visible=False), # map_tab | |
| "Optimizing route..." | |
| ) | |
| time_module.sleep(0.5) | |
| # Call TSPTW solver | |
| self.add_reasoning_message("optimizer", | |
| "Using tool: optimize_route_with_time_windows()", | |
| "tool") | |
| time_module.sleep(0.8) | |
| # Generate simulated optimized route | |
| route_data = { | |
| 'stops': [] | |
| } | |
| base_lat, base_lng = 25.0330, 121.5654 | |
| for i, (task, poi) in enumerate(zip(self.task_list, self.poi_results)): | |
| stop = { | |
| 'order': i + 1, | |
| 'poi_name': poi['name'], | |
| 'description': task['description'], | |
| 'location': { | |
| 'lat': base_lat + (i * 0.01), | |
| 'lng': base_lng + (i * 0.01) | |
| }, | |
| 'arrival_time': f"{9 + i}:{'00' if i == 0 else '15'}", | |
| 'departure_time': f"{9 + i}:{45 if i == 0 else 45}", | |
| 'duration': task['estimated_duration'], | |
| 'priority': task['priority'] | |
| } | |
| route_data['stops'].append(stop) | |
| self.add_reasoning_message("optimizer", | |
| f"Route optimization complete!<br>Total stops: {len(route_data['stops'])} | Estimated total time: {sum(t.get('estimated_duration', 0) for t in self.task_list)} minutes", | |
| "success") | |
| agent_updates[2] = self.create_agent_card("optimizer", "completed", "Route optimization complete ✓") | |
| agent_updates[3] = self.create_agent_card("validator", "working", "Validating feasibility...") | |
| yield ( | |
| *agent_updates, | |
| self.get_reasoning_html(), | |
| "", # trip_summary_display | |
| None, # map_output | |
| gr.update(visible=False), # map_tab | |
| "Route optimization complete, validating..." | |
| ) | |
| time_module.sleep(0.5) | |
| # Validator validation | |
| self.add_reasoning_message("validator", "Starting feasibility validation...", "thinking") | |
| time_module.sleep(0.3) | |
| self.add_reasoning_message("validator", | |
| "Using tool: validate_feasibility()", | |
| "tool") | |
| time_module.sleep(0.5) | |
| self.add_reasoning_message("validator", | |
| "Validation complete! All constraints satisfied ✓<br>• Time window compliance: ✓<br>• Priority task completion: ✓<br>• Deadline met: ✓", | |
| "success") | |
| agent_updates[3] = self.create_agent_card("validator", "completed", "Feasibility validation complete ✓") | |
| # Generate complete trip summary | |
| trip_summary_html = self.create_trip_summary(route_data) | |
| # Generate map | |
| map_fig = self.create_map(route_data) | |
| self.planning_completed = True | |
| yield ( | |
| *agent_updates, | |
| self.get_reasoning_html(), | |
| trip_summary_html, | |
| map_fig, | |
| gr.update(visible=True), # Show map tab | |
| "✅ Trip planning complete!" | |
| ) | |
| def create_trip_summary(self, route_data: Dict) -> str: | |
| """Create trip summary report""" | |
| stops = route_data.get('stops', []) | |
| html = f""" | |
| <div style="background: var(--background-fill-secondary); border-radius: 12px; padding: 20px; margin: 15px 0;"> | |
| <h2 style="color: var(--body-text-color); margin-top: 0;">🎉 Trip Planning Complete</h2> | |
| <div style="background: var(--color-accent-soft); border-radius: 8px; padding: 15px; margin: 15px 0;"> | |
| <h3 style="color: var(--body-text-color); margin-top: 0;">📊 Overview</h3> | |
| <div style="color: var(--body-text-color); opacity: 0.9; line-height: 1.8;"> | |
| <p><strong>Total stops:</strong> {len(stops)}</p> | |
| <p><strong>Total time:</strong> {sum(s.get('duration', 0) for s in stops)} minutes</p> | |
| <p><strong>Start time:</strong> {stops[0]['arrival_time'] if stops else 'N/A'}</p> | |
| <p><strong>Estimated completion:</strong> {stops[-1]['departure_time'] if stops else 'N/A'}</p> | |
| </div> | |
| </div> | |
| <h3 style="color: var(--body-text-color);">📍 Detailed Itinerary</h3> | |
| """ | |
| priority_colors = { | |
| 'HIGH': '#ff4444', | |
| 'MEDIUM': '#FFA500', | |
| 'LOW': '#4A90E2' | |
| } | |
| for stop in stops: | |
| color = priority_colors.get(stop.get('priority', 'MEDIUM'), '#4A90E2') | |
| html += f""" | |
| <div style="background: var(--background-fill-primary); border-left: 4px solid {color}; | |
| border-radius: 8px; padding: 15px; margin: 10px 0;"> | |
| <div style="display: flex; justify-content: space-between; align-items: start;"> | |
| <div style="flex: 1;"> | |
| <div style="font-size: 1.3em; font-weight: bold; color: var(--body-text-color);"> | |
| {stop['order']}. {stop['poi_name']} | |
| </div> | |
| <div style="color: var(--body-text-color); opacity: 0.8; margin-top: 5px;"> | |
| {stop['description']} | |
| </div> | |
| </div> | |
| <div style="text-align: right;"> | |
| <span style="background: {color}; color: white; padding: 4px 12px; | |
| border-radius: 12px; font-size: 0.85em;"> | |
| {stop.get('priority', 'MEDIUM')} | |
| </span> | |
| </div> | |
| </div> | |
| <div style="color: var(--body-text-color); opacity: 0.7; margin-top: 10px; | |
| display: flex; gap: 15px; flex-wrap: wrap;"> | |
| <span>🕐 Arrival: {stop['arrival_time']}</span> | |
| <span>🕐 Departure: {stop['departure_time']}</span> | |
| <span>⏱️ Duration: {stop['duration']} min</span> | |
| </div> | |
| </div> | |
| """ | |
| html += """ | |
| <div style="background: var(--color-accent-soft); border-radius: 8px; padding: 15px; margin-top: 20px;"> | |
| <h3 style="color: var(--body-text-color); margin-top: 0;">✅ Validation Results</h3> | |
| <div style="color: var(--body-text-color); opacity: 0.9; line-height: 1.8;"> | |
| <p>✓ All time windows satisfied</p> | |
| <p>✓ Deadline met</p> | |
| <p>✓ Priority tasks completed</p> | |
| <p>✓ Route feasibility: 100%</p> | |
| </div> | |
| </div> | |
| </div> | |
| """ | |
| return html | |
| def save_settings(self, google_key, weather_key, anthropic_key, model): | |
| """Save settings""" | |
| self.settings['google_maps_api_key'] = google_key | |
| self.settings['openweather_api_key'] = weather_key | |
| self.settings['anthropic_api_key'] = anthropic_key | |
| self.settings['model'] = model | |
| return "✅ Settings saved successfully" | |
| def build_interface(self): | |
| """Build Gradio interface""" | |
| with gr.Blocks( | |
| title="LifeFlow AI - Intelligent Trip Planning", | |
| theme=gr.themes.Soft(), | |
| css=""" | |
| .breathing { | |
| animation: breathing 2s ease-in-out infinite; | |
| } | |
| @keyframes breathing { | |
| 0%, 100% { opacity: 1; } | |
| 50% { opacity: 0.7; } | |
| } | |
| .slide-in { | |
| animation: slideIn 0.5s ease-out; | |
| } | |
| @keyframes slideIn { | |
| from { | |
| opacity: 0; | |
| transform: translateY(-20px); | |
| } | |
| to { | |
| opacity: 1; | |
| transform: translateY(0); | |
| } | |
| } | |
| """ | |
| ) as demo: | |
| # Title | |
| gr.HTML(""" | |
| <div style="text-align: center; padding: 30px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); | |
| border-radius: 15px; margin-bottom: 20px;"> | |
| <h1 style="color: white; margin: 0; font-size: 2.5em;">🗺️ LifeFlow AI</h1> | |
| <p style="color: white; opacity: 0.9; margin-top: 10px; font-size: 1.2em;"> | |
| Intelligent Life Trip Planning System | |
| </p> | |
| <p style="color: white; opacity: 0.8; margin-top: 5px;"> | |
| Multi-Agent Collaboration | TSPTW Optimization | Real-time Route Planning | |
| </p> | |
| </div> | |
| """) | |
| with gr.Row(): | |
| # ========== Left: Input and Team status ========== | |
| with gr.Column(scale=2, min_width=400): | |
| # Input area (initially visible) | |
| with gr.Group(visible=True) as input_area: | |
| gr.Markdown("### 📝 Enter Your Requirements") | |
| user_input = gr.Textbox( | |
| label="Trip Requirements", | |
| placeholder="E.g.: Tomorrow morning go to NTU Hospital, then go to PX Mart for shopping, need to go to post office to mail package before 3 PM", | |
| lines=4 | |
| ) | |
| with gr.Row(): | |
| start_location = gr.Textbox( | |
| label="Starting Location", | |
| placeholder="E.g.: Daan District, Taipei", | |
| scale=2 | |
| ) | |
| start_time = gr.Textbox( | |
| label="Start Time", | |
| placeholder="08:30", | |
| value="08:30", | |
| scale=1 | |
| ) | |
| deadline = gr.Textbox( | |
| label="Deadline (Optional)", | |
| placeholder="15:00", | |
| value="15:00" | |
| ) | |
| analyze_btn = gr.Button("🚀 Start Analysis", variant="primary", size="lg") | |
| # Task confirmation area (initially hidden) | |
| with gr.Group(visible=False) as task_confirm_area: | |
| gr.Markdown("### ✅ Confirm Tasks") | |
| task_list_display = gr.HTML() # Task list | |
| task_summary_display = gr.HTML() # Task summary | |
| with gr.Row(): | |
| back_btn = gr.Button("← Return to Modify", variant="secondary") | |
| confirm_btn = gr.Button("✅ Confirm Planning", variant="primary") | |
| # Team status area (initially hidden) | |
| with gr.Group(visible=False) as team_area: | |
| gr.Markdown("### 🤝 Team Collaboration Status") | |
| agent_displays = [] | |
| for agent_key in ['planner', 'scout', 'optimizer', 'validator', 'weather', 'traffic']: | |
| agent_card = gr.HTML(value=self.create_agent_card(agent_key, "idle", "Standby")) | |
| agent_displays.append(agent_card) | |
| trip_summary_display = gr.HTML() # Trip summary (user-facing) | |
| # ========== Right: Status/Chat/Tabs ========== | |
| with gr.Column(scale=5, min_width=600): | |
| # Status bar and chat combined area | |
| with gr.Group(): | |
| status_bar = gr.Textbox( | |
| label="📊 Real-time Status", | |
| value="Waiting for input...", | |
| interactive=False, | |
| max_lines=1 | |
| ) | |
| # Chat modification area (only shown during task confirmation) | |
| with gr.Group(visible=False) as chat_modify_area: | |
| gr.Markdown("### 💬 Chat with AI to Modify Tasks") | |
| chatbot = gr.Chatbot( | |
| value=[], | |
| height=150, | |
| show_label=False | |
| ) | |
| with gr.Row(): | |
| chat_input = gr.Textbox( | |
| placeholder="E.g.: Change task 2 to high priority", | |
| show_label=False, | |
| scale=4 | |
| ) | |
| chat_send = gr.Button("Send", variant="primary", scale=1) | |
| # Tabs area | |
| with gr.Tabs() as tabs: | |
| # Tab 1: AI conversation log (default) | |
| with gr.Tab("💬 AI Conversation Log", id="chat_tab"): | |
| gr.Markdown("*Shows reasoning process of all AI Agents*") | |
| reasoning_output = gr.HTML( | |
| value=self.get_reasoning_html() | |
| ) | |
| # Tab 2: Complete report (shown after planning completes) | |
| with gr.Tab("📊 Complete Report", id="summary_tab", visible=False) as summary_tab: | |
| gr.Markdown("*Detailed trip planning report*") | |
| complete_summary_output = gr.HTML() | |
| # Tab 3: Route map (shown after planning completes) | |
| with gr.Tab("🗺️ Route Map", id="map_tab", visible=False) as map_tab: | |
| gr.Markdown("*Optimized route map*") | |
| map_output = gr.Plot(value=self.create_map(), show_label=False) | |
| # Tab 4: Settings | |
| with gr.Tab("⚙️ Settings", id="settings_tab"): | |
| gr.Markdown("### 🔑 API Keys Configuration") | |
| google_maps_key = gr.Textbox( | |
| label="Google Maps API Key", | |
| placeholder="AIzaSy...", | |
| type="password", | |
| value=self.settings['google_maps_api_key'] | |
| ) | |
| openweather_key = gr.Textbox( | |
| label="OpenWeather API Key", | |
| placeholder="...", | |
| type="password", | |
| value=self.settings['openweather_api_key'] | |
| ) | |
| anthropic_key = gr.Textbox( | |
| label="Anthropic API Key", | |
| placeholder="sk-ant-...", | |
| type="password", | |
| value=self.settings['anthropic_api_key'] | |
| ) | |
| gr.Markdown("### 🤖 Model Settings") | |
| model_choice = gr.Dropdown( | |
| label="Select Model", | |
| choices=[ | |
| "claude-sonnet-4-20250514", | |
| "claude-opus-4-20250514", | |
| "gpt-4o", | |
| "gemini-pro" | |
| ], | |
| value=self.settings['model'] | |
| ) | |
| save_settings_btn = gr.Button("💾 Save Settings", variant="primary") | |
| settings_status = gr.Textbox(label="Status", value="", interactive=False, max_lines=1) | |
| # Examples | |
| gr.Examples( | |
| examples=[ | |
| ["Tomorrow morning go to NTU Hospital for consultation, then go to PX Mart for shopping, need to go to post office to mail package before 3 PM", "Daan District, Taipei", "08:30", "15:00"] | |
| ], | |
| inputs=[user_input, start_location, start_time, deadline] | |
| ) | |
| # ========== Event binding ========== | |
| # Start analysis | |
| analyze_btn.click( | |
| fn=self.step1_analyze_tasks, | |
| inputs=[user_input], | |
| outputs=[ | |
| *agent_displays, | |
| reasoning_output, | |
| task_list_display, | |
| task_summary_display, | |
| task_confirm_area, | |
| chat_modify_area, | |
| chatbot, | |
| status_bar | |
| ] | |
| ).then( | |
| fn=lambda: (gr.update(visible=False), gr.update(visible=True)), | |
| outputs=[input_area, task_confirm_area] | |
| ) | |
| # Return to modify | |
| back_btn.click( | |
| fn=lambda: ( | |
| gr.update(visible=True), | |
| gr.update(visible=False), | |
| gr.update(visible=False), | |
| "Waiting for input..." | |
| ), | |
| outputs=[input_area, task_confirm_area, chat_modify_area, status_bar] | |
| ) | |
| # Chat modification | |
| chat_send.click( | |
| fn=self.modify_task_chat, | |
| inputs=[chat_input, chatbot], | |
| outputs=[chatbot, task_list_display] | |
| ).then( | |
| fn=lambda: "", | |
| outputs=[chat_input] | |
| ) | |
| chat_input.submit( | |
| fn=self.modify_task_chat, | |
| inputs=[chat_input, chatbot], | |
| outputs=[chatbot, task_list_display] | |
| ).then( | |
| fn=lambda: "", | |
| outputs=[chat_input] | |
| ) | |
| # Confirm planning | |
| confirm_btn.click( | |
| fn=lambda: ( | |
| gr.update(visible=False), | |
| gr.update(visible=False), | |
| gr.update(visible=True) | |
| ), | |
| outputs=[task_confirm_area, chat_modify_area, team_area] | |
| ).then( | |
| fn=self.step2_search_pois, | |
| outputs=[ | |
| *agent_displays, | |
| reasoning_output, | |
| status_bar | |
| ] | |
| ).then( | |
| fn=self.step3_optimize_route, | |
| outputs=[ | |
| *agent_displays, | |
| reasoning_output, | |
| trip_summary_display, | |
| map_output, | |
| map_tab, | |
| status_bar | |
| ] | |
| ).then( | |
| # After planning completes, automatically switch to Complete Report Tab and display | |
| fn=lambda trip_summary: ( | |
| trip_summary, # Complete report content | |
| gr.update(visible=True), # Show complete report tab | |
| gr.update(selected="summary_tab") # Automatically switch to complete report tab | |
| ), | |
| inputs=[trip_summary_display], | |
| outputs=[complete_summary_output, summary_tab, tabs] | |
| ) | |
| # Save settings | |
| save_settings_btn.click( | |
| fn=self.save_settings, | |
| inputs=[google_maps_key, openweather_key, anthropic_key, model_choice], | |
| outputs=[settings_status] | |
| ) | |
| return demo | |
| def main(): | |
| app = LifeFlowInteractive() | |
| demo = app.build_interface() | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, # 7860 | |
| share=True, | |
| show_error=True | |
| ) | |
| if __name__ == "__main__": | |
| main() | |