src/internal/chatbot/ui_chatbot.py

from abc import ABC, abstractmethod
from typing import AsyncGenerator, Dict, Any, Optional, List, Tuple
import asyncio
import gradio as gr
from dataclasses import dataclass
from enum import Enum
from src.internal.chatbot.chatbot_handler import RAGChatbot
class ChatbotInterface(ABC):
    """Abstract interface untuk chatbot UI"""
    
    @abstractmethod
    def create_interface(self) -> gr.Blocks:
        """Create the Gradio interface"""
        pass
    
    @abstractmethod
    def launch(self, **kwargs) -> None:
        """Launch the interface"""
        pass

class ChatbotUI(ChatbotInterface):
    """Gradio implementation of RAG chatbot interface"""
    
    def __init__(self, chatbot: RAGChatbot):
        self.chatbot = chatbot
        self.demo = None
    
    def create_interface(self) -> gr.Blocks:
        """Create the Gradio interface"""
        with gr.Blocks(css=self.chatbot.css, title=self.chatbot.title) as demo:
            # Header
            gr.Markdown(f"""
            # 🤖 {self.chatbot.title}
            ### Intelligent Document Assistant
            """)
            
            # Status indicator
            with gr.Row():
                status_text = gr.Textbox(
                    value="✅ RAG System Ready",
                    label="System Status",
                    interactive=False,
                    container=True
                )
            
            # Main chat interface
            with gr.Row():
                with gr.Column(scale=3):
                    chatbot_ui = gr.Chatbot(
                        elem_id="chatbot",
                        show_label=False,
                        container=True,
                        bubble_full_width=False,
                        show_copy_button=True,
                        layout="panel"
                    )
                    
                    # Message input
                    with gr.Row():
                        msg = gr.Textbox(
                            placeholder="Ask anything about your documents...",
                            show_label=False,
                            scale=4,
                            container=False,
                            lines=1,
                            max_lines=3,
                            autofocus=True
                        )
                        send_btn = gr.Button("Send", variant="primary", scale=1)
                    
                    # Control buttons
                    with gr.Row():
                        clear_btn = gr.Button("🗑️ Clear Chat", variant="secondary")
                        stop_btn = gr.Button("⏹️ Stop", variant="stop", visible=False)
                
                # Settings panel
                with gr.Column(scale=1):
                    gr.Markdown("### ⚙️ RAG Settings")
                    with gr.Group():
                        gr.Markdown("""
                        - **K**: 3 (documents retrieved)
                        - **Template**: Friendly
                        - **Reranking**: Disabled
                        - **Vectorstore**: ChromaDB
                        - **Streaming**: Enabled
                        """)
            
            # State management
            is_generating = gr.State(False)
            
            # Event bindings
            self._bind_events(msg, chatbot_ui, send_btn, clear_btn, stop_btn, is_generating)
            
            # Info panel
            with gr.Accordion("ℹ️ Usage Information", open=False):
                gr.Markdown("""
                ### How to Use:
                1. **Chat**: Type questions about your documents
                2. **Stream**: Responses appear in real-time
                3. **Stop**: Use stop button to halt generation
                4. **Clear**: Reset conversation history
                
                ### Technology Stack:
                - **LLM**: Advanced language model with streaming
                - **Embedding**: High-quality text embeddings
                - **Vectorstore**: ChromaDB for efficient retrieval
                - **Search**: Hybrid search (dense + sparse)
                """)
        
        self.demo = demo
        return demo
    
    def _bind_events(self, msg, chatbot_ui, send_btn, clear_btn, stop_btn, is_generating):
        """Bind all event handlers"""
        # Submit message events
        chat_memory = gr.State([]) 
        submit_event = msg.submit(
            self.chatbot._user_message,
            inputs=[msg, chatbot_ui, is_generating],
            outputs=[msg, chatbot_ui, is_generating, stop_btn, send_btn]
        ).then(
            self.chatbot._bot_message_stream,
            inputs=[chatbot_ui, is_generating, chat_memory],
            outputs=[chatbot_ui, is_generating, stop_btn, send_btn]
        )
        
        send_event = send_btn.click(
            self.chatbot._user_message,
            inputs=[msg, chatbot_ui, is_generating],
            outputs=[msg, chatbot_ui, is_generating, stop_btn, send_btn]
        ).then(
            self.chatbot._bot_message_stream,
            inputs=[chatbot_ui, is_generating, chat_memory],
            outputs=[chatbot_ui, is_generating, stop_btn, send_btn]
        )
        
        # Clear chat event
        clear_btn.click(
            self.chatbot._clear_chat,
            outputs=[chatbot_ui, msg]
        ).then(
            lambda: (False, gr.update(visible=False), gr.update(interactive=True)),
            outputs=[is_generating, stop_btn, send_btn]
        )
        
        # Stop generation event
        stop_btn.click(
            self.chatbot._stop_generation,
            outputs=[is_generating, stop_btn, send_btn],
            cancels=[submit_event, send_event]
        )
    
    def launch(self, port = 7861, **kwargs) -> None:
        """Launch the Gradio interface"""
        if self.demo is None:
            self.create_interface()
        
        # Default launch parameters
        default_params = {
            'share': False,
            'server_name': "0.0.0.0",
            'server_port': port,
            'show_error': True,
            'show_api': False,
            'share':True,
        }
        
        # Merge with user parameters
        launch_params = {**default_params, **kwargs}
        
        print("Launching Gradio interface...")
        self.demo.launch(**launch_params)