migrate from app.py to app/main.py

app/config/__init__.py

@@ -0,0 +1,3 @@
+from .base import get_app_name_and_config, AppName, REQUIRED_ATTRIBUTES, CONFIG_MAPPING
+
+__all__ = ['get_app_name_and_config', 'AppName', 'REQUIRED_ATTRIBUTES', 'CONFIG_MAPPING']

app/config/base.py

@@ -0,0 +1,81 @@
+from typing import Dict, Type, Set
+import importlib
+import os
+
+class AppName:
+    """Enum for application names."""
+    NCSLM_LPD = 'NCSLM_LPD'
+    SEL = 'SEL'
+    SEL_COACH = 'SEL_COACH'
+
+# Define required attributes that must be present in config
+REQUIRED_ATTRIBUTES: Set[str] = {
+    'ASSISTANT_MODEL',
+    'ASSISTANT_NAME',
+    'ASSISTANT_DESCRIPTION',
+    'ASSISTANT_INSTRUCTION',
+    'RESPONSE_FORMAT',
+    'SHOW_CANVAS',
+    'CONVERSATION_STARTER_SAMPLES'
+}
+
+# Mapping of app names to their config module paths
+CONFIG_MAPPING: Dict[str, str] = {
+    AppName.NCSLM_LPD: 'NCSLM_LPD.assistant_config',
+    AppName.SEL: 'SEL.assistant_config',
+    AppName.SEL_COACH: 'SEL_COACH.assistant_config',
+    'default': 'assistant_config'
+}
+
+def validate_config(config: Type) -> tuple[bool, list[str]]:
+    """
+    Validate that the config has all required attributes.
+    
+    Args:
+        config: The configuration module to validate
+        
+    Returns:
+        tuple[bool, list[str]]: (is_valid, list of missing attributes)
+    """
+    missing_attrs = [attr for attr in REQUIRED_ATTRIBUTES if not hasattr(config, attr)]
+    return len(missing_attrs) == 0, missing_attrs
+
+def get_app_name_and_config() -> tuple[str, Type]:
+    """
+    Get the appropriate configuration module based on APP_NAME.
+    Validates that the config has all required attributes.
+    
+    Returns:
+        Type: The configuration module
+        
+    Raises:
+        ImportError: If the config module cannot be imported
+        ValueError: If the config is missing required attributes
+    """
+    app_name = os.getenv('app_name')
+    config_path = CONFIG_MAPPING.get(app_name, CONFIG_MAPPING['default'])
+    
+    try:
+        config = importlib.import_module(config_path)
+    except ImportError as e:
+        print(f"Error importing config {config_path}: {e}")
+        config = importlib.import_module(CONFIG_MAPPING['default'])
+    
+    # Validate the config
+    is_valid, missing_attrs = validate_config(config)
+    if not is_valid:
+        error_msg = f"Config {config_path} is missing required attributes: {missing_attrs}"
+        print(f"Warning: {error_msg}")
+        
+        # Try to load default config if current config is invalid
+        if config_path != CONFIG_MAPPING['default']:
+            print("Attempting to load default config...")
+            default_config = importlib.import_module(CONFIG_MAPPING['default'])
+            is_valid, missing_attrs = validate_config(default_config)
+            if not is_valid:
+                raise ValueError(f"Default config is also invalid. Missing: {missing_attrs}")
+            return default_config
+        else:
+            raise ValueError(error_msg)
+    
+    return app_name, config

app/handlers/event_handler.py

@@ -0,0 +1,22 @@
+from typing_extensions import override
+from openai import AssistantEventHandler
+from openai.types.beta.threads import Text, TextDelta
+from openai.types.beta.threads.runs import ToolCall
+
+class EventHandler(AssistantEventHandler):
+    """Handler for assistant events during streaming."""
+    
+    @override
+    def on_text_created(self, text: Text) -> None:
+        """Called when text is created."""
+        print(f"\nassistant > ", end="", flush=True)
+
+    @override
+    def on_text_delta(self, delta: TextDelta, snapshot: Text) -> None:
+        """Called when text is updated."""
+        print(delta.value, end="", flush=True)
+
+    @override
+    def on_tool_call_created(self, tool_call: ToolCall) -> None:
+        """Called when a tool call is created."""
+        print(f"\nassistant > {tool_call.type}\n", flush=True)

app/handlers/response_handler.py

@@ -0,0 +1,101 @@
+import gradio as gr
+from typing import Generator, List, Tuple, Dict, Any
+
+class ResponseHandler:
+    def __init__(self, chat_manager, app_name, active_config):
+        self.chat_manager = chat_manager
+        self.app_name = app_name
+        self.active_config = active_config
+
+    def handle_response(self, message: str, history: List[Tuple[str, str]], 
+                    textbox_content: str, username: str) -> Generator:
+        """Handle the chat response"""
+        # Get or create chat_id (which is the thread_id)
+        chat_id = self.chat_manager.get_or_create_thread()
+
+        preprocessed_data = self.active_config.handle_thread_before_chat(self.chat_manager.client, chat_id, message, history, textbox_content)
+
+        full_response = ""
+        canvas_result = ""
+        chat_response = ""
+        next_step_prompt = [[]]
+        prompt_tokens = 0
+        total_tokens = 0
+        
+        try:
+            print(f"Processing request with chat_id: {chat_id}")
+            with self.chat_manager.client.beta.threads.runs.stream(
+                thread_id=chat_id,
+                assistant_id=self.chat_manager.assistant_id,
+                timeout=60
+            ) as stream:
+                try:
+                    for stream_delta in stream:
+                        if not stream_delta:
+                            print("Warning: Empty text delta received", flush=True)
+                            continue
+
+                        if stream_delta.event == 'thread.run.failed':
+                            print(f"Stream delta received: {stream_delta}", flush=True)
+                            if hasattr(stream_delta.data, 'last_error') and stream_delta.data.last_error is not None:
+                                chat_response = stream_delta.data.last_error.model_dump_json()
+                            else:
+                                chat_response = "Sorry, there was an error processing the response. Please try again."
+                            continue
+                        elif stream_delta.event != 'thread.message.delta':
+                            print(f"Stream delta received: {stream_delta}", flush=True)
+                            
+                            if stream_delta.event == 'thread.run.step.completed' and hasattr(stream_delta, 'data'):
+                                if hasattr(stream_delta.data, 'usage'):
+                                    prompt_tokens = stream_delta.data.usage.prompt_tokens
+                                    total_tokens = stream_delta.data.usage.total_tokens
+                                    print(f"\nPrompt tokens: {prompt_tokens}, Total tokens: {total_tokens}\n", flush=True)
+                            continue
+                            
+                        if stream_delta.data.delta.content[0].text:
+                            full_response += stream_delta.data.delta.content[0].text.value
+                            
+                        print(f"Accumulated response length: {len(full_response)}", flush=True)
+                        
+                        if len(full_response) < 2:
+                            continue
+                            
+                        chat_response, canvas_result, next_step_prompt = self.active_config.handle_stream_delta(full_response)
+                        yield chat_response, canvas_result, next_step_prompt
+                        
+                except Exception as stream_error:
+                    print(f"Stream processing error: {str(stream_error)}")
+                    print(f"Error type: {type(stream_error).__name__}")
+                    yield "Sorry, there was an error processing the response. Please try again.", "", [[]]
+                    
+        except Exception as connection_error:
+            print(f"Connection error: {str(connection_error)}")
+            print(f"Error type: {type(connection_error).__name__}")
+            yield "Sorry, the connection timed out. Please try again.", "", [[]]
+
+        # print(f"Current Messages: {message}")
+        # print(f"Suggestion: {chat_response}")
+        # print(f"Current Response: {full_response}")
+        print(f"Prompt tokens: {prompt_tokens}, Total tokens: {total_tokens}")
+
+        final_chat_response, final_canvas_result, final_next_step_prompt, msg_records = self.active_config.handle_stream_end(
+            message, history, chat_response, textbox_content, full_response, canvas_result, preprocessed_data
+        )
+        
+        self.chat_manager.save_chat(username, chat_id, msg_records, canvas_result, self.app_name)
+        yield final_chat_response, final_canvas_result, final_next_step_prompt
+
+    def handle_quick_response_click(self, selected: str) -> str:
+        """Handle quick response selection"""
+        return selected[0]
+
+    def handle_quick_response_samples(self, next_step_prompt: List[List[str]]) -> gr.Dataset:
+        """Handle quick response samples"""
+        if len(next_step_prompt) > 0 and len(next_step_prompt[0]) > 0:
+            return gr.Dataset(samples=next_step_prompt, visible=True)
+        return gr.Dataset(samples=[['-']], visible=False)
+
+    def _get_assistant_response(self, message: str) -> str:
+        """Get response from the assistant"""
+        # Implementation of assistant response logic
+        pass

app/main.py

@@ -0,0 +1,161 @@
+import os
+import gradio as gr
+from openai import OpenAI
+from .config.base import get_app_name_and_config
+from .handlers.event_handler import EventHandler
+from .handlers.response_handler import ResponseHandler
+from .utils.chat import ChatManager
+from .utils.utils import print_assistant_info, check_password
+from .ui.components import create_chatbot, create_textbox, create_prompt_input, create_quick_response, create_hidden_list, create_chat_selector, create_new_chat_button
+
+
+# Initialize OpenAI client
+OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
+client = OpenAI(api_key=OPENAI_API_KEY)
+
+# Get the active config
+try:
+    app_name, active_config = get_app_name_and_config()
+except (ImportError, ValueError) as e:
+    print(f"Fatal error loading config: {e}")
+    raise
+
+# View existing assistants
+existed_assistants = client.beta.assistants.list(
+    order="desc",
+    limit="20",
+)
+
+# print(len(existed_assistants.data), existed_assistants.data)
+
+# Assistant settings
+assistant_id = os.getenv('assistant_id')
+ASSISTANT_MODEL = active_config.ASSISTANT_MODEL
+ASSISTANT_NAME = active_config.ASSISTANT_NAME
+ASSISTANT_DESCRIPTION = active_config.ASSISTANT_DESCRIPTION
+ASSISTANT_INSTRUCTION = active_config.ASSISTANT_INSTRUCTION
+RESPONSE_FORMAT = active_config.RESPONSE_FORMAT
+VECTOR_STORE_NAME = "lesson-plan"
+SHOW_CANVAS = active_config.SHOW_CANVAS
+CONVERSATION_STARTER_SAMPLES = active_config.CONVERSATION_STARTER_SAMPLES
+
+# Initialize chat history manager
+chat_manager = ChatManager(client, assistant_id)
+response_handler = ResponseHandler(chat_manager, app_name, active_config)
+
+# Retrieve assistant
+try:
+    assistant = client.beta.assistants.retrieve(assistant_id)
+    print(f"retrieve assistant success.")
+    print_assistant_info(assistant)
+except Exception as e:
+    print(f"retrieve Assistant Fail: {e}")
+
+# Components
+chatbot = create_chatbot()
+textbox = create_textbox()
+prompt_input = create_prompt_input()
+quick_response = create_quick_response(CONVERSATION_STARTER_SAMPLES)
+hidden_list = create_hidden_list()
+chat_selector = create_chat_selector()
+new_chat_btn = create_new_chat_button()
+
+CORRECT_PASSWORD = os.getenv('ui_password')
+
+# Functions
+def clear_input():
+    return ""
+
+def chat_submit(message, history, textbox, username):
+    if message:
+        msg_records = [{'role': msg['role'], 'content': msg['content']} for msg in history]
+        msg_records.append({'role': 'user', 'content': message})
+        response_generator = response_handler.handle_response(message, history, textbox, username)
+        for suggestion, current_lesson_plan, next_step_prompt in response_generator:            
+            yield msg_records + [{'role': 'assistant', 'content': suggestion}], current_lesson_plan, [[]]
+    yield msg_records + [{'role': 'assistant', 'content': suggestion}], current_lesson_plan, next_step_prompt
+
+# Main UI
+with gr.Blocks(css="""
+    .full-height {height: 100%}
+""") as demo:
+    # password UI popup
+    with gr.Group(visible=True) as password_popup:
+        username_input = gr.Textbox(label="請輸入使用者名稱")
+        password_input = gr.Textbox(label="請輸入密碼", type="password")
+        submit_button = gr.Button("提交")
+        error_message = gr.Textbox(label="", visible=False, interactive=False)
+    
+    # Store username in state
+    user_state = gr.State(value="default_user")
+    
+    # Main UI 
+    with gr.Group(visible=False) as main_ui:
+        with gr.Row(equal_height=True, elem_classes="contain"):
+            with gr.Column(scale=1, min_width=200, render=False):
+                chat_selector.render()
+            with gr.Column(scale=4):
+                with gr.Row(equal_height=True, height="90vh"):
+                    with gr.Column():                        
+                        chatbot.render()
+                        prompt_input.render()
+                        quick_response.render()
+                        hidden_list.render()
+                    with gr.Column(elem_classes="full-height", visible=SHOW_CANVAS) as textbox_column:                        
+                        textbox.render()
+
+    # Event handlers
+    submit_button.click(
+        check_password,
+        inputs=[username_input, password_input],
+        outputs=[password_popup, main_ui, error_message, user_state]
+    )
+    
+    password_input.submit(
+        check_password,        
+        inputs=[username_input, password_input],
+        outputs=[password_popup, main_ui, error_message, user_state]
+    )
+
+    prompt_input.submit(
+        chat_submit,        
+        inputs=[prompt_input, chatbot, textbox, user_state],
+        outputs=[chatbot, textbox, hidden_list]
+    ).then(
+        clear_input,
+        outputs=prompt_input
+    )
+
+    quick_response.click(
+        response_handler.handle_quick_response_click,
+        quick_response,
+        prompt_input
+    )
+
+    hidden_list.change(
+        response_handler.handle_quick_response_samples, 
+        hidden_list, 
+        quick_response
+    )
+
+    # Update chat selector when page loads
+    demo.load(
+        chat_manager.reset_chat_on_load,
+        inputs=[user_state],
+        outputs=[chat_selector, chatbot]
+    )
+    
+    # Handle chat switching
+    chat_selector.change(
+        chat_manager.switch_chat,
+        inputs=[chat_selector, user_state],
+        outputs=[chatbot]
+    )        
+    
+    new_chat_btn.click(
+        chat_manager.create_new_chat,
+        outputs=[chat_selector, chatbot]
+    )
+
+if __name__ == "__main__":
+    demo.launch(debug=True)

app/ui/components.py

@@ -0,0 +1,63 @@
+import gradio as gr
+
+def create_chatbot() -> gr.Chatbot:
+    """Create the main chatbot component."""
+    return gr.Chatbot(
+        type="messages",
+        container=True,
+        scale=8,
+        autoscroll=True
+    )
+
+def create_textbox() -> gr.Textbox:
+    """Create the text editing component."""
+    return gr.Textbox(
+        label="教案編輯",
+        lines=30,
+        render=False,
+        interactive=True,
+        scale=1
+    )
+
+def create_prompt_input() -> gr.Textbox:
+    """Create the user prompt input component."""
+    return gr.Textbox(
+        label="用戶需求",
+        submit_btn=True,
+        render=False,
+        scale=1
+    )
+
+def create_quick_response(samples) -> gr.Dataset:
+    """Create the quick response suggestions component."""
+    return gr.Dataset(
+        samples=samples,
+        components=[create_prompt_input()],
+        render=False,
+        scale=1
+    )
+
+def create_chat_selector() -> gr.Dropdown:
+    """Create the chat selection dropdown."""
+    return gr.Dropdown(
+        label="Select Chat",
+        choices=[],
+        interactive=True,
+        container=True,
+        scale=0
+    )
+
+def create_new_chat_button() -> gr.Button:
+    """Create the new chat button."""
+    return gr.Button(
+        "New Chat",
+        size="sm",
+    )
+
+def create_hidden_list() -> gr.JSON:
+    """Create the hidden list component."""
+    return gr.JSON(
+        value=[[]],
+        render=False,
+        visible=False
+    )

app/utils/chat.py

@@ -0,0 +1,96 @@
+import os
+from openai import OpenAI
+import gradio as gr
+from typing import Tuple
+from supabase import create_client
+from datetime import datetime
+class ChatManager:
+    def __init__(self, client: OpenAI, assistant_id: str, supabase_url: str = None, supabase_key: str = None):
+        self.client = client
+        self.current_chat_id = None
+        self.assistant_id = assistant_id
+        """Initialize Supabase client"""
+        self.supabase = create_client(
+            supabase_url or os.environ.get("SUPABASE_URL"),
+            supabase_key or os.environ.get("SUPABASE_KEY")
+        )
+
+    def create_new_chat(self):
+        """Create a new chat using OpenAI thread ID as chat_id"""
+        thread = self.client.beta.threads.create()
+        self.current_chat_id = thread.id
+        return self.current_chat_id
+
+    def get_or_create_thread(self):
+        """Get or create a thread for the chat"""
+        if not self.current_chat_id:
+            self.current_chat_id = self.create_new_chat()
+        return self.current_chat_id
+
+    def reset_chat_on_load(self, username: str) -> Tuple[gr.update, list]:
+        """Reset chat ID on page load and update chat selector"""
+        self.create_new_chat()
+        
+        # Update chat selector with available chats for this user
+        chats = self.list_user_chats(username)
+        return gr.update(choices=[(c["preview"], c["chat_id"]) for c in chats], value=self.current_chat_id), []
+
+    def load_chat_history(self, user_id="default_user"):
+        """Load chat history for a user"""
+        chat_id = self.get_latest_chat_id(user_id)
+        if chat_id:
+            return self.load_chat(user_id, chat_id)
+        return []
+
+    def list_user_chats(self, user_id="default_user"):
+        """List all chats for a user"""
+        response = self.supabase.table("chats") \
+            .select("chat_id,last_updated,messages") \
+            .eq("user_id", user_id) \
+            .execute()
+        
+        chats = [{
+            "chat_id": chat["chat_id"],
+            "last_updated": chat["last_updated"],
+            "preview": chat["messages"][0]["content"] if chat["messages"] else "Empty chat"
+        } for chat in response.data]
+        
+        return sorted(chats, key=lambda x: x["last_updated"], reverse=True)
+
+
+    def switch_chat(self, chat_id, user_id="default_user"):
+        """Switch to a different chat"""
+        self.current_chat_id = chat_id
+        messages = self.load_chat(user_id, chat_id)
+        return messages
+    
+    def save_chat(self, user_id, chat_id, messages, current_lesson_plan, app_name):
+        """Save chat history to Supabase"""
+        chat_data = {
+            "user_id": user_id,
+            "chat_id": chat_id,
+            "last_updated": datetime.now().isoformat(),
+            "messages": messages,
+            "current_lesson_plan": current_lesson_plan,
+            "app_name": app_name
+        }
+        
+        # Check if chat exists
+        existing_chat = self.supabase.table("chats") \
+            .select("*") \
+            .eq("user_id", user_id) \
+            .eq("chat_id", chat_id) \
+            .execute()
+
+        if existing_chat.data:
+            # Update existing chat
+            self.supabase.table("chats") \
+                .update(chat_data) \
+                .eq("user_id", user_id) \
+                .eq("chat_id", chat_id) \
+                .execute()
+        else:
+            # Insert new chat
+            self.supabase.table("chats") \
+                .insert(chat_data) \
+                .execute()

app/utils/utils.py

@@ -0,0 +1,41 @@
+import json
+import gradio as gr
+import os
+
+CORRECT_PASSWORD = os.getenv('ui_password')
+
+def show_json(obj):
+    print(json.loads(obj.model_dump_json()))
+
+def print_assistant_info(assistant):
+    """Print key information about an assistant object."""
+    # Get model
+    model = assistant.model
+    
+    # Get description (limited to 2 lines)
+    desc_lines = assistant.description.split('\n')[:2]
+    desc = '\n'.join(desc_lines)
+    if len(desc) > 150:  # Truncate if too long
+        desc = desc[:147] + "..."
+    
+    # Get instructions (limited to 2 lines)
+    instr_lines = assistant.instructions.split('\n')[:2]
+    instr = '\n'.join(instr_lines)
+    if len(instr) > 150:  # Truncate if too long
+        instr = instr[:147] + "..."
+    
+    # Print information
+    print(f"Assistant ID: {assistant.id}")
+    print(f"Name: {assistant.name}")
+    print(f"Model: {model}")
+    print(f"Description: {desc}")
+    print(f"Instructions: {instr}")
+    print(f"Tools: {[tool.type for tool in assistant.tools]}")
+    print(f"Response Format: {assistant.response_format}")
+    print(f"Created at: {assistant.created_at}")
+    
+def check_password(username, input_password):
+    if input_password == CORRECT_PASSWORD:
+        return gr.update(visible=False), gr.update(visible=True), "", username
+    else:
+        return gr.update(visible=True), gr.update(visible=False), gr.update(value="密码错误，请重试。hint: channel name", visible=True), username