Create conversational_agent.py

core/conversational_agent.py

@@ -0,0 +1,195 @@
+import time
+import traceback
+from typing import List, Dict, Any, Tuple
+from sentence_transformers import SentenceTransformer
+from models.schemas import RAGSearchResult
+from core.multilingual_manager import MultilingualManager
+from core.enhanced_rag import EnhancedRAGSystem
+
+class ConversationalAgent:
+    """Lớp xử lý hội thoại với grounding từ RAG"""
+    
+    def __init__(self, rag_system: EnhancedRAGSystem, multilingual_manager: MultilingualManager):
+        self.rag_system = rag_system
+        self.multilingual_manager = multilingual_manager
+        self.conversation_history = []
+        self.max_history_length = 10
+        
+    def process_query(self, query: str, chat_history: List[Tuple[str, str]] = None) -> Tuple[str, List[Dict]]:
+        """Xử lý truy vấn hội thoại với grounding từ RAG"""
+        try:
+            print(f"🤖 Xử lý truy vấn: {query}")
+            
+            # Phát hiện ngôn ngữ
+            language = self.multilingual_manager.detect_language(query)
+            print(f"🌐 Ngôn ngữ phát hiện: {language}")
+            
+            # Tìm kiếm ngữ nghĩa trong RAG
+            search_results = self.rag_system.semantic_search(query, top_k=3)
+            
+            # Tạo context từ kết quả tìm kiếm
+            context = self._build_context_from_results(search_results, query, language)
+            
+            # Cập nhật lịch sử hội thoại
+            self._update_conversation_history("user", query)
+            
+            # Tạo prompt với context
+            prompt = self._create_grounded_prompt(query, context, language, chat_history)
+            
+            # Tạo phản hồi (ở đây có thể tích hợp với LLM thực tế)
+            response = self._generate_response(prompt, language)
+            
+            # Cập nhật lịch sử
+            self._update_conversation_history("assistant", response)
+            
+            # Trả về response và search results để hiển thị
+            return response, search_results
+            
+        except Exception as e:
+            error_msg = f"Xin lỗi, đã có lỗi xảy ra: {str(e)}"
+            print(f"❌ Lỗi trong ConversationalAgent: {traceback.format_exc()}")
+            return error_msg, []
+    
+    def _build_context_from_results(self, results: List[RAGSearchResult], query: str, language: str) -> str:
+        """Xây dựng context từ kết quả tìm kiếm"""
+        if not results:
+            if language == 'vi':
+                return "Không có thông tin liên quan trong cơ sở tri thức."
+            else:
+                return "No relevant information found in the knowledge base."
+        
+        if language == 'vi':
+            context_parts = [f"Dựa trên câu hỏi '{query}', đây là thông tin liên quan từ cơ sở tri thức:"]
+        else:
+            context_parts = [f"Based on the question '{query}', here is relevant information from the knowledge base:"]
+        
+        for i, result in enumerate(results[:3]):  # Lấy 3 kết quả đầu
+            context_parts.append(f"\n{i+1}. {result.text}")
+            if result.metadata.get('source'):
+                if language == 'vi':
+                    context_parts.append(f"   (Nguồn: {result.metadata.get('source')})")
+                else:
+                    context_parts.append(f"   (Source: {result.metadata.get('source')})")
+        
+        return "\n".join(context_parts)
+    
+    def _create_grounded_prompt(self, query: str, context: str, language: str, 
+                               chat_history: List[Tuple[str, str]] = None) -> str:
+        """Tạo prompt với grounding từ context"""
+        
+        # Xây dựng phần lịch sử hội thoại
+        history_text = ""
+        if chat_history and len(chat_history) > 0:
+            if language == 'vi':
+                history_parts = ["Lịch sử hội thoại gần đây:"]
+                for user_msg, assistant_msg in chat_history[-3:]:  # Lấy 3 cặp gần nhất
+                    history_parts.append(f"Người dùng: {user_msg}")
+                    history_parts.append(f"Trợ lý: {assistant_msg}")
+            else:
+                history_parts = ["Recent conversation history:"]
+                for user_msg, assistant_msg in chat_history[-3:]:
+                    history_parts.append(f"User: {user_msg}")
+                    history_parts.append(f"Assistant: {assistant_msg}")
+            history_text = "\n".join(history_parts) + "\n\n"
+        
+        # Tạo prompt dựa trên ngôn ngữ
+        if language == 'vi':
+            prompt = f"""{history_text}CONTEXT TỪ CƠ SỞ TRI THỨC:
+{context}
+
+CÂU HỎI HIỆN TẠI: {query}
+
+HƯỚNG DẪN:
+1. Trả lời dựa trên context được cung cấp
+2. Nếu context không đủ thông tin, hãy nói rõ
+3. Trả lời bằng tiếng Việt
+4. Giữ câu trả lời ngắn gọn, chính xác
+5. Có thể thêm thông tin bổ sung nếu cần
+
+TRẢ LỜI:"""
+        else:
+            prompt = f"""{history_text}CONTEXT FROM KNOWLEDGE BASE:
+{context}
+
+CURRENT QUESTION: {query}
+
+INSTRUCTIONS:
+1. Answer based on the provided context
+2. If context doesn't have enough information, clearly state that
+3. Respond in {language}
+4. Keep answers concise and accurate
+5. You may add supplementary information if needed
+
+ANSWER:"""
+        
+        return prompt
+    
+    def _generate_response(self, prompt: str, language: str) -> str:
+        """Tạo phản hồi từ prompt (có thể tích hợp với LLM thực)"""
+        # Đây là mô phỏng - trong thực tế sẽ gọi API LLM
+        time.sleep(0.5)  # Giả lập thời gian xử lý
+        
+        # Phân tích prompt để tạo phản hồi phù hợp
+        no_info_vi = "Không có thông tin liên quan trong cơ sở tri thức"
+        no_info_en = "No relevant information found in the knowledge base"
+        
+        if no_info_vi in prompt or no_info_en in prompt:
+            if language == 'vi':
+                return "Xin lỗi, tôi không tìm thấy thông tin cụ thể về câu hỏi này trong cơ sở tri thức hiện tại. Bạn có thể thử diễn đạt câu hỏi khác hoặc thêm thông tin vào cơ sở tri thức."
+            else:
+                return "Sorry, I couldn't find specific information about this question in the current knowledge base. You might try rephrasing your question or adding more information to the knowledge base."
+        
+        # Tạo phản hồi dựa trên ngôn ngữ
+        if language == 'vi':
+            responses = [
+                "Dựa trên thông tin trong cơ sở tri thức, tôi có thể cung cấp cho bạn thông tin sau:",
+                "Theo dữ liệu có sẵn, đây là thông tin liên quan:",
+                "Dựa vào nguồn thông tin được cung cấp, tôi có thể trả lời như sau:"
+            ]
+            follow_up = "\n\nTôi đã tìm thấy một số thông tin hữu ích liên quan đến câu hỏi của bạn. Bạn có thể xem chi tiết các nguồn thông tin bên dưới."
+        else:
+            responses = [
+                "Based on the information in the knowledge base, I can provide you with the following:",
+                "According to the available data, here is the relevant information:",
+                "Based on the provided sources, I can answer as follows:"
+            ]
+            follow_up = "\n\nI've found some useful information related to your question. You can view the detailed sources below."
+        
+        import random
+        base_response = random.choice(responses)
+        
+        return f"{base_response}{follow_up}"
+    
+    def _update_conversation_history(self, role: str, message: str):
+        """Cập nhật lịch sử hội thoại"""
+        self.conversation_history.append({
+            "role": role,
+            "message": message,
+            "timestamp": time.time()
+        })
+        
+        # Giới hạn độ dài lịch sử
+        if len(self.conversation_history) > self.max_history_length:
+            self.conversation_history = self.conversation_history[-self.max_history_length:]
+    
+    def clear_conversation_history(self):
+        """Xóa lịch sử hội thoại"""
+        self.conversation_history = []
+        print("🧹 Đã xóa lịch sử hội thoại")
+    
+    def get_conversation_stats(self) -> Dict:
+        """Lấy thống kê hội thoại"""
+        user_msgs = [m for m in self.conversation_history if m["role"] == "user"]
+        assistant_msgs = [m for m in self.conversation_history if m["role"] == "assistant"]
+        
+        return {
+            "total_messages": len(self.conversation_history),
+            "user_messages": len(user_msgs),
+            "assistant_messages": len(assistant_msgs),
+            "last_user_message": user_msgs[-1]["message"] if user_msgs else None,
+            "last_activity": self.conversation_history[-1]["timestamp"] if self.conversation_history else None
+        }
+    
+    def get_recent_conversation(self, limit: int = 5) -> List[Dict]:
+        """Lấy lịch sử hội thoại gần đây"""
+        return self.conversation_history[-limit:] if self.conversation_history else []