""" 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"""
{agent['avatar']}
{agent['name']}
{icon} {message}
""" 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"""
{avatar} {name} {timestamp} {icon}
{message}
""" self.reasoning_messages.append(msg_html) def get_reasoning_html(self): """Get HTML of all reasoning messages (scrollable)""" if not self.reasoning_messages: return """

AI conversation logs will be displayed here

""" # Combine all messages and add auto-scroll to bottom script messages_html = "".join(self.reasoning_messages) return f"""
{messages_html}
""" def create_task_list_ui(self, tasks: List[Dict]): """Create task list UI (theme adaptive)""" if not tasks: return "

No tasks yet

" priority_colors = { 'HIGH': '#ff4444', 'MEDIUM': '#FFA500', 'LOW': '#4A90E2' } html = f"""

πŸ“‹ Identified Task List

""" for i, task in enumerate(tasks): priority = task.get('priority', 'MEDIUM') color = priority_colors.get(priority, '#4A90E2') html += f"""
{i + 1}. {task['description']}
{priority} πŸ“ {task.get('category', 'Other')} {f" | ⏰ {task.get('time_window', 'Anytime')}" if task.get('time_window') else ""}
""" html += f"""
πŸ“Š Overview
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')}
""" 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
" + "
".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"""

πŸ“Š Task Summary

Total tasks: {len(self.task_list)}

High priority tasks: {sum(1 for t in self.task_list if t.get('priority') == 'HIGH')}

Estimated total time: {sum(t.get('estimated_duration', 0) for t in self.task_list)} minutes

""" # 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']}
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!
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 βœ“
β€’ Time window compliance: βœ“
β€’ Priority task completion: βœ“
β€’ 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"""

πŸŽ‰ Trip Planning Complete

πŸ“Š Overview

Total stops: {len(stops)}

Total time: {sum(s.get('duration', 0) for s in stops)} minutes

Start time: {stops[0]['arrival_time'] if stops else 'N/A'}

Estimated completion: {stops[-1]['departure_time'] if stops else 'N/A'}

πŸ“ Detailed Itinerary

""" priority_colors = { 'HIGH': '#ff4444', 'MEDIUM': '#FFA500', 'LOW': '#4A90E2' } for stop in stops: color = priority_colors.get(stop.get('priority', 'MEDIUM'), '#4A90E2') html += f"""
{stop['order']}. {stop['poi_name']}
{stop['description']}
{stop.get('priority', 'MEDIUM')}
πŸ• Arrival: {stop['arrival_time']} πŸ• Departure: {stop['departure_time']} ⏱️ Duration: {stop['duration']} min
""" html += """

βœ… Validation Results

βœ“ All time windows satisfied

βœ“ Deadline met

βœ“ Priority tasks completed

βœ“ Route feasibility: 100%

""" 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("""

πŸ—ΊοΈ LifeFlow AI

Intelligent Life Trip Planning System

Multi-Agent Collaboration | TSPTW Optimization | Real-time Route Planning

""") 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()