optimize timing

app/constants.py

@@ -192,4 +192,4 @@ FOUND_REGULATIONS_MESSAGES = [
 ]
 
 SHEET_RANGE = 'chat!A2:N'
-VERSION_NUMBER = 123456792
+VERSION_NUMBER = 123456793

app/message_processor.py

@@ -30,6 +30,8 @@ class MessageProcessor:
             if field not in message_data:
                 logger.error(f"[ERROR] Missing field {field} in message_data: {message_data}")
                 return
+        
+        loop = asyncio.get_event_loop()
         sender_id = message_data["sender_id"]
         page_id = message_data["page_id"]
         message_text = message_data["text"]
@@ -41,38 +43,24 @@ class MessageProcessor:
             logger.info(f"[DEBUG] Không có message_text và attachments, không xử lý...")
             return
 
-        loop = asyncio.get_event_loop()
         sheets_client = self.channel.get_sheets_client()
         history = await loop.run_in_executor(
             None, lambda: sheets_client.get_conversation_history(sender_id, page_id)
         )
         logger.info(f"[DEBUG] history: ... {history[-3:]}")
 
-        # --- SỬA LỖI LOGIC CHỐNG TRÙNG LẶP TẠI ĐÂY ---
-        # Kiểm tra xem timestamp của sự kiện webhook này đã tồn tại trong lịch sử chưa
         for row in history:
-            # Chuyển đổi an toàn sang string để so sánh
             sheet_timestamps = [str(ts) for ts in row.get('timestamp', [])]
             if str(timestamp) in sheet_timestamps:
                 logger.warning(f"Webhook lặp lại cho sự kiện đã tồn tại (timestamp: {timestamp}). Bỏ qua.")
-                return # Bỏ qua hoàn toàn để tránh xử lý lại
+                return
 
-        # --- LUỒNG XỬ LÝ GỐC CỦA BẠN ĐƯỢC GIỮ NGUYÊN ---
         log_kwargs = {
-            'conversation_id': None,
-            'recipient_id': sender_id,
-            'page_id': page_id,
-            'originaltext': message_text,
-            'originalcommand': '',
-            'originalcontent': '',
-            'originalattachments': attachments,
-            'originalvehicle': '',
-            'originalaction': '',
-            'originalpurpose': '',
-            'originalquestion': '',
-            'systemresponse': '',
-            'timestamp': [timestamp],
-            'isdone': False
+            'conversation_id': None, 'recipient_id': sender_id, 'page_id': page_id,
+            'originaltext': message_text, 'originalcommand': '', 'originalcontent': '',
+            'originalattachments': attachments, 'originalvehicle': '', 'originalaction': '',
+            'originalpurpose': '', 'originalquestion': '', 'systemresponse': '',
+            'timestamp': [timestamp], 'isdone': False
         }
         
         logger.info(f"[DEBUG] Message cơ bản: {log_kwargs}")
@@ -81,13 +69,11 @@ class MessageProcessor:
             logger.error("Không thể tạo conversation mới!")
             return
         logger.info(f"[DEBUG] Message history sau lần ghi đầu: {conv}")
-
-        # Thêm timestamp mới nếu chưa có (logic này có thể không cần thiết nữa nhưng giữ lại để không thay đổi luồng)
+        
         conv['timestamp'] = self.flatten_timestamp(conv['timestamp'])
         if timestamp not in conv['timestamp']:
             conv['timestamp'].append(timestamp)
         
-        # Lần gọi thứ 2 để cập nhật thêm thông tin ban đầu (nếu có)
         conv_after_update1 = await loop.run_in_executor(None, lambda: sheets_client.log_conversation(**conv))
         if conv_after_update1:
             conv = conv_after_update1
@@ -98,7 +84,8 @@ class MessageProcessor:
             return
         
         try:
-            await self.facebook.send_message(message=get_random_message(PROCESSING_STATUS_MESSAGES))
+            # TỐI ƯU 1: Gửi tin nhắn trạng thái và không cần chờ đợi
+            asyncio.create_task(self.facebook.send_message(message=get_random_message(PROCESSING_STATUS_MESSAGES)))
         except Exception as e:
             if "expired" in str(e).lower():
                 logger.warning("[FACEBOOK] Token expired, invalidate and refresh")
@@ -131,127 +118,111 @@ class MessageProcessor:
         else:
             keywords = extract_keywords(message_text, VEHICLE_KEYWORDS)
             cau_hoi = message_text
-            for kw in keywords:
-                cau_hoi = cau_hoi.replace(kw, "")
+            for kw in keywords: cau_hoi = cau_hoi.replace(kw, "")
             cau_hoi = cau_hoi.strip()
         
         logger.info(f"[DEBUG] Phương tiện: {keywords} - Hành vi: {tu_khoa} - Mục đích: {muc_dich} - Câu hỏi: {cau_hoi}")
 
-        # Hợp nhất dữ liệu đã phân tích vào `conv`
-        conv['originalcommand'] = command
-        conv['originalcontent'] = remaining_text
-        conv['originalvehicle'] = ','.join(keywords)
-        conv['originalaction'] = tu_khoa
-        conv['originalpurpose'] = muc_dich
-        conv['originalquestion'] = cau_hoi or ""
+        conv.update({
+            'originalcommand': command, 'originalcontent': remaining_text, 'originalvehicle': ','.join(keywords),
+            'originalaction': tu_khoa, 'originalpurpose': muc_dich, 'originalquestion': cau_hoi or ""
+        })
 
         muc_dich_to_use = muc_dich or conv.get('originalpurpose')
         logger.info(f"[DEBUG] Định hướng mục đích xử lý: {muc_dich_to_use}")
-
         conversation_context = self.get_llm_history(history)
 
         response = None
+        handlers = {
+            "hỏi về mức phạt": self.handle_muc_phat,
+            "hỏi về quy tắc giao thông": self.handle_quy_tac,
+            "hỏi về báo hiệu đường bộ": self.handle_bao_hieu,
+            "hỏi về quy trình xử lý vi phạm giao thông": self.handle_quy_trinh,
+            "thông tin cá nhân của AI": self.handle_ca_nhan
+        }
+
         if not command:
-            if muc_dich_to_use == "hỏi về mức phạt":
-                response = await self.handle_muc_phat(conv, conversation_context, page_token, sender_id)
-            elif muc_dich_to_use == "hỏi về quy tắc giao thông":
-                response = await self.handle_quy_tac(conv, conversation_context, page_token, sender_id)
-            elif muc_dich_to_use == "hỏi về báo hiệu đường bộ":
-                response = await self.handle_bao_hieu(conv, conversation_context, page_token, sender_id)
-            elif muc_dich_to_use == "hỏi về quy trình xử lý vi phạm giao thông":
-                response = await self.handle_quy_trinh(conv, conversation_context, page_token, sender_id)
-            elif muc_dich_to_use == "thông tin cá nhân của AI":
-                response = await self.handle_ca_nhan(conv, conversation_context, page_token, sender_id)
-            else:
-                response = await self.handle_khac(conv, conversation_context, page_token, sender_id)
+            handler = handlers.get(muc_dich_to_use, self.handle_khac)
+            response = await handler(conv, conversation_context, page_token, sender_id)
         else:
             if command == "xong":
                 post_url = await self.create_facebook_post(page_token, conv['recipient_id'], [conv])
-                if post_url:
-                    response = f"Bài viết đã được tạo thành công! Bạn có thể xem tại: {post_url}"
-                else:
-                    response = "Đã xảy ra lỗi khi tạo bài viết. Vui lòng thử lại sau."
+                response = f"Bài viết đã được tạo thành công! Bạn có thể xem tại: {post_url}" if post_url else "Đã xảy ra lỗi khi tạo bài viết."
                 conv['isdone'] = True
             else:
                 response = "Vui lòng cung cấp thêm thông tin và gõ lệnh \\xong khi hoàn tất."
                 conv['isdone'] = False
 
-        await self.facebook.send_message(message=response)
+        # TỐI ƯU 2: Gửi câu trả lời cuối cùng và không cần chờ
+        asyncio.create_task(self.facebook.send_message(message=response))
         
         conv['systemresponse'] = response
         
         logger.info(f"Chuẩn bị ghi/cập nhật dữ liệu cuối cùng vào sheet: {conv}")
-        await loop.run_in_executor(None, lambda: sheets_client.log_conversation(**conv))
+        # TỐI ƯU 3: Ghi log cuối cùng và không cần chờ, cho phép webhook trả về ngay
+        asyncio.create_task(
+            loop.run_in_executor(None, lambda: sheets_client.log_conversation(**conv))
+        )
         return
     
+    # ... (các hàm get_latest_timestamp, get_llm_history, flatten_timestamp, normalize_vehicle_keyword không đổi) ...
     def get_latest_timestamp(self,ts_value):
-            if isinstance(ts_value, (int, float)): return int(ts_value)
-            if isinstance(ts_value, str):
-                try: return int(json.loads(ts_value))
-                except: 
-                    try: return int(ts_value)
-                    except: return 0
-            if isinstance(ts_value, list):
-                if not ts_value: return 0
-                return max([self.get_latest_timestamp(item) for item in ts_value]) if ts_value else 0
-            return 0
+        if isinstance(ts_value, (int, float)): return int(ts_value)
+        if isinstance(ts_value, str):
+            try: return int(json.loads(ts_value))
+            except: 
+                try: return int(ts_value)
+                except: return 0
+        if isinstance(ts_value, list):
+            if not ts_value: return 0
+            return max([self.get_latest_timestamp(item) for item in ts_value]) if ts_value else 0
+        return 0
     
     def get_llm_history(self, history: List):
         sorted_history = sorted(history, key=lambda row: self.get_latest_timestamp(row.get('timestamp', 0)))
-
         total_chars = 0
         MAX_CONTEXT_CHARS = 20_000
         conversation_context = []
         for row in reversed(sorted_history):
             temp_blocks = []
-            
-            # --- SỬA LỖI THỨ TỰ LỊCH SỬ TẠI ĐÂY ---
-            # Đảm bảo lượt nói của user luôn được thêm vào trước.
             if row.get('originaltext'):
                 temp_blocks.append({"role": "user", "content": row['originaltext']})
-            # Lượt nói của trợ lý ảo được thêm vào sau.
             if row.get('systemresponse'):
                 temp_blocks.append({"role": "assistant", "content": row['systemresponse']})
-            
             temp_total = sum(len(block['content']) for block in temp_blocks)
             if total_chars + temp_total > MAX_CONTEXT_CHARS: continue
-            
-            # Thêm cặp hỏi-đáp vào đầu danh sách context, duy trì thứ tự thời gian
             conversation_context = temp_blocks + conversation_context
             total_chars += temp_total
         return conversation_context
 
     def flatten_timestamp(self, ts):
         flat = []
-        if not isinstance(ts, list):
-             ts = [ts]
+        if not isinstance(ts, list): ts = [ts]
         for t in ts:
-            if isinstance(t, list):
-                flat.extend(self.flatten_timestamp(t))
-            else:
-                flat.append(t)
+            if isinstance(t, list): flat.extend(self.flatten_timestamp(t))
+            else: flat.append(t)
         return flat
 
     def normalize_vehicle_keyword(self, keyword: str) -> str:
         from app.constants import VEHICLE_KEYWORDS
         import difflib
-        if not keyword:
-            return ""
+        if not keyword: return ""
         matches = difflib.get_close_matches(keyword.lower(), [k.lower() for k in VEHICLE_KEYWORDS], n=1, cutoff=0.6)
         if matches:
             for k in VEHICLE_KEYWORDS:
-                if k.lower() == matches[0]:
-                    return k
+                if k.lower() == matches[0]: return k
         return keyword
-
+        
     async def format_search_results(self, conversation_context: str, question: str, matches: List[Dict[str, Any]], page_token: str, sender_id: str) -> str:
         if not matches:
             return "Không tìm thấy kết quả phù hợp."
-        await self.facebook.send_message(message=get_random_message(FOUND_REGULATIONS_MESSAGES))
+
+        # TỐI ƯU: Gửi tin nhắn trạng thái và không chờ
+        asyncio.create_task(self.facebook.send_message(message=get_random_message(FOUND_REGULATIONS_MESSAGES)))
+
         try:
             reranked = await self.channel.reranker.rerank(question, matches, top_k=10)
-            if reranked:
-                matches = reranked
+            if reranked: matches = reranked
         except Exception as e:
             logger.error(f"[RERANK] Lỗi khi rerank: {e}")
         
@@ -265,16 +236,11 @@ class MessageProcessor:
             fullContent = (match.get('fullcontent') or '').strip()
             full_result_text += f"{fullContent}"
             hpbsnoidung = arr_to_str(match.get('hpbsnoidung'), sep="; ")
-            if hpbsnoidung:
-                full_result_text += f"\n- Hình phạt bổ sung: {hpbsnoidung}"
+            if hpbsnoidung: full_result_text += f"\n- Hình phạt bổ sung: {hpbsnoidung}"
             bpkpnoidung = arr_to_str(match.get('bpkpnoidung'), sep="; ")
-            if bpkpnoidung:
-                full_result_text += f"\n- Biện pháp khắc phục: {bpkpnoidung}"
-            impounding = match.get('impounding')
-            if impounding:
-                full_result_text += f"\n- Tạm giữ phương tiện: 07 ngày"
+            if bpkpnoidung: full_result_text += f"\n- Biện pháp khắc phục: {bpkpnoidung}"
+            if match.get('impounding'): full_result_text += f"\n- Tạm giữ phương tiện: 07 ngày"
         
-        # --- PROMPT MỚI ĐƯỢC CẬP NHẬT TẠI ĐÂY ---
         prompt = (
             "Bạn là một trợ lý pháp lý AI chuyên nghiệp. Nhiệm vụ của bạn là tổng hợp thông tin từ hai nguồn: **Lịch sử trò chuyện** và **Các đoạn luật liên quan** để đưa ra một câu trả lời duy nhất, liền mạch và tự nhiên cho người dùng.\n\n"
             "**QUY TẮC BẮT BUỘC:**\n"
@@ -288,7 +254,9 @@ class MessageProcessor:
             "### Trả lời:"
         )
         
-        await self.facebook.send_message(message=f"{get_random_message(SUMMARY_STATUS_MESSAGES)}")
+        # TỐI ƯU: Gửi tin nhắn trạng thái và không chờ
+        asyncio.create_task(self.facebook.send_message(message=f"{get_random_message(SUMMARY_STATUS_MESSAGES)}"))
+        
         try:
             answer = await self.channel.llm.generate_text(prompt)
             if answer and answer.strip():
@@ -299,13 +267,10 @@ class MessageProcessor:
         except Exception as e:
             logger.error(f"LLM không sẵn sàng: {e}\n{traceback.format_exc()}")
         
-        # Fallback response
-        return "Dựa trên thông tin bạn cung cấp, tôi đã tìm thấy một số quy định liên quan. Tuy nhiên, tôi đang gặp chút khó khăn trong việc tóm tắt. Bạn vui lòng tham khảo nội dung chi tiết trong các văn bản luật nhé."
+        return "Dựa trên thông tin bạn cung cấp..." # Fallback
 
     async def create_facebook_post(self, page_token: str, sender_id: str, history: List[Dict[str, Any]]) -> str:
         logger.info(f"[MOCK] Creating Facebook post for sender_id={sender_id} with history={history}")
-        # In a real scenario, you would use the Facebook Graph API to create a post.
-        # This is a mock implementation.
         return "https://facebook.com/mock_post_url" 
 
     async def handle_muc_phat(self, conv, conversation_context, page_token, sender_id):
@@ -322,12 +287,19 @@ class MessageProcessor:
             embedding = await self.channel.embedder.create_embedding(search_query)
             logger.info(f"[DEBUG] embedding: {embedding[:5]} ... (total {len(embedding)})")
             
+            loop = asyncio.get_event_loop()
             match_count = get_settings().match_count
-            matches = self.channel.supabase.match_documents(
-                embedding,
-                match_count=match_count,
-                user_question=search_query
+            
+            # TỐI ƯU 4: Chạy tác vụ I/O đồng bộ trong executor để không chặn event loop
+            matches = await loop.run_in_executor(
+                None,
+                lambda: self.channel.supabase.match_documents(
+                    embedding,
+                    match_count=match_count,
+                    user_question=search_query
+                )
             )
+
             logger.info(f"[DEBUG] matches: {matches[:2]}...{matches[-2:]}")
             if matches:
                 response = await self.format_search_results(conversation_context, question or action, matches, page_token, sender_id)
@@ -341,7 +313,6 @@ class MessageProcessor:
         return response
 
     async def _handle_general_question(self, conversation_context: str, message_text: str, topic: str) -> str:
-        """Hàm chung để xử lý các câu hỏi kiến thức chung."""
         prompt = (
             "Bạn là một trợ lý AI am hiểu về luật giao thông Việt Nam. "
             "Dựa vào lịch sử trò chuyện và kiến thức của bạn, hãy trả lời câu hỏi của người dùng một cách rõ ràng, ngắn gọn và chính xác.\n"
@@ -365,17 +336,14 @@ class MessageProcessor:
 
     async def handle_quy_tac(self, conv, conversation_context, page_token, sender_id):
         conv['isdone'] = True
-        # return await self._handle_general_question(conversation_context, conv['originaltext'], "quy tắc giao thông")
         return await self.handle_muc_phat(conv, conversation_context, page_token, sender_id)
 
     async def handle_bao_hieu(self, conv, conversation_context, page_token, sender_id):
         conv['isdone'] = True
-        # return await self._handle_general_question(conversation_context, conv['originaltext'], "báo hiệu đường bộ")
         return await self.handle_muc_phat(conv, conversation_context, page_token, sender_id)
 
     async def handle_quy_trinh(self, conv, conversation_context, page_token, sender_id):
         conv['isdone'] = True
-        # return await self._handle_general_question(conversation_context, conv['originaltext'], "quy trình xử lý vi phạm giao thông")
         return await self.handle_muc_phat(conv, conversation_context, page_token, sender_id)
 
     async def handle_ca_nhan(self, conv, conversation_context, page_token, sender_id):
@@ -394,4 +362,4 @@ class MessageProcessor:
             return answer.strip() if answer and answer.strip() else "Chào bạn, mình là WeThoong AI đây!"
         except Exception as e:
             logger.error(f"Lỗi khi xử lý câu hỏi cá nhân: {e}")
-            return "Chào bạn, mình là WeThoong AI, trợ lý giao thông thông minh của bạn!"
+            return "Chào bạn, mình là WeThoong AI, trợ lý giao thông thông minh của bạn!"

app/reranker.py

@@ -1,4 +1,6 @@
 from typing import List, Dict
+
+from app.utils import timing_decorator_async
 from .config import get_settings
 from .gemini_client import GeminiClient
 from loguru import logger
@@ -203,6 +205,7 @@ class Reranker:
             doc['rerank_score'] = 0
             return doc
 
+    @timing_decorator_async
     async def rerank(self, query: str, docs: List[Dict], top_k: int = 5) -> List[Dict]:
         """
         Rerank docs theo độ liên quan với query, trả về top_k docs.