Spaces:
Running
Running
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 | |