import logging import os import json from typing import List, Dict, Optional import gradio as gr from ai_agent.utils.previews import _build_preview_for_vlm from ai_agent.retriever.software_doc import SoftwareDoc from .handlers import respond from .visualizations import ( create_tool_usage_chart, create_tool_timeline, create_disabled_tools_display, ) from .utils import get_available_models, get_default_model_display_name from .state import format_stats_markdown log = logging.getLogger("chat_components") # Load model configurations from config.yaml MODEL_CONFIGS = get_available_models() def get_model_config(model_display_name: str) -> Dict[str, Optional[str]]: """Get model configuration from display name.""" return MODEL_CONFIGS.get( model_display_name, { "name": model_display_name, "base_url": None, "provider": "Unknown", "api_key_env": "OPENAI_API_KEY", }, ) def create_chat_interface(doc_index: Dict[str, SoftwareDoc]): """ Create the chat-based Gradio interface. Args: doc_index: Mapping of tool name -> SoftwareDoc for formatting Returns: Gradio Blocks interface """ # Custom CSS for Imaging Plaza theme custom_css = """ /* Imaging Plaza EPFL Green Theme */ :root { --imaging-green: #00A991; --imaging-green-dark: #008875; --imaging-green-light: #E6F7F4; } .main-header { background: linear-gradient(135deg, var(--imaging-green) 0%, var(--imaging-green-dark) 100%); padding: 1.5rem 2rem; border-radius: 8px; margin-bottom: 1.5rem; display: flex; align-items: center; gap: 1rem; } .logo-container { display: flex; align-items: center; gap: 1rem; } .logo-image { width: 48px; height: 48px; background: white; border-radius: 8px; padding: 8px; } .header-title { color: white; font-size: 1.8rem; font-weight: 600; margin: 0; } .header-subtitle { color: rgba(255, 255, 255, 0.9); font-size: 0.95rem; margin: 0; } button.primary { background: var(--imaging-green) !important; border-color: var(--imaging-green) !important; } button.primary:hover { background: var(--imaging-green-dark) !important; border-color: var(--imaging-green-dark) !important; } .panel-border { border: 2px solid var(--imaging-green-light); border-radius: 8px; padding: 1rem; } """ with gr.Blocks( title="Imaging Plaza - AI Assistant", theme=gr.themes.Soft( primary_hue="green", secondary_hue="teal", ), css=custom_css, fill_height=True, ) as demo: # Header with logo with gr.Row(elem_classes="main-header"): gr.HTML("""
Imaging Plaza Logo

AI Assistant

Find the right imaging tools for your research

""") # Settings section (collapsed by default) with gr.Accordion("⚙️ Settings", open=False): with gr.Row(): # Use agent_model from config as default default_model = get_default_model_display_name() model_dropdown = gr.Dropdown( choices=list(MODEL_CONFIGS.keys()), value=default_model, label="Model", info="Select AI model and inference server", ) top_k_slider = gr.Slider( minimum=5, maximum=20, value=int(os.getenv("TOP_K", "12")), step=1, label="Top K Candidates", info="Number of tools to retrieve", ) num_choices_slider = gr.Slider( minimum=1, maximum=5, value=int(os.getenv("NUM_CHOICES", "3")), step=1, label="Number of Recommendations", info="Tools to recommend to user", ) with gr.Row(equal_height=True): # ================================================================ # LEFT: Chat section # ================================================================ with gr.Column(scale=7): chatbot = gr.Chatbot( label="💬 Chat", type="messages", height=600, show_copy_button=True, avatar_images=("👤", "🤖"), ) # Tool approval box (appears inline when approval needed) with gr.Group(visible=False) as approval_box: gr.Markdown("### 🤖 Tool Recommendation") approve_tool_btn = gr.Button( "🚀 Run Tool", variant="primary", size="lg", scale=1, ) # File downloads section download_files = gr.File( label="📥 Download Results", file_count="multiple", type="filepath", visible=True, height=100, ) with gr.Row(): with gr.Column(scale=8): msg_input = gr.Textbox( label="Your message", placeholder=( "e.g., 'I need to segment lungs in CT scans' or " "'Find tools for microscopy image denoising'" ), lines=2, ) with gr.Column(scale=2): file_input = gr.File( label="📎 Attach files", file_count="multiple", file_types=[ ".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tif", ".tiff", ".dcm", ".nii", ".nii.gz", ".csv", ".json", ".xml", ".mp3", ".wav", ".mp4", ".avi", ], ) with gr.Row(): submit_btn = gr.Button("Send", variant="primary", scale=2) clear_btn = gr.Button("Clear chat", scale=1) # ================================================================ # RIGHT: Analytics and State section # ================================================================ with gr.Column(scale=3, visible=True): # Tool usage visualizations gr.Markdown("### 📊 Tool Usage Statistics") tool_usage_plot = gr.Plot( label="Tool Call Frequency", show_label=False, ) tool_timeline_plot = gr.Plot( label="Tool Call Timeline", show_label=False, ) disabled_tools_text = gr.Markdown( value="✅ No tools disabled", label="Disabled Tools", ) # Collapsible state display with gr.Accordion("🔧 Raw Conversation State", open=False): state_display = gr.JSON( value={}, show_label=False, ) chat_state = gr.State({}) # ==================================================================== # Event Handlers # ==================================================================== def handle_chat( message: str, history: List[dict], files: List, state_dict: dict, model: str, top_k: int, num_choices: int, ): """ Handle chat message with streaming response. Yields updated history and state after each step. """ # Convert Gradio messages format to internal format if needed if not history: history = [] # Show user message immediately user_msg = {"role": "user", "content": message or ""} # Add file attachments to user message if files: file_list = "\n".join( [ f"📎 {os.path.basename(f.name if hasattr(f, 'name') else str(f))}" for f in files ] ) if message: user_msg["content"] = f"{message}\n\n{file_list}" else: user_msg["content"] = file_list history.append(user_msg) yield history, state_dict, gr.update(), gr.update(), gr.update(), gr.update(), None, gr.update( visible=False ), gr.update() # If files were uploaded, build and show preview immediately if files: file_paths = [] for f in files: if isinstance(f, str): file_paths.append(f) elif hasattr(f, "name"): file_paths.append(f.name) if file_paths: # Build preview try: preview_path, meta_text = _build_preview_for_vlm(file_paths) if preview_path: # Show preview message preview_text = "📋 **Preview for analysis:**" if meta_text: preview_text += f"\n\n_{meta_text}_" history.append( {"role": "assistant", "content": preview_text} ) history.append( { "role": "assistant", "content": {"path": preview_path}, } ) yield history, state_dict, gr.update(), gr.update(), gr.update(), gr.update(), None, gr.update( visible=False ), gr.update() except Exception as e: log.warning("Preview generation failed: %r", e) # Show "thinking" indicator for agent processing thinking_msg = {"role": "assistant", "content": "🤔 Finding tools..."} history.append(thinking_msg) yield history, state_dict, gr.update(), gr.update(), gr.update(), gr.update(), None, gr.update( visible=False ), gr.update() # Call respond function with settings try: reply, new_state = respond( message=message or "", files=files or [], state_dict=state_dict, doc_index=doc_index, model=model, top_k=int(top_k), num_choices=int(num_choices), ) # Remove thinking indicator if history and history[-1] == thinking_msg: history.pop() # Add assistant response with rich media # Build text content first text_content = reply.text # Add stats if available text_content += format_stats_markdown(reply.stats or {}) # Add file links if reply.files: text_content += "\n\n" + "\n".join( [f"📎 [{label}]({path})" for path, label in reply.files] ) # Add JSON if reply.json_data: text_content += ( "\n\n```json\n" + json.dumps(reply.json_data, indent=2) + "\n```" ) # Add code blocks for lang, code in reply.code_blocks: text_content += f"\n\n```{lang}\n{code}\n```" # Add text message first history.append({"role": "assistant", "content": text_content}) # Add each image as a separate message for proper Gradio rendering for img_path in reply.images: if os.path.exists(img_path): history.append( {"role": "assistant", "content": {"path": img_path}} ) # Update state displays state_dict_updated = new_state.to_dict() # Generate visualizations usage_chart = create_tool_usage_chart( state_dict_updated.get("tool_calls", []) ) timeline_chart = create_tool_timeline( state_dict_updated.get("tool_calls", []) ) disabled_text = create_disabled_tools_display( state_dict_updated.get("tool_calls", []) ) # Extract downloadable files downloaded_files = ( [path for path, _label in reply.files] if reply.files else None ) # Determine button visibility and label using registry box_visible = new_state.pending_tool_approval is not None if box_visible and new_state.pending_tool_approval: from ai_agent.agent.tools.mcp import ( get_tool_display_name, get_tool_icon, ) display_name = get_tool_display_name( new_state.pending_tool_approval ) icon = get_tool_icon(new_state.pending_tool_approval) button_label = f"{icon} Run {display_name}" else: button_label = "🚀 Run Tool" yield ( history, state_dict_updated, gr.update(value=usage_chart), gr.update(value=timeline_chart), gr.update(value=disabled_text), gr.update(value=state_dict_updated), downloaded_files, gr.update(visible=box_visible), # approval_box gr.update(value=button_label), # approve_tool_btn ) except Exception as e: log.exception("Error in chat handler") if history: history.pop() # Remove thinking indicator error_msg = { "role": "assistant", "content": ( f"❌ Error: {str(e)}\n\n" "Please try again or rephrase your request." ), } history.append(error_msg) yield history, state_dict, gr.update(), gr.update(), gr.update(), gr.update(), None, gr.update( visible=False ), gr.update() def clear_chat(): """Reset everything.""" empty_chart = create_tool_usage_chart([]) empty_timeline = create_tool_timeline([]) return ( [], {}, empty_chart, empty_timeline, "✅ No tools disabled", gr.update(value={}), None, gr.update(visible=False), gr.update(), ) def handle_tool_approval(history: List[dict], state_dict: dict): """Handle tool approval button click - executes the pending tool.""" from .handlers import execute_tool_with_approval from .state import ChatState state = ChatState.from_dict(state_dict) if not state.pending_tool_approval: return history, state_dict, None, gr.update(visible=False), gr.update() # Execute the tool reply, new_state = execute_tool_with_approval( state.pending_tool_approval, state.pending_tool_params, state ) # Build response text with stats text_content = reply.text text_content += format_stats_markdown(reply.stats or {}) # Add text message history.append({"role": "assistant", "content": text_content}) # Add images for img_path in reply.images: if os.path.exists(img_path): history.append({"role": "assistant", "content": {"path": img_path}}) # Extract downloadable files downloaded_files = ( [path for path, _label in reply.files] if reply.files else None ) # Update state and hide button state_dict_updated = new_state.to_dict() return ( history, state_dict_updated, downloaded_files, gr.update(visible=False), gr.update(), ) # Wire up events submit_btn.click( handle_chat, inputs=[ msg_input, chatbot, file_input, chat_state, model_dropdown, top_k_slider, num_choices_slider, ], outputs=[ chatbot, chat_state, tool_usage_plot, tool_timeline_plot, disabled_tools_text, state_display, download_files, approval_box, approve_tool_btn, ], ).then( lambda: ("", None), # Clear inputs inputs=None, outputs=[msg_input, file_input], ) msg_input.submit( handle_chat, inputs=[ msg_input, chatbot, file_input, chat_state, model_dropdown, top_k_slider, num_choices_slider, ], outputs=[ chatbot, chat_state, tool_usage_plot, tool_timeline_plot, disabled_tools_text, state_display, download_files, approval_box, approve_tool_btn, ], ).then( lambda: ("", None), # Clear inputs inputs=None, outputs=[msg_input, file_input], ) approve_tool_btn.click( handle_tool_approval, inputs=[chatbot, chat_state], outputs=[ chatbot, chat_state, download_files, approval_box, approve_tool_btn, ], ) clear_btn.click( clear_chat, inputs=None, outputs=[ chatbot, chat_state, tool_usage_plot, tool_timeline_plot, disabled_tools_text, state_display, download_files, approval_box, approve_tool_btn, ], ) return demo