Spaces:
Running
Running
# Методи для додавання до класу JiraAIAssistant | |
# Ці методи потрібно додати до існуючого файлу ai_assistant.py | |
def load_indices(self, indices_dir): | |
""" | |
Завантаження індексів з директорії. | |
Args: | |
indices_dir (str): Шлях до директорії з індексами | |
Returns: | |
bool: True, якщо індекси успішно завантажено, False інакше | |
""" | |
try: | |
from llama_index.core import load_index_from_storage | |
from llama_index.core.storage import StorageContext | |
logger.info(f"Завантаження індексів з директорії: {indices_dir}") | |
# Перевірка наявності директорії | |
if not os.path.exists(indices_dir): | |
logger.error(f"Директорія індексів не існує: {indices_dir}") | |
return False | |
# Перевірка наявності необхідних файлів | |
required_files = ["docstore.json"] | |
for file in required_files: | |
if not os.path.exists(os.path.join(indices_dir, file)): | |
logger.error(f"Відсутній необхідний файл: {file}") | |
return False | |
# Завантажуємо контекст зберігання | |
storage_context = StorageContext.from_defaults(persist_dir=indices_dir) | |
# Завантажуємо індекс | |
self.index = load_index_from_storage(storage_context) | |
# Отримуємо docstore з контексту зберігання | |
self.docstore = storage_context.docstore | |
# Отримуємо доступ до документів | |
node_dict = self.docstore.docs | |
self.nodes = list(node_dict.values()) | |
# Створюємо BM25 retriever | |
self.retriever_bm25 = BM25Retriever.from_defaults( | |
docstore=self.docstore, | |
similarity_top_k=self.similarity_top_k | |
) | |
# Створюємо векторний retriever | |
self.retriever_vector = self.index.as_retriever( | |
similarity_top_k=self.similarity_top_k | |
) | |
# Створюємо гібридний retriever | |
self.retriever_fusion = QueryFusionRetriever( | |
[ | |
self.retriever_bm25, # Пошук на основі BM25 (ключові слова) | |
self.retriever_vector, # Векторний пошук (семантичний) | |
], | |
mode="reciprocal_rerank", # Режим переранжування результатів | |
similarity_top_k=self.similarity_top_k, | |
num_queries=1, # Використовуємо тільки оригінальний запит | |
use_async=True, # Асинхронне виконання для швидкості | |
) | |
# Створюємо query engine на основі гібридного ретривера | |
self.query_engine = RetrieverQueryEngine(self.retriever_fusion) | |
# Відновлюємо jira_documents з вузлів | |
try: | |
self.jira_documents = [] | |
for node in self.nodes: | |
# Створюємо документ з текстом та метаданими вузла | |
doc = Document( | |
text=node.text, | |
metadata=node.metadata | |
) | |
self.jira_documents.append(doc) | |
logger.info(f"Відновлено {len(self.jira_documents)} документів") | |
except Exception as e: | |
logger.warning(f"Не вдалося відновити jira_documents: {e}") | |
logger.info(f"Успішно завантажено індекси з {indices_dir}") | |
return True | |
except Exception as e: | |
logger.error(f"Помилка при завантаженні індексів: {e}") | |
return False | |
def save_indices(self, indices_dir): | |
""" | |
Збереження індексів у директорію. | |
Args: | |
indices_dir (str): Шлях до директорії для збереження індексів | |
Returns: | |
bool: True, якщо індекси успішно збережено, False інакше | |
""" | |
try: | |
logger.info(f"Збереження індексів у директорію: {indices_dir}") | |
# Перевірка наявності директорії | |
if not os.path.exists(indices_dir): | |
os.makedirs(indices_dir) | |
# Перевірка наявності індексу | |
if not hasattr(self, 'index') or self.index is None: | |
logger.error("Відсутній індекс для збереження") | |
return False | |
# Збереження індексу | |
self.index.storage_context.persist(persist_dir=indices_dir) | |
# Збереження додаткових метаданих | |
try: | |
metadata = { | |
"created_at": datetime.now().isoformat(), | |
"documents_count": len(self.jira_documents) if hasattr(self, 'jira_documents') else 0, | |
"nodes_count": len(self.nodes) if hasattr(self, 'nodes') else 0, | |
"embedding_model": str(self.embed_model) if hasattr(self, 'embed_model') else "unknown" | |
} | |
with open(os.path.join(indices_dir, "metadata.json"), "w", encoding="utf-8") as f: | |
json.dump(metadata, f, ensure_ascii=False, indent=2) | |
except Exception as meta_err: | |
logger.warning(f"Помилка при збереженні метаданих: {meta_err}") | |
logger.info(f"Індекси успішно збережено у {indices_dir}") | |
return True | |
except Exception as e: | |
logger.error(f"Помилка при збереженні індексів: {e}") | |
return False |