jira-ai-assistant / modules /ai_analysis /ai_assistant_methods.py
DocUA's picture
Єдиний коміт - очищення історії
4ad5efa
# Методи для додавання до класу 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