Sửa lỗi requests.post chặn event loop khiến chainlit không xử lý được tín hiệu WebSocket gây disconnect

app.py

@@ -2,13 +2,12 @@ import re
 import time
 import chainlit as cl
 import pandas as pd
-import requests
+import httpx
 import asyncio
 from typing import Dict, List, Any, Optional, Callable
 from dataclasses import dataclass, field
 import os
 import uuid
-import threading
 from datetime import datetime, timedelta
 
 
@@ -25,9 +24,9 @@ class ConversationState:
     outputs: Optional[Dict[str, Any]] = None
     selected_model: str = "Gemini 2.0 Flash"
     product_model_search: bool = False
-    # New fields for delayed cleanup
+    # New fields for delayed cleanup - now using asyncio
     pending_cleanup: bool = False
-    cleanup_timer: Optional[threading.Timer] = None
+    cleanup_task: Optional[asyncio.Task] = None
     last_activity: datetime = field(default_factory=datetime.now)
     
     def reset(self):
@@ -39,16 +38,16 @@ class ConversationState:
         self.outputs = None
         self.selected_model = "Gemini 2.0 Flash"
         self.product_model_search = False
-        # Reset cleanup fields but don't touch timers
+        # Reset cleanup fields but don't touch tasks
         self.pending_cleanup = False
         self.last_activity = datetime.now()
 
-    def cancel_cleanup_timer(self):
-        """Cancel pending cleanup timer if exists"""
-        if self.cleanup_timer:
-            self.cleanup_timer.cancel()
-            self.cleanup_timer = None
-            print(f"🚫 Cancelled cleanup timer for session: {self.session_id}")
+    def cancel_cleanup_task(self):
+        """Cancel pending cleanup task if exists"""
+        if self.cleanup_task and not self.cleanup_task.done():
+            self.cleanup_task.cancel()
+            self.cleanup_task = None
+            print(f"🚫 Cancelled cleanup task for session: {self.session_id}")
 
 
 class StateManager:
@@ -56,12 +55,12 @@ class StateManager:
     
     # CLASS-LEVEL session storage for isolation between different browser sessions
     _session_states: Dict[str, ConversationState] = {}
-    _lock = threading.Lock()  # Thread safety for timer operations
+    _lock = asyncio.Lock()  # Async lock for consistency
     
     @staticmethod
-    def get_or_create_session_state(session_id: str) -> ConversationState:
+    async def get_or_create_session_state(session_id: str) -> ConversationState:
         """Get existing session state or create new one"""
-        with StateManager._lock:
+        async with StateManager._lock:
             if session_id not in StateManager._session_states:
                 state = ConversationState()
                 state.session_id = session_id
@@ -73,7 +72,7 @@ class StateManager:
                 
                 # CRITICAL: If session was pending cleanup, cancel it because user is active again
                 if state.pending_cleanup:
-                    state.cancel_cleanup_timer()
+                    state.cancel_cleanup_task()
                     state.pending_cleanup = False
                     print(f"♻️ User activity detected! Cancelled pending cleanup for: {session_id}")
             
@@ -82,35 +81,41 @@ class StateManager:
             return state
     
     @staticmethod
-    def schedule_delayed_cleanup(session_id: str, delay_seconds: int = 3600):
-        """Schedule delayed cleanup for a session (default 1 hours for disconnect tolerance)"""
-        with StateManager._lock:
+    async def schedule_delayed_cleanup(session_id: str, delay_seconds: int = 3600):
+        """Schedule delayed cleanup for a session using asyncio (default 1 hour for disconnect tolerance)"""
+        async with StateManager._lock:
             if session_id not in StateManager._session_states:
                 print(f"⚠️ Cannot schedule cleanup for non-existent session: {session_id}")
                 return
                 
             state = StateManager._session_states[session_id]
             
-            # Cancel existing timer if any
-            state.cancel_cleanup_timer()
+            # Cancel existing task if any
+            state.cancel_cleanup_task()
             
             # Mark as pending cleanup
             state.pending_cleanup = True
             
-            # Schedule new cleanup
-            def delayed_cleanup():
-                print(f"⏰ Executing delayed cleanup for session: {session_id}")
-                StateManager._perform_actual_cleanup(session_id)
+            # Schedule new cleanup using asyncio
+            async def delayed_cleanup():
+                try:
+                    await asyncio.sleep(delay_seconds)
+                    print(f"⏰ Executing delayed cleanup for session: {session_id}")
+                    await StateManager._perform_actual_cleanup(session_id)
+                except asyncio.CancelledError:
+                    print(f"🚫 Cleanup task cancelled for session: {session_id}")
+                    raise
+                except Exception as e:
+                    print(f"❌ Error in delayed cleanup for {session_id}: {e}")
             
-            state.cleanup_timer = threading.Timer(delay_seconds, delayed_cleanup)
-            state.cleanup_timer.start()
+            state.cleanup_task = asyncio.create_task(delayed_cleanup())
             
             print(f"⏱️ Scheduled cleanup in {delay_seconds}s for session: {session_id} (likely disconnect)")
     
     @staticmethod
-    def _perform_actual_cleanup(session_id: str):
+    async def _perform_actual_cleanup(session_id: str):
         """Perform the actual cleanup after delay"""
-        with StateManager._lock:
+        async with StateManager._lock:
             if session_id not in StateManager._session_states:
                 print(f"⚠️ Session already cleaned or doesn't exist: {session_id}")
                 return
@@ -122,7 +127,7 @@ class StateManager:
                 print(f"🚫 Cleanup cancelled - user activity detected for: {session_id}")
                 return
             
-            # Perform API cleanup
+            # Perform API cleanup using httpx
             try:
                 if API_BASE_URL:
                     payload = {
@@ -130,8 +135,9 @@ class StateManager:
                         "reset_model": False, 
                         "session_id": session_id
                     }
-                    response = requests.post(f"{API_BASE_URL}/clear_memory", json=payload, timeout=30)
-                    print(f"Clear memory response for {session_id}: {response.status_code}")
+                    async with httpx.AsyncClient(timeout=30.0) as client:
+                        response = await client.post(f"{API_BASE_URL}/clear_memory", json=payload)
+                        print(f"Clear memory response for {session_id}: {response.status_code}")
             except Exception as e:
                 print(f"Warning: clear_memory failed for {session_id}: {e}")
             
@@ -140,13 +146,13 @@ class StateManager:
             print(f"🗑️ Successfully cleaned up session: {session_id}")
     
     @staticmethod
-    def cleanup_session_immediate(session_id: str):
+    async def cleanup_session_immediate(session_id: str):
         """Immediate cleanup (for testing or forced cleanup)"""
-        with StateManager._lock:
+        async with StateManager._lock:
             if session_id in StateManager._session_states:
                 state = StateManager._session_states[session_id]
-                state.cancel_cleanup_timer()
-                StateManager._perform_actual_cleanup(session_id)
+                state.cancel_cleanup_task()
+                await StateManager._perform_actual_cleanup(session_id)
     
     @staticmethod
     async def clear_chat_state(state: ConversationState):
@@ -158,8 +164,9 @@ class StateManager:
                     "reset_model": False, 
                     "session_id": state.session_id
                 }
-                response = requests.post(f"{API_BASE_URL}/clear_memory", json=payload, timeout=30)
-                print(f"Clear memory response: {response.status_code}")
+                async with httpx.AsyncClient(timeout=30.0) as client:
+                    response = await client.post(f"{API_BASE_URL}/clear_memory", json=payload)
+                    print(f"Clear memory response: {response.status_code}")
             except Exception as e:
                 print(f"Warning: clear_memory failed: {e}")
         
@@ -169,26 +176,26 @@ class StateManager:
         state.session_id = session_id
     
     @staticmethod
-    def change_model(state: ConversationState, model_name: str):
+    async def change_model(state: ConversationState, model_name: str):
         """Change the selected model"""
         state.selected_model = model_name
         state.last_activity = datetime.now()
     
     @staticmethod
-    def toggle_product_model_search(state: ConversationState):
+    async def toggle_product_model_search(state: ConversationState):
         """Toggle product model search mode"""
         state.product_model_search = not state.product_model_search
         state.last_activity = datetime.now()
     
     @staticmethod
-    def get_session_status() -> Dict[str, Dict[str, Any]]:
+    async def get_session_status() -> Dict[str, Dict[str, Any]]:
         """Get status of all sessions (for debugging)"""
-        with StateManager._lock:
+        async with StateManager._lock:
             status = {}
             for session_id, state in StateManager._session_states.items():
                 status[session_id] = {
                     "pending_cleanup": state.pending_cleanup,
-                    "has_timer": state.cleanup_timer is not None,
+                    "has_task": state.cleanup_task is not None and not state.cleanup_task.done(),
                     "last_activity": state.last_activity.isoformat(),
                     "selected_model": state.selected_model,
                     "product_model_search": state.product_model_search
@@ -197,7 +204,7 @@ class StateManager:
 
 
 class ChatService:
-    """Handles chat-related operations"""
+    """Handles chat-related operations with async HTTP calls"""
     
     @staticmethod
     async def respond_to_chat(
@@ -205,7 +212,7 @@ class ChatService:
         message: str, 
         image_path: Optional[str] = None
     ) -> str:
-        """Handle chat responses with image support"""
+        """Handle chat responses with image support using async HTTP"""
         print(f"🔄 === DEBUG STATE ===\n Chat request with model: {state.selected_model}, Product Model Search: {state.product_model_search}, Session ID: {state.session_id}")
         
         # Update activity timestamp - this is KEY to prevent cleanup during active use
@@ -219,12 +226,14 @@ class ChatService:
         if not state.session_id:
             return "Error: Session ID not initialized"
         
-        # Call API
+        # Call API using httpx for async HTTP
         try:
-            if image_path:
-                # For image uploads, use form-data format as expected by API
-                with open(image_path, 'rb') as f:
-                    files = {"image": f}
+            async with httpx.AsyncClient(timeout=600.0) as client:
+                if image_path:
+                    # For image uploads, use form-data format as expected by API
+                    with open(image_path, 'rb') as f:
+                        files = {"image": f.read()}
+                    
                     data = {
                         "message": message,
                         "product_model_search": str(state.product_model_search).lower(),
@@ -232,26 +241,27 @@ class ChatService:
                         "llm_model": state.selected_model,
                         "debug": "Normal"
                     }
-                    resp = requests.post(
+                    
+                    # Use multipart form data for image upload
+                    files_dict = {"image": ("image.jpg", files["image"], "image/jpeg")}
+                    resp = await client.post(
                         f"{API_BASE_URL}/chat_with_image", 
-                        files=files, 
-                        data=data, 
-                        timeout=600
+                        files=files_dict, 
+                        data=data
+                    )
+                else:
+                    # For text messages, use form-data format as expected by API
+                    data = {
+                        "message": message,
+                        "session_id": state.session_id,
+                        "debug": "Normal",
+                        "product_model_search": str(state.product_model_search).lower(),
+                        "llm_model": state.selected_model
+                    }
+                    resp = await client.post(
+                        f"{API_BASE_URL}/chat",
+                        data=data  # Form data format
                     )
-            else:
-                # For text messages, use form-data format as expected by API
-                data = {
-                    "message": message,
-                    "session_id": state.session_id,
-                    "debug": "Normal",
-                    "product_model_search": state.product_model_search,
-                    "llm_model": state.selected_model
-                }
-                resp = requests.post(
-                    f"{API_BASE_URL}/chat",
-                    data=data,  # Changed from json to data for form format
-                    timeout=600
-                )
 
             if resp.status_code == 200:
                 j = resp.json()
@@ -283,12 +293,12 @@ class ChatService:
 
         # Filter products based on query
         if state.specs_advantages is not None:
-            ChatService.get_specific_product_from_query(message, state)
+            await ChatService.get_specific_product_from_query(message, state)
 
         return response + f"\n\n*Thời gian xử lí: {end - start:.6f}s*"
 
     @staticmethod
-    def get_specific_product_from_query(query, state):
+    async def get_specific_product_from_query(query, state):
         """Filter specs_advantages based on models found in query"""
         specs_map = state.specs_advantages or {}
         product_model_list = []
@@ -314,10 +324,10 @@ class ChatService:
 
 
 class DisplayService:
-    """Handles display-related operations"""
+    """Handles display-related operations with async HTTP calls"""
     
     @staticmethod
-    def show_specs(state: ConversationState) -> str:
+    async def show_specs(state: ConversationState) -> str:
         """Generate specifications table"""
         specs_map = state.specs_advantages
         columns = ["Thông số"]
@@ -420,7 +430,7 @@ class DisplayService:
         return f"📄 **Thông số kỹ thuật**\n\n{markdown_table}"
     
     @staticmethod
-    def show_advantages(state: ConversationState) -> str:
+    async def show_advantages(state: ConversationState) -> str:
         """Generate advantages as bullet list instead of table"""
         specs_map = state.specs_advantages
 
@@ -452,7 +462,7 @@ class DisplayService:
         return content
     
     @staticmethod
-    def show_solution_packages(state: ConversationState) -> str:
+    async def show_solution_packages(state: ConversationState) -> str:
         """Show solution packages in a structured format"""
         packages = state.solution_packages
 
@@ -463,7 +473,8 @@ class DisplayService:
         return markdown_table
 
     @staticmethod
-    def show_all_products_table(state: ConversationState):
+    async def show_all_products_table(state: ConversationState):
+        """Show all products table using async HTTP"""
         outputs = state.outputs or {}
 
         if not outputs:
@@ -472,8 +483,8 @@ class DisplayService:
         try:
             # Updated to match API format - send outputs in request body
             payload = {"outputs": outputs}
-            resp = requests.post(f"{API_BASE_URL}/products_by_category",
-                                 json=payload, timeout=60)
+            async with httpx.AsyncClient(timeout=60.0) as client:
+                resp = await client.post(f"{API_BASE_URL}/products_by_category", json=payload)
 
             if resp.status_code == 200:
                 data = resp.json()
@@ -542,8 +553,8 @@ class UIService:
         return msg
 
 
-# HELPER FUNCTIONS: Session management with proper error handling
-def ensure_session_state() -> Optional[ConversationState]:
+# HELPER FUNCTIONS: Session management with proper async error handling
+async def ensure_session_state() -> Optional[ConversationState]:
     """Ensure session state exists, create if not"""
     try:
         session_id = cl.user_session.get("session_id")
@@ -552,21 +563,21 @@ def ensure_session_state() -> Optional[ConversationState]:
             print(f"Lỗi: Không lấy được session id ở ensure_session_state")
             return None
         
-        return StateManager.get_or_create_session_state(session_id)
+        return await StateManager.get_or_create_session_state(session_id)
         
     except Exception as e:
         print(f"⚠️ Error ensuring session state: {e}")
         return None
 
 
-def get_current_session_state() -> Optional[ConversationState]:
+async def get_current_session_state() -> Optional[ConversationState]:
     """Get current session state using Chainlit's session system"""
     try:
         # Use Chainlit's user session to get unique session ID
         chainlit_session_id = cl.user_session.get("session_id")
         
         if chainlit_session_id:
-            return StateManager.get_or_create_session_state(chainlit_session_id)
+            return await StateManager.get_or_create_session_state(chainlit_session_id)
         else:
             print("⚠️ No Chainlit session ID found")
             return None
@@ -586,7 +597,7 @@ async def on_chat_start():
     else:
         print(f"🔄 Reusing existing session_id: {session_id}")
         
-    app_state = StateManager.get_or_create_session_state(session_id)
+    app_state = await StateManager.get_or_create_session_state(session_id)
 
     await cl.Message(
         content=f"🛍️ **RangDong Sales Agent** (Session: {session_id[:8]}...)\n\n"
@@ -607,7 +618,7 @@ async def on_chat_start():
 
 @cl.on_chat_end
 async def on_chat_end():
-    """Handle chat session end with delayed cleanup mechanism"""
+    """Handle chat session end with delayed cleanup mechanism using asyncio"""
     try:
         session_id = cl.user_session.get("session_id")
         print(f"📤 on_chat_end triggered for session {session_id}")
@@ -615,7 +626,7 @@ async def on_chat_end():
         if session_id:
             # Schedule delayed cleanup instead of immediate cleanup
             # Use shorter delay (30s) since this is likely just a temporary disconnect
-            StateManager.schedule_delayed_cleanup(session_id, delay_seconds=3600)
+            await StateManager.schedule_delayed_cleanup(session_id, delay_seconds=3600)
             print(f"⏳ Scheduled delayed cleanup for session {session_id} (1h delay for disconnect tolerance)")
         else:
             print("⚠️ No session_id found in on_chat_end")
@@ -627,58 +638,58 @@ async def on_chat_end():
 @cl.action_callback("show_specs")
 async def on_show_specs(action):
     """Handle show specifications action"""
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    specs_content = DisplayService.show_specs(app_state)
+    specs_content = await DisplayService.show_specs(app_state)
     await UIService.send_message_with_buttons(specs_content, app_state, author="assistant")
 
 
 @cl.action_callback("show_advantages")
 async def on_show_advantages(action):
     """Handle show advantages action"""
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    adv_content = DisplayService.show_advantages(app_state)
+    adv_content = await DisplayService.show_advantages(app_state)
     await UIService.send_message_with_buttons(adv_content, app_state, author="assistant")
 
 
 @cl.action_callback("show_packages")
 async def on_show_packages(action):
     """Handle show packages action"""
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    pkg_content = DisplayService.show_solution_packages(app_state)
+    pkg_content = await DisplayService.show_solution_packages(app_state)
     await UIService.send_message_with_buttons(pkg_content, app_state, author="assistant")
 
 @cl.action_callback("show_all_products")
 async def on_show_all_products(action):
     """Handle show all products action"""
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    all_products_content = DisplayService.show_all_products_table(app_state)
+    all_products_content = await DisplayService.show_all_products_table(app_state)
     await UIService.send_message_with_buttons(all_products_content, app_state, author="assistant")
 
 @cl.action_callback("toggle_product_search")
 async def on_toggle_product_search(action):
     """Handle toggle product model search action"""
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    StateManager.toggle_product_model_search(app_state)
+    await StateManager.toggle_product_model_search(app_state)
     
     status_message = (
         "✅ **Đã bật tìm kiếm theo mã sản phẩm**\n\n"
@@ -694,7 +705,7 @@ async def on_toggle_product_search(action):
 @cl.action_callback("change_model")
 async def on_change_model(action):
     """Handle model change action"""
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
@@ -720,7 +731,7 @@ async def on_change_model(action):
 @cl.action_callback("back_to_main")
 async def on_back_to_main(action):
     """Handle back to main menu action"""
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
@@ -735,34 +746,34 @@ async def on_back_to_main(action):
 
 @cl.action_callback("select_model_0")
 async def on_select_model_0(action):
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    StateManager.change_model(app_state, "Gemini 2.0 Flash")
+    await StateManager.change_model(app_state, "Gemini 2.0 Flash")
     await UIService.send_message_with_buttons("✅ Đã chuyển sang **Gemini 2.0 Flash**", app_state, author="assistant")
 
 
 @cl.action_callback("select_model_1")
 async def on_select_model_1(action):
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    StateManager.change_model(app_state, "Gemini 2.5 Flash Lite")
+    await StateManager.change_model(app_state, "Gemini 2.5 Flash Lite")
     await UIService.send_message_with_buttons("✅ Đã chuyển sang **Gemini 2.5 Flash Lite**", app_state, author="assistant")
 
 
 @cl.action_callback("select_model_2")
 async def on_select_model_2(action):
-    app_state = ensure_session_state()
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
         
-    StateManager.change_model(app_state, "Gemini 2.0 Flash Lite")
+    await StateManager.change_model(app_state, "Gemini 2.0 Flash Lite")
     await UIService.send_message_with_buttons("✅ Đã chuyển sang **Gemini 2.0 Flash Lite**", app_state, author="assistant")
 
 
@@ -771,7 +782,7 @@ async def on_select_model_2(action):
 async def on_debug_sessions(action):
     """Debug action to show session status (can be added to debug builds)"""
     try:
-        status = StateManager.get_session_status()
+        status = await StateManager.get_session_status()
         debug_content = "🔍 **Debug: Session Status**\n\n"
         
         if not status:
@@ -780,7 +791,7 @@ async def on_debug_sessions(action):
             for session_id, info in status.items():
                 debug_content += f"**Session: {session_id[:8]}...**\n"
                 debug_content += f"- Pending cleanup: {info['pending_cleanup']}\n"
-                debug_content += f"- Has timer: {info['has_timer']}\n"
+                debug_content += f"- Has task: {info['has_task']}\n"
                 debug_content += f"- Last activity: {info['last_activity']}\n"
                 debug_content += f"- Model: {info['selected_model']}\n"
                 debug_content += f"- Product search: {info['product_model_search']}\n\n"
@@ -792,8 +803,8 @@ async def on_debug_sessions(action):
 
 @cl.on_message
 async def main(message: cl.Message):
-    """Main message handler"""
-    app_state = ensure_session_state()
+    """Main message handler with proper async flow"""
+    app_state = await ensure_session_state()
     if app_state is None:
         await cl.Message(content="Error: Session state not found", author="assistant").send()
         return
@@ -809,14 +820,11 @@ async def main(message: cl.Message):
     # Show typing animation
     typing_msg = await UIService.create_typing_animation()
     
-    # Get response from API
+    # Get response from API - now fully async, won't block event loop
     response = await ChatService.respond_to_chat(app_state, message.content, image_path)
     
     # Update the typing message with final response and buttons
     typing_msg.content = response
     typing_msg.actions = UIService.create_action_buttons(app_state)
     typing_msg.author = "assistant"
-    await typing_msg.update()
-
-
-
+    await typing_msg.update()