jira-ai-assistant / modules /ai_analysis /ai_assistant_integration_mod.py
DocUA's picture
Єдиний коміт - очищення історії
4ad5efa
import os
import logging
import gradio as gr
from pathlib import Path
import traceback
from datetime import datetime
import pandas as pd
import uuid
import json
from typing import Dict, List, Any, Optional, Tuple, Union
from modules.ai_analysis.jira_hybrid_chat import JiraHybridChat
from modules.config.ai_settings import (
SIMILARITY_TOP_K
)
# Налаштування логування
logger = logging.getLogger("jira_assistant_interface")
# Імпорт необхідних модулів
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
logger.warning("Не вдалося імпортувати python-dotenv. Змінні середовища не будуть завантажені з .env файлу.")
try:
from modules.ai_analysis.jira_ai_report import JiraAIReport
REPORT_MODULE_AVAILABLE = True
logger.info("Успішно імпортовано JiraAIReport")
except ImportError:
REPORT_MODULE_AVAILABLE = False
logger.warning("Модуль JiraAIReport недоступний. Буде використано стандартний JiraAIAssistant для звітів.")
# Імпорт спеціалізованого Q/A асистента, якщо він доступний
try:
from modules.ai_analysis.jira_qa_assistant import JiraQAAssistant
QA_ASSISTANT_AVAILABLE = True
logger.info("Успішно імпортовано JiraQAAssistant")
except ImportError:
QA_ASSISTANT_AVAILABLE = False
logger.warning("Модуль JiraQAAssistant недоступний. Буде використано стандартний JiraAIAssistant для Q/A.")
# Імпорт LlamaIndex компонентів (перевірка чи доступні)
try:
from llama_index.core import Document, VectorStoreIndex, Settings
from llama_index.core.llms import ChatMessage
LLAMA_INDEX_AVAILABLE = True
except ImportError:
LLAMA_INDEX_AVAILABLE = False
logger.warning("LlamaIndex не доступний. Деякі функції можуть бути недоступні.")
# Допоміжні функції
def strip_assistant_prefix(text):
"""Видаляє префікс 'assistant:' з тексту відповіді"""
if isinstance(text, str) and text.startswith("assistant:"):
return text.replace("assistant:", "", 1).strip()
return text
def get_indices_dir(timestamp=None):
"""
Формує шлях до директорії для збереження індексів.
Args:
timestamp (str, optional): Часова мітка для унікальної ідентифікації.
Якщо None, буде створена автоматично.
Returns:
str: Шлях до директорії індексів
"""
# Якщо часова мітка не вказана, створюємо нову
if timestamp is None:
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
# Формуємо шлях до директорії
indices_dir = Path("temp") / "indices" / timestamp
# Створюємо директорію, якщо вона не існує
os.makedirs(indices_dir, exist_ok=True)
return str(indices_dir)
# Функції для роботи з індексами FAISS
def try_import_faiss_utils():
"""Імпортує FAISS утиліти, якщо вони доступні"""
try:
from modules.ai_analysis.faiss_utils import (
find_latest_indices,
find_indices_by_hash,
cleanup_old_indices,
generate_file_hash,
save_indices_metadata
)
return {
"find_latest_indices": find_latest_indices,
"find_indices_by_hash": find_indices_by_hash,
"cleanup_old_indices": cleanup_old_indices,
"generate_file_hash": generate_file_hash,
"save_indices_metadata": save_indices_metadata
}
except ImportError as faiss_err:
logger.warning(f"Не вдалося імпортувати FAISS утиліти: {faiss_err}. Будуть використані стандартні методи.")
return None
# Клас для управління сесіями
class UserSessionManager:
"""Управління сесіями користувачів"""
def __init__(self):
self.user_sessions = {}
def get_or_create_user_session(self, user_id=None):
"""
Отримує існуючу сесію або створює нову.
Args:
user_id (str, optional): ID користувача. Якщо не вказано, генерується випадковий.
Returns:
str: ID сесії
"""
# Якщо ID користувача не вказано, генеруємо випадковий
if not user_id:
user_id = str(uuid.uuid4())
# Якщо сесія вже існує, повертаємо її
if user_id in self.user_sessions:
return self.user_sessions[user_id]
# Інакше створюємо нову сесію
session_id = f"{user_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}"
self.user_sessions[user_id] = {"session_id": session_id, "chat_history": []}
logger.info(f"Створено нову сесію {session_id} для користувача {user_id}")
return self.user_sessions[user_id]
def get_chat_history(self, user_id):
"""Отримує історію чату користувача"""
session = self.get_or_create_user_session(user_id)
return session.get("chat_history", [])
def update_chat_history(self, user_id, message, response):
"""Оновлює історію чату користувача"""
session = self.get_or_create_user_session(user_id)
if "chat_history" not in session:
session["chat_history"] = []
session["chat_history"].append({"role": "user", "content": message})
session["chat_history"].append({"role": "assistant", "content": response})
return session["chat_history"]
# Клас для інтеграції AI асистентів
class AIAssistantIntegration:
"""Інтеграція різних AI асистентів та інтерфейсу"""
def __init__(self, app):
"""
Ініціалізація інтеграції.
Args:
app: Екземпляр JiraAssistantApp
"""
self.app = app
self.session_manager = UserSessionManager()
# Отримуємо ключі API з .env
self.api_key_openai = os.getenv("OPENAI_API_KEY", "")
self.api_key_gemini = os.getenv("GEMINI_API_KEY", "")
# Імпортуємо FAISS утиліти, якщо доступні
self.faiss_utils = try_import_faiss_utils()
self.faiss_utils_available = self.faiss_utils is not None
if self.faiss_utils_available:
logger.info("FAISS утиліти успішно імпортовано")
def run_full_context_qa(self, question, model_type, temperature):
"""
Запускає режим Q/A з повним контекстом.
Args:
question (str): Питання користувача
model_type (str): Тип моделі
temperature (float): Температура генерації
Returns:
str: Відповідь на питання
"""
# Перевіряємо, чи є завантажений файл або дані
if (not hasattr(self.app, 'last_loaded_csv') or self.app.last_loaded_csv is None) and \
(not hasattr(self.app, 'current_data') or self.app.current_data is None):
return "Помилка: спочатку завантажте CSV файл у вкладці 'CSV Аналіз' або ініціалізуйте дані з локальних файлів"
if not question or question.strip() == "":
return "Будь ласка, введіть питання."
try:
# Перевіряємо доступність спеціалізованого Q/A асистента
if QA_ASSISTANT_AVAILABLE:
return self._run_qa_with_specialized_assistant(question, model_type, temperature)
else:
return self._run_qa_with_standard_assistant(question, model_type, temperature)
except Exception as e:
error_msg = f"Помилка при виконанні запиту: {str(e)}\n\n{traceback.format_exc()}"
logger.error(error_msg)
return error_msg
def _run_qa_with_specialized_assistant(self, question, model_type, temperature):
"""Виконує Q/A з використанням спеціалізованого асистента JiraQAAssistant"""
qa_assistant = JiraQAAssistant(
api_key_openai=self.api_key_openai,
api_key_gemini=self.api_key_gemini,
model_type=model_type,
temperature=float(temperature)
)
# Отримання даних для аналізу
if hasattr(self.app, 'current_data') and self.app.current_data is not None:
# Завантаження даних з DataFrame
logger.info("Використовуємо DataFrame з пам'яті для Q/A")
success = qa_assistant.load_documents_from_dataframe(self.app.current_data)
if not success:
return "Помилка: не вдалося завантажити дані з DataFrame"
else:
# Завантаження даних з файлу
temp_file_path = self.app.last_loaded_csv
if not os.path.exists(temp_file_path):
return f"Помилка: файл {temp_file_path} не знайдено"
# Зчитуємо DataFrame з файлу
df = pd.read_csv(temp_file_path)
success = qa_assistant.load_documents_from_dataframe(df)
if not success:
return "Помилка: не вдалося завантажити дані з CSV файлу"
# Виконуємо Q/A запит
result = qa_assistant.run_qa(question)
if "error" in result:
return f"Помилка: {result['error']}"
# Форматуємо відповідь з інформацією про токени
answer = result["answer"]
answer = strip_assistant_prefix(answer) # Видаляємо префікс "assistant:"
metadata = result["metadata"]
tokens_info = f"\n\n---\n*Використано токенів: питання={metadata['question_tokens']}, "
tokens_info += f"контекст={metadata['context_tokens']}, "
tokens_info += f"відповідь={metadata['response_tokens']}, "
tokens_info += f"всього={metadata['total_tokens']}*"
return answer + tokens_info
def _run_qa_with_standard_assistant(self, question, model_type, temperature):
"""Виконує Q/A з використанням стандартного асистента JiraAIAssistant"""
# Перевіряємо доступність індексів
indices_path = None
# 1. Спочатку перевіряємо шлях до індексів в додатку
if hasattr(self.app, 'indices_path') and self.app.indices_path and os.path.exists(self.app.indices_path):
indices_path = self.app.indices_path
logger.info(f"Використовуємо наявні індекси з app.indices_path: {indices_path}")
# 2. Перевіряємо індекси, пов'язані з сесією
elif hasattr(self.app, 'current_session_id') and self.app.current_session_id:
session_indices_dir = Path("temp/sessions") / self.app.current_session_id / "indices"
if session_indices_dir.exists():
indices_path = str(session_indices_dir)
logger.info(f"Використовуємо індекси сесії: {indices_path}")
# Зберігаємо шлях для майбутнього використання
self.app.indices_path = indices_path
# 3. Якщо є шлях до завантаженого файлу, шукаємо індекси для нього
elif hasattr(self.app, 'last_loaded_csv') and self.app.last_loaded_csv and os.path.exists(self.app.last_loaded_csv):
if self.faiss_utils_available:
try:
file_path = self.app.last_loaded_csv
csv_hash = self.faiss_utils["generate_file_hash"](file_path)
if csv_hash:
indices_exist, found_indices_path = self.faiss_utils["find_indices_by_hash"](csv_hash)
if indices_exist:
indices_path = found_indices_path
logger.info(f"Знайдено індекси за хешем CSV: {indices_path}")
# Зберігаємо шлях для майбутнього використання
self.app.indices_path = indices_path
except Exception as e:
logger.warning(f"Помилка при пошуку індексів за хешем: {e}")
# Підготовка об'єкту для кешування індексів, якщо його немає
if not hasattr(self, "_indices_cache"):
self._indices_cache = {}
# Пріоритет віддаємо кешованим індексам
assistant = None
if indices_path and indices_path in self._indices_cache:
# Використовуємо кешований асистент
assistant = self._indices_cache[indices_path]
# Оновлюємо параметри
assistant.model_type = model_type
assistant.temperature = float(temperature)
assistant._initialize_llm()
logger.info(f"Використовуємо кешований асистент для {indices_path}")
else:
# Створення нового асистента
assistant = JiraHybridChat(
api_key_openai=self.api_key_openai,
api_key_gemini=self.api_key_gemini,
model_type=model_type,
temperature=float(temperature)
)
# Спроба використання індексів
if indices_path and os.path.exists(indices_path):
# Завантажуємо індекси
logger.info(f"Спроба завантажити індекси з шляху: {indices_path}")
success = assistant.load_indices(indices_path)
if success and hasattr(assistant, 'index') and assistant.index is not None:
logger.info(f"Успішно завантажено індекси з {indices_path}")
# Також завантажуємо DataFrame для повної функціональності
if hasattr(self.app, 'current_data') and self.app.current_data is not None:
assistant.df = self.app.current_data
# Додаємо в кеш
self._indices_cache[indices_path] = assistant
logger.info(f"Додано асистента в кеш для {indices_path}")
else:
logger.warning(f"Не вдалося завантажити індекси з {indices_path}")
# Якщо не вдалося завантажити індекси, завантажуємо дані напряму
if not hasattr(assistant, 'index') or assistant.index is None:
if hasattr(self.app, 'current_data') and self.app.current_data is not None:
# Завантаження даних з DataFrame
logger.info("Використовуємо DataFrame з пам'яті")
success = assistant.load_data_from_dataframe(self.app.current_data)
if not success:
return "Помилка: не вдалося завантажити дані з DataFrame"
else:
# Завантаження даних з файлу
temp_file_path = self.app.last_loaded_csv
if not os.path.exists(temp_file_path):
return f"Помилка: файл {temp_file_path} не знайдено"
# Завантаження даних з CSV
success = assistant.load_data_from_csv(temp_file_path)
if not success:
return "Помилка: не вдалося завантажити дані з CSV-файлу. Перевірте формат файлу."
# Виконуємо запит
result = assistant.run_full_context_qa(question)
if "error" in result:
return f"Помилка: {result['error']}"
# Форматуємо відповідь з інформацією про токени
answer = result["answer"]
answer = strip_assistant_prefix(answer) # Видаляємо префікс "assistant:"
metadata = result["metadata"]
tokens_info = f"\n\n---\n*Використано токенів: питання={metadata['question_tokens']}, "
tokens_info += f"контекст={metadata['context_tokens']}, "
tokens_info += f"відповідь={metadata['response_tokens']}, "
tokens_info += f"всього={metadata['total_tokens']}*"
# Зберігаємо індекси, якщо вони створені і ще не збережені
if not indices_path and hasattr(assistant, 'index') and assistant.index is not None:
if self.faiss_utils_available and hasattr(self.app, 'last_loaded_csv'):
try:
file_path = self.app.last_loaded_csv
csv_hash = self.faiss_utils["generate_file_hash"](file_path)
if csv_hash and hasattr(assistant, 'save_indices'):
logger.info("Зберігаємо індекси для майбутнього використання")
new_indices_dir = get_indices_dir()
if assistant.save_indices(new_indices_dir):
# Збережемо метадані з хешем CSV
metadata_obj = {
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"csv_hash": csv_hash,
"document_count": len(assistant.jira_documents) if hasattr(assistant, 'jira_documents') else 0,
"storage_format": "binary" # Додаємо інформацію про формат зберігання
}
self.faiss_utils["save_indices_metadata"](new_indices_dir, metadata_obj)
logger.info(f"Індекси збережено у {new_indices_dir}")
# Зберігаємо шлях для майбутнього використання
self.app.indices_path = new_indices_dir
# Додаємо в кеш
self._indices_cache[new_indices_dir] = assistant
# Очистимо старі індекси
self.faiss_utils["cleanup_old_indices"](max_indices=3)
except Exception as save_err:
logger.warning(f"Помилка при збереженні індексів: {save_err}")
return answer + tokens_info
def process_chat_message(self, message, chat_history, model_type, temperature):
"""
Обробка повідомлення користувача в чаті.
Args:
message (str): Повідомлення користувача
chat_history (list): Історія чату у форматі Gradio
model_type (str): Тип моделі
temperature (float): Температура генерації
Returns:
tuple: (очищене поле вводу, оновлена історія чату)
"""
if not message or message.strip() == "":
return "", chat_history
# Перевіряємо, чи є завантажений файл або дані
if (not hasattr(self.app, 'last_loaded_csv') or self.app.last_loaded_csv is None) and \
(not hasattr(self.app, 'current_data') or self.app.current_data is None):
chat_history.append((message, "Помилка: спочатку завантажте CSV файл у вкладці 'CSV Аналіз' або ініціалізуйте дані з локальних файлів"))
return "", chat_history
try:
# Використовуємо jira_hybrid_chat для обробки запиту
from modules.ai_analysis.jira_hybrid_chat import JiraHybridChat
# Створюємо екземпляр чату з передачею app
chat = JiraHybridChat(
indices_dir=self.app.indices_path if hasattr(self.app, 'indices_path') else None,
app=self.app, # Передаємо app для доступу до current_data
api_key_openai=self.api_key_openai,
api_key_gemini=self.api_key_gemini,
model_type=model_type,
temperature=float(temperature)
)
# Конвертуємо історію чату у формат для асистента
formatted_history = []
for user_msg, ai_msg in chat_history:
formatted_history.append({"role": "user", "content": user_msg})
formatted_history.append({"role": "assistant", "content": ai_msg})
# Отримуємо відповідь
result = chat.chat_with_hybrid_search(message, formatted_history)
if "error" in result:
chat_history.append((message, f"Помилка: {result['error']}"))
return "", chat_history
# Форматуємо відповідь з інформацією про токени
answer = result["answer"]
metadata = result["metadata"]
# Додаємо інформацію про релевантні документи
docs_info = "\n\n*Релевантні документи:*\n"
if "relevant_documents" in metadata:
for doc in metadata['relevant_documents'][:SIMILARITY_TOP_K]: # Показуємо топ-3 документа
docs_info += f"*{doc.get('rank', '?')}.* [{doc.get('ticket_id', '?')}](https://jira.healthprecision.net/browse/{doc.get('ticket_id', '?')}) "
docs_info += f"(релевантність: {doc.get('relevance', 0):.4f}): {doc.get('summary', '')[:50]}...\n"
# Додаємо інформацію про токени
tokens_info = f"\n\n---\n*Використано токенів: питання={metadata.get('question_tokens', 0)}, "
tokens_info += f"контекст={metadata.get('context_tokens', 0)}, "
tokens_info += f"відповідь={metadata.get('response_tokens', 0)}, "
tokens_info += f"всього={metadata.get('total_tokens', 0)}*"
# Формуємо повну відповідь
full_answer = answer + docs_info + tokens_info
# Оновлюємо історію чату
chat_history.append((message, full_answer))
# Зберігаємо індекси, якщо вони створені і ще не збережені
if not hasattr(self.app, 'indices_path') and hasattr(chat, 'index') and chat.index is not None:
if hasattr(self, 'faiss_utils_available') and self.faiss_utils_available and hasattr(self.app, 'last_loaded_csv'):
try:
file_path = self.app.last_loaded_csv
csv_hash = self.faiss_utils["generate_file_hash"](file_path)
if csv_hash and hasattr(chat, 'save_indices'):
logger.info("Зберігаємо індекси для майбутнього використання")
new_indices_dir = get_indices_dir()
if chat.save_indices(new_indices_dir):
# Збережемо метадані з хешем CSV
metadata_obj = {
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"csv_hash": csv_hash,
"document_count": len(chat.jira_documents) if hasattr(chat, 'jira_documents') else 0,
"storage_format": "binary"
}
self.faiss_utils["save_indices_metadata"](new_indices_dir, metadata_obj)
logger.info(f"Індекси збережено у {new_indices_dir}")
# Зберігаємо шлях для майбутнього використання
self.app.indices_path = new_indices_dir
except Exception as save_err:
logger.warning(f"Помилка при збереженні індексів: {save_err}")
return "", chat_history
except Exception as e:
import traceback
error_msg = f"Помилка при обробці повідомлення: {str(e)}\n\n{traceback.format_exc()}"
logger.error(error_msg)
chat_history.append((message, f"Помилка: {str(e)}"))
return "", chat_history
def generate_ai_report(self, format_type, model_type, temperature):
"""
Генерація аналітичного звіту на основі даних Jira.
Args:
format_type (str): Формат звіту ("markdown", "html")
model_type (str): Тип моделі для використання
temperature (float): Температура для генерації
Returns:
str: Згенерований звіт або повідомлення про помилку
"""
try:
# Перевіряємо, чи є завантажений файл або дані
if (not hasattr(self.app, 'last_loaded_csv') or self.app.last_loaded_csv is None) and \
(not hasattr(self.app, 'current_data') or self.app.current_data is None):
return "Помилка: спочатку завантажте CSV файл у вкладці 'CSV Аналіз' або ініціалізуйте дані з локальних файлів"
# Використовуємо спеціалізований генератор звітів, якщо доступний
if REPORT_MODULE_AVAILABLE:
logger.info("Використовуємо спеціалізований модуль JiraAIReport для генерації звіту")
# Створення генератора звітів
report_generator = JiraAIReport(
api_key_openai=self.api_key_openai,
api_key_gemini=self.api_key_gemini,
model_type=model_type,
temperature=float(temperature)
)
# Завантаження даних
if hasattr(self.app, 'current_data') and self.app.current_data is not None:
# Завантаження даних з DataFrame
logger.info("Використовуємо DataFrame з пам'яті для генерації звіту")
success = report_generator.load_documents_from_dataframe(self.app.current_data)
if not success:
return "Помилка: не вдалося завантажити дані з DataFrame"
else:
# Завантаження даних з файлу
logger.info(f"Читаємо CSV файл: {self.app.last_loaded_csv}")
df = pd.read_csv(self.app.last_loaded_csv)
success = report_generator.load_documents_from_dataframe(df)
if not success:
return "Помилка: не вдалося завантажити дані з CSV файлу"
# Генерація звіту
result = report_generator.generate_report(format_type=format_type)
if "error" in result:
return f"Помилка: {result['error']}"
# Форматуємо відповідь з інформацією про токени
report = result["report"]
report = strip_assistant_prefix(report) # Видаляємо префікс "assistant:"
metadata = result["metadata"]
# Додаємо інформацію про токени
tokens_info = f"\n\n---\n*Використано токенів: контекст={metadata['context_tokens']}, "
tokens_info += f"звіт={metadata['report_tokens']}, "
tokens_info += f"всього={metadata['total_tokens']}, "
tokens_info += f"проаналізовано документів: {metadata['documents_used']}*"
if format_type.lower() == "markdown":
return report + tokens_info
else:
# Для HTML додаємо інформацію про токени внизу
tokens_html = f'<div style="margin-top: 20px; color: #666; font-size: 0.9em;">'
tokens_html += f'Використано токенів: контекст={metadata["context_tokens"]}, '
tokens_html += f'звіт={metadata["report_tokens"]}, '
tokens_html += f'всього={metadata["total_tokens"]}, '
tokens_html += f'проаналізовано документів: {metadata["documents_used"]}'
tokens_html += '</div>'
return report + tokens_html
else:
# Використовуємо стандартний механізм генерації звітів
logger.warning("Модуль JiraAIReport недоступний, використовуємо стандартний JiraAIAssistant")
# Створення асистента
assistant = JiraHybridChat(
api_key_openai=self.api_key_openai,
api_key_gemini=self.api_key_gemini,
model_type=model_type,
temperature=float(temperature)
)
# Завантаження даних
if hasattr(self.app, 'current_data') and self.app.current_data is not None:
# Завантаження даних з DataFrame
success = assistant.load_data_from_dataframe(self.app.current_data)
if not success:
return "Помилка: не вдалося завантажити дані з DataFrame"
else:
# Завантаження даних з файлу
success = assistant.load_data_from_csv(self.app.last_loaded_csv)
if not success:
return "Помилка: не вдалося завантажити дані з файлу"
# Отримуємо статистику для звіту
stats = assistant.get_statistics()
# Підготовка даних для звіту
data_summary = f"СТАТИСТИКА ПРОЕКТУ JIRA:\n\n"
data_summary += f"Загальна кількість тікетів: {stats['document_count']}\n\n"
data_summary += "Розподіл за статусами:\n"
for status, count in stats['status_counts'].items():
percentage = (count / stats['document_count'] * 100) if stats['document_count'] > 0 else 0
data_summary += f"- {status}: {count} ({percentage:.1f}%)\n"
data_summary += "\nРозподіл за типами:\n"
for type_name, count in stats['type_counts'].items():
percentage = (count / stats['document_count'] * 100) if stats['document_count'] > 0 else 0
data_summary += f"- {type_name}: {count} ({percentage:.1f}%)\n"
data_summary += "\nРозподіл за пріоритетами:\n"
for priority, count in stats['priority_counts'].items():
percentage = (count / stats['document_count'] * 100) if stats['document_count'] > 0 else 0
data_summary += f"- {priority}: {count} ({percentage:.1f}%)\n"
data_summary += "\nТоп виконавці завдань:\n"
for assignee, count in stats['top_assignees'].items():
data_summary += f"- {assignee}: {count} тікетів\n"
# Генерація звіту
result = assistant.generate_report(data_summary, format_type=format_type)
if "error" in result:
return f"Помилка: {result['error']}"
# Форматуємо відповідь з інформацією про токени
report = result["report"]
report = strip_assistant_prefix(report) # Видаляємо префікс "assistant:"
metadata = result["metadata"]
# Додаємо інформацію про токени
tokens_info = f"\n\n---\n*Використано токенів: контекст={metadata['context_tokens']}, "
tokens_info += f"звіт={metadata['report_tokens']}, "
tokens_info += f"всього={metadata['total_tokens']}*"
if format_type.lower() == "markdown":
return report + tokens_info
else:
# Для HTML додаємо інформацію про токени внизу
tokens_html = f'<div style="margin-top: 20px; color: #666; font-size: 0.9em;">'
tokens_html += f'Використано токенів: контекст={metadata["context_tokens"]}, '
tokens_html += f'звіт={metadata["report_tokens"]}, '
tokens_html += f'всього={metadata["total_tokens"]}'
tokens_html += '</div>'
return report + tokens_html
except Exception as e:
error_msg = f"Помилка при генерації звіту: {str(e)}\n\n{traceback.format_exc()}"
logger.error(error_msg)
return error_msg
def setup_ai_assistant_tab(app, interface):
"""
Налаштування вкладки AI асистентів з підтримкою Q/A, чату та звітів.
Args:
app: Екземпляр JiraAssistantApp
interface: Блок інтерфейсу Gradio
Returns:
bool: True якщо ініціалізація пройшла успішно
"""
try:
# Створюємо інтеграцію AI асистента
ai_integration = AIAssistantIntegration(app)
# Створюємо вкладку для AI асистентів
with gr.Tab("AI Асистенти"):
gr.Markdown("## AI Асистенти для Jira")
# Спільні параметри для всіх режимів (в один рядок)
with gr.Row():
model_type = gr.Dropdown(
choices=["gemini", "openai"],
value="gemini",
label="Модель LLM",
scale=1
)
temperature = gr.Slider(
minimum=0.0,
maximum=1.0,
value=0.2,
step=0.1,
label="Температура",
scale=2
)
# Інформація про необхідність завантажити файл
gr.Markdown("""
**❗ Примітка:** Для роботи AI асистентів спочатку завантажте CSV файл у вкладці "CSV Аналіз" або ініціалізуйте дані з локальних файлів
""")
# Розділяємо режими по вкладках
with gr.Tabs():
with gr.Tab("Q/A з повним контекстом"):
gr.Markdown("""
**У цьому режимі бот має доступ до всіх даних тікетів одночасно.**
Використовуйте цей режим для загальних питань про проект,
статистику, тренди та загальний аналіз.
""")
qa_question = gr.Textbox(
label="Ваше питання",
placeholder="Наприклад: Які тікети мають найвищий пріоритет?",
lines=3
)
qa_button = gr.Button("Отримати відповідь")
qa_answer = gr.Markdown(label="Відповідь")
# Прив'язуємо обробник
qa_button.click(
ai_integration.run_full_context_qa,
inputs=[qa_question, model_type, temperature],
outputs=[qa_answer]
)
with gr.Tab("Гібридний чат"):
gr.Markdown("""
**У цьому режимі бот використовує гібридний пошук (BM25 + векторний) для кращої якості результатів.**
Гібридний пошук поєднує переваги пошуку за ключовими словами та семантичного векторного пошуку.
Підходить для більшості запитів, забезпечуючи високу релевантність відповідей.
""")
# Використовуємо компонент Chatbot для історії повідомлень
chatbot = gr.Chatbot(
height=500,
avatar_images=["Human:", "AI:"]
)
# Поле для вводу повідомлення
msg = gr.Textbox(
placeholder="Після введення питанні натисність Shift+Enter",
lines=2,
show_label=False,
)
# Кнопка очищення історії
clear = gr.Button("Очистити історію")
# Прив'язуємо обробники
msg.submit(
ai_integration.process_chat_message,
inputs=[msg, chatbot, model_type, temperature],
outputs=[msg, chatbot]
)
# Функція для очищення історії чату
clear.click(lambda: [], None, chatbot, queue=False)
with gr.Tab("Генерація звіту"):
gr.Markdown("""
**Автоматична генерація аналітичного звіту на основі даних Jira.**
AI проаналізує дані CSV файлу та створить структурований звіт.
""")
with gr.Row():
format_type = gr.Radio(
choices=["markdown", "html"],
value="markdown",
label="Формат звіту"
)
report_button = gr.Button("Згенерувати звіт")
ai_report = gr.Markdown(label="Звіт", elem_id="ai_report_output")
# Додаємо CSS для стилізації звіту
gr.HTML("""
<style>
#ai_report_output {
height: 600px;
overflow-y: auto;
border: 1px solid #ddd;
padding: 20px;
border-radius: 4px;
background-color: #f9f9f9;
}
</style>
""")
# Прив'язуємо обробник
report_button.click(
ai_integration.generate_ai_report,
inputs=[format_type, model_type, temperature],
outputs=[ai_report]
)
return True
except ImportError as e:
logger.error(f"Помилка імпорту модулів для AI асистента: {e}")
# Якщо не вдалося імпортувати модулі, створюємо заглушку
with gr.Tab("AI Асистенти"):
gr.Markdown("## AI Асистенти для Jira")
gr.Markdown(f"""
### ⚠️ Потрібні додаткові залежності
Для роботи AI асистентів потрібно встановити додаткові бібліотеки:
```bash
pip install llama-index-llms-gemini llama-index llama-index-embeddings-openai llama-index-retrievers-bm25 llama-index-vector-stores-faiss faiss-cpu tiktoken
```
Помилка: {str(e)}
""")
return False