import requests import json import faiss import numpy as np import os from typing import List, Dict import hashlib # Paths to Knowledgebases DB_PATH = './.faiss_db' KB_FnB_FILE_PATH = './toolfnB.json' KB_RECEPTION_FILE_PATH = './ToolReception.json' KB_ROOMSERVICE_FILE_PATH = './ToolRoomService.json' KB_TECHHELP_FILE_PATH = './ToolTechHelp.json' KB_CONCIERGE_FILE_PATH = './ToolConcierge.json' KB_MAINTAINENCE_FILE_PATH = './ToolMaintenance.json' KB_SECURITY_FILE_PATH = './ToolSecurity.json' class JinaEmbedding: def __init__(self): self.api_key = "jina_e6e06cc026404b7da095afeff164160daM5ZSelCfisibWfUt-5GVp173yUn" self.dimension = 768 def embed_document(self, content: str) -> np.ndarray: print(f"Generating new embedding for: {content[:50]}...") jina_api_url = "https://api.jina.ai/v1/embeddings" payload = { "model": "jina-embeddings-v3", "dimensions": self.dimension, "normalized": True, "embedding_type": "float", "input": [{"text": content}] } headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } try: response = requests.post(jina_api_url, headers=headers, json=payload) response.raise_for_status() result = response.json() return np.array(result['data'][0]['embedding'], dtype=np.float32) except Exception as e: print(f"Error in embedding: {e}") return None class VectorStore: def __init__(self): self.embedding_model = JinaEmbedding() self.dimension = self.embedding_model.dimension os.makedirs(DB_PATH, exist_ok=True) self.collections = self.initialize_collections() def initialize_collections(self): collections = { 'FnB': {'index': None, 'metadata': [], 'embeddings': None}, 'Reception': {'index': None, 'metadata': [], 'embeddings': None}, 'RoomService': {'index': None, 'metadata': [], 'embeddings': None}, 'TechHelp': {'index': None, 'metadata': [], 'embeddings': None}, 'Concierge': {'index': None, 'metadata': [], 'embeddings': None}, 'Maintenance': {'index': None, 'metadata': [], 'embeddings': None}, 'Security': {'index': None, 'metadata': [], 'embeddings': None} } # First try to load existing collections for name in collections: if self.load_saved_collection(name, collections[name]): print(f"Successfully loaded saved collection: {name}") else: print(f"No saved collection found for: {name}") # Then load from JSON files only if needed self.load_from_json_files(collections) return collections def load_saved_collection(self, name: str, collection: Dict) -> bool: index_path = os.path.join(DB_PATH, f"{name}.index") metadata_path = os.path.join(DB_PATH, f"{name}_metadata.json") embeddings_path = os.path.join(DB_PATH, f"{name}_embeddings.npy") # Check if all necessary files exist if not all(os.path.exists(p) for p in [index_path, metadata_path, embeddings_path]): return False try: # Load index collection['index'] = faiss.read_index(index_path) # Load metadata with open(metadata_path, 'r') as f: collection['metadata'] = json.load(f) # Load embeddings collection['embeddings'] = np.load(embeddings_path) return True except Exception as e: print(f"Error loading saved collection {name}: {e}") return False def load_from_json_files(self, collections): kb_files = { 'RoomService': KB_ROOMSERVICE_FILE_PATH, 'FnB': KB_FnB_FILE_PATH, 'Reception': KB_RECEPTION_FILE_PATH, 'TechHelp': KB_TECHHELP_FILE_PATH, 'Concierge': KB_CONCIERGE_FILE_PATH, 'Maintenance': KB_MAINTAINENCE_FILE_PATH, 'Security': KB_SECURITY_FILE_PATH } for name, file_path in kb_files.items(): # Only load and generate embeddings if collection is not already loaded if collections[name]['index'] is None and os.path.exists(file_path): try: with open(file_path, 'r', encoding='utf-8') as f: kb_data = json.load(f) self.initialize_new_collection(name, kb_data, collections[name]) except Exception as e: print(f"Error loading {name} from JSON: {e}") def initialize_new_collection(self, name: str, kb_data: List[Dict], collection: Dict): print(f"Initializing new collection for {name}...") # Prepare texts and metadata texts = [entry['user_query'] for entry in kb_data] + [entry['bot_response'] for entry in kb_data] metadata = kb_data + kb_data # Generate embeddings embeddings = [] for text in texts: embedding = self.embedding_model.embed_document(text) if embedding is not None: embeddings.append(embedding) if embeddings: embeddings_array = np.array(embeddings) # Initialize index index = faiss.IndexFlatL2(self.dimension) index.add(embeddings_array) # Update collection collection['index'] = index collection['metadata'] = metadata collection['embeddings'] = embeddings_array # Save to disk self.save_collection(name, collection) print(f"Saved new collection: {name}") def save_collection(self, name: str, collection: Dict): try: # Save index index_path = os.path.join(DB_PATH, f"{name}.index") faiss.write_index(collection['index'], index_path) # Save metadata metadata_path = os.path.join(DB_PATH, f"{name}_metadata.json") with open(metadata_path, 'w') as f: json.dump(collection['metadata'], f) # Save embeddings embeddings_path = os.path.join(DB_PATH, f"{name}_embeddings.npy") np.save(embeddings_path, collection['embeddings']) except Exception as e: print(f"Error saving collection {name}: {e}") def query_collection(self, collection_name: str, query: str, n_results: int) -> Dict: if not query: return {"distances": [], "metadata": [], "best_response": ""} query_embedding = self.embedding_model.embed_document(query) if query_embedding is None: return {"distances": [], "metadata": [], "best_response": ""} try: query_embedding = np.array([query_embedding]) collection = self.collections[collection_name] if collection['index'] is None or collection['index'].ntotal == 0: print(f"Warning: {collection_name} index is empty") return {"distances": [], "metadata": [], "best_response": ""} n_results = min(n_results, collection['index'].ntotal) distances, indices = collection['index'].search(query_embedding, n_results) results_metadata = [collection['metadata'][idx] for idx in indices[0] if idx < len(collection['metadata'])] best_response = "" if results_metadata and distances[0][0] < 0.9: best_response = results_metadata[0].get('bot_response', '') return { "distances": distances[0].tolist(), "metadata": results_metadata, "best_response": best_response } except Exception as e: print(f"Error in query_collection for {collection_name}: {e}") return {"distances": [], "metadata": [], "best_response": ""} def query_roomservice(self, query: str): print(f"Querying room service with: {query}") result = self.query_collection('RoomService', query, 3) if result["best_response"]: return result return { "distances": [], "metadata": [], "best_response": "I'll help you with your room service request. Could you please provide more details about what you need?" } def query_fnb(self, query: str): result = self.query_collection('FnB', query, 2) if not result["best_response"]: result["best_response"] = "I can help you with food and beverage requests. What would you like to order?" return result def query_reception(self, query: str): result = self.query_collection('Reception', query, 3) if not result["best_response"]: result["best_response"] = "I can assist you with reception-related queries. How may I help you?" return result def query_techhelp(self, query: str): result = self.query_collection('TechHelp', query, 1) if not result["best_response"]: result["best_response"] = "I can help you with technical issues. What problem are you experiencing?" return result def query_concierge(self, query: str): result = self.query_collection('Concierge', query, 5) if not result["best_response"]: result["best_response"] = "I can help you with information related queries. How may I help you?" return result def query_maintenance(self, query: str): result = self.query_collection('Maintenance', query, 7) if not result["best_response"]: result["best_response"] = "I can help you with maintenance issues. What problem are you experiencing?" return result def query_security(self, query: str): result = self.query_collection('Security', query, 8) if not result["best_response"]: result["best_response"] = "I can help you with security issues. What problem are you experiencing?" return result