import os import json import logging import requests from typing import Dict, List, Any, Optional from datetime import datetime, timedelta logger = logging.getLogger(__name__) class LLMConnector: """ Клас для взаємодії з LLM (OpenAI, Google Gemini, тощо) """ def __init__(self, api_key=None, model_type="openai"): """ Ініціалізація з'єднання з LLM. Args: api_key (str): API ключ для доступу до LLM. Якщо None, спробує використати змінну середовища. model_type (str): Тип моделі ("openai" або "gemini") """ self.model_type = model_type.lower() if self.model_type == "openai": self.api_key = api_key or os.getenv("OPENAI_API_KEY") if not self.api_key: logger.warning("API ключ OpenAI не вказано") self.model = "gpt-4o-mini" # Оновлено стандартну модель # Використовуємо новий спосіб ініціалізації клієнта OpenAI # без проксі та інших застарілих параметрів from openai import OpenAI self.client = OpenAI(api_key=self.api_key) elif self.model_type == "gemini": self.api_key = api_key or os.getenv("GEMINI_API_KEY") if not self.api_key: logger.warning("API ключ Gemini не вказано") self.model = "gemini-2.0-flash" # Оновлена модель для Gemini else: raise ValueError(f"Непідтримуваний тип моделі: {model_type}") def analyze_jira_data(self, stats, inactive_issues, temperature=0.2): """ Аналіз даних Jira за допомогою LLM. Args: stats (dict): Базова статистика даних Jira inactive_issues (dict): Дані про неактивні тікети temperature (float): Параметр температури для генерації Returns: str: Результат аналізу """ try: # Підготовка даних для аналізу data_summary = self._prepare_data_for_llm(stats, inactive_issues) # Відправлення запиту до LLM if self.model_type == "openai": return self._analyze_with_openai(data_summary, temperature) elif self.model_type == "gemini": return self._analyze_with_gemini(data_summary, temperature) except Exception as e: logger.error(f"Помилка при аналізі даних за допомогою LLM: {e}") return f"Помилка при аналізі даних: {str(e)}" def _prepare_data_for_llm(self, stats, inactive_issues): """ Підготовка даних для аналізу за допомогою LLM. Args: stats (dict): Базова статистика inactive_issues (dict): Дані про неактивні тікети Returns: str: Дані для аналізу у текстовому форматі """ summary = [] # Додавання загальної статистики summary.append(f"Загальна кількість тікетів: {stats.get('total_tickets', 'Невідомо')}") # Додавання статистики за статусами if 'status_counts' in stats and stats['status_counts']: summary.append("\nТікети за статусами:") for status, count in stats['status_counts'].items(): summary.append(f"- {status}: {count}") # Додавання статистики за типами if 'type_counts' in stats and stats['type_counts']: summary.append("\nТікети за типами:") for issue_type, count in stats['type_counts'].items(): summary.append(f"- {issue_type}: {count}") # Додавання статистики за пріоритетами if 'priority_counts' in stats and stats['priority_counts']: summary.append("\nТікети за пріоритетами:") for priority, count in stats['priority_counts'].items(): summary.append(f"- {priority}: {count}") # Аналіз створених тікетів if 'created_stats' in stats and stats['created_stats']: summary.append("\nСтатистика створених тікетів:") for key, value in stats['created_stats'].items(): if key == 'last_7_days': summary.append(f"- Створено за останні 7 днів: {value}") elif key == 'min': summary.append(f"- Найраніший тікет створено: {value}") elif key == 'max': summary.append(f"- Найпізніший тікет створено: {value}") # Аналіз неактивних тікетів if inactive_issues: total_inactive = inactive_issues.get('total_count', 0) percentage = inactive_issues.get('percentage', 0) summary.append(f"\nНеактивні тікети: {total_inactive} ({percentage}% від загальної кількості)") if 'by_status' in inactive_issues and inactive_issues['by_status']: summary.append("Неактивні тікети за статусами:") for status, count in inactive_issues['by_status'].items(): summary.append(f"- {status}: {count}") if 'top_inactive' in inactive_issues and inactive_issues['top_inactive']: summary.append("\nНайбільш неактивні тікети:") for i, ticket in enumerate(inactive_issues['top_inactive']): key = ticket.get('key', 'Невідомо') status = ticket.get('status', 'Невідомо') days = ticket.get('days_inactive', 'Невідомо') summary.append(f"- {key} (Статус: {status}, Днів неактивності: {days})") return "\n".join(summary) def _analyze_with_openai(self, data_summary, temperature=0.2): """ Аналіз даних за допомогою OpenAI. Args: data_summary (str): Дані для аналізу temperature (float): Параметр температури Returns: str: Результат аналізу """ try: if not self.api_key: return "Не вказано API ключ OpenAI" # Створення запиту до LLM з новим API system_prompt = """Ви аналітик Jira з досвідом у процесах розробки ПЗ. Проаналізуйте надані дані про тікети та надайте корисні інсайти та рекомендації для покращення процесу. Будьте конкретними та орієнтованими на дії. Виділіть сильні та слабкі сторони, а також потенційні ризики та можливості. Аналіз повинен бути структурованим і легким для сприйняття менеджерами проекту.""" user_prompt = f"Проаналізуйте наступні дані Jira та надайте рекомендації:\n\n{data_summary}" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ] # Надсилання запиту через новий клієнт OpenAI response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, max_tokens=2048 ) # Отримання результату analysis_result = response.choices[0].message.content logger.info("Успішно отримано аналіз від OpenAI") return analysis_result except Exception as e: logger.error(f"Помилка при взаємодії з OpenAI: {e}") return f"Помилка при взаємодії з OpenAI: {str(e)}" def _analyze_with_gemini(self, data_summary, temperature=0.2): """ Аналіз даних за допомогою Google Gemini. Args: data_summary (str): Дані для аналізу temperature (float): Параметр температури Returns: str: Результат аналізу """ try: if not self.api_key: return "Не вказано API ключ Gemini" # Імпортуємо необхідні бібліотеки try: from google import genai from google.genai import types except ImportError: logger.error("Бібліотека google-generativeai не встановлена") return "Помилка: бібліотека google-generativeai не встановлена. Встановіть її командою: pip install google-generativeai" # Налаштування клієнта client = genai.Client(api_key=self.api_key) # Додаємо логування для діагностики logger.info(f"Відправляємо запит до Gemini API. Довжина запиту: {len(data_summary)} символів") # Системна інструкція system_instruction = """Ви аналітик Jira з досвідом у процесах розробки ПЗ. Проаналізуйте надані дані про тікети та надайте корисні інсайти та рекомендації для покращення процесу. Будьте конкретними та орієнтованими на дії. Виділіть сильні та слабкі сторони, а також потенційні ризики та можливості. Аналіз повинен бути структурованим і легким для сприйняття менеджерами проекту.""" # Запит користувача user_prompt = f"Проаналізуйте наступні дані Jira та надайте рекомендації:\n\n{data_summary}" # Створення запиту contents = [ types.Content( role="user", parts=[types.Part.from_text(text=user_prompt)] ) ] # Налаштування генерації з системною інструкцією generate_content_config = types.GenerateContentConfig( temperature=temperature, top_p=0.95, top_k=40, max_output_tokens=2048, response_mime_type="text/plain", system_instruction=[ types.Part.from_text(text=system_instruction) ], ) # Відправка запиту try: response = client.models.generate_content( model=self.model, contents=contents, config=generate_content_config, ) # Перевірка відповіді if hasattr(response, 'text') and response.text: logger.info("Успішно отримано аналіз від Gemini") return response.text else: logger.error("Порожня відповідь від Gemini API") return "Помилка: порожня відповідь від Gemini API" except Exception as api_error: logger.error(f"Помилка API Gemini: {str(api_error)}") # Спробуємо з іншою моделлю, якщо поточна не працює try: logger.info("Спроба з альтернативною моделлю gemini-1.5-flash") alternative_model = "gemini-1.5-flash" response = client.models.generate_content( model=alternative_model, contents=contents, config=generate_content_config, ) if hasattr(response, 'text') and response.text: logger.info("Успішно отримано аналіз від альтернативної моделі Gemini") return response.text else: return "Помилка: порожня відповідь від альтернативної моделі Gemini API" except Exception as retry_error: logger.error(f"Повторна помилка API Gemini: {str(retry_error)}") return f"Помилка при взаємодії з Gemini: {str(api_error)}" except Exception as e: import traceback error_msg = f"Помилка при взаємодії з Gemini: {str(e)}\n\n{traceback.format_exc()}" logger.error(error_msg) return error_msg def ask_question(self, question, context=None, temperature=0.3): """ Задати питання до LLM на основі даних Jira. Args: question (str): Питання користувача context (str): Додатковий контекст (може містити JSON дані тікетів) temperature (float): Параметр температури Returns: str: Відповідь на питання """ try: # Формування запиту для LLM if self.model_type == "openai": messages = [ {"role": "system", "content": """Ви асистент, який допомагає аналізувати дані Jira. Відповідайте на питання користувача на основі наданого контексту. Якщо контекст недостатній для відповіді, чесно визнайте це."""} ] if context: messages.append({"role": "user", "content": f"Контекст: {context}"}) messages.append({"role": "user", "content": question}) # Відправлення запиту response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, ) # Отримання відповіді answer = response.choices[0].message.content logger.info("Успішно отримано відповідь від OpenAI") return answer elif self.model_type == "gemini": # TODO: Реалізувати для Gemini return "Gemini API для Q&A ще не реалізовано" except Exception as e: logger.error(f"Помилка при отриманні відповіді від LLM: {e}") return f"Помилка при обробці запитання: {str(e)}" def generate_summary(self, jira_data, output_format="markdown", temperature=0.3): """ Генерація підсумкового звіту по даним Jira. Args: jira_data (str): Дані Jira для аналізу output_format (str): Формат виводу ("markdown", "html", "text") temperature (float): Параметр температури Returns: str: Згенерований звіт """ try: # Формування запиту для LLM if self.model_type == "openai": format_instruction = "" if output_format == "markdown": format_instruction = "Використовуйте Markdown для форматування звіту." elif output_format == "html": format_instruction = "Створіть звіт у форматі HTML з використанням відповідних тегів." else: format_instruction = "Створіть звіт у простому текстовому форматі." response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": f"""Ви аналітик, який створює професійні звіти на основі даних Jira. Проаналізуйте надані дані та створіть структурований звіт з наступними розділами: 1. Короткий огляд проекту 2. Аналіз поточного стану 3. Ризики та проблеми 4. Рекомендації {format_instruction}"""}, {"role": "user", "content": f"Дані для аналізу:\n\n{jira_data}"} ], temperature=temperature, ) # Отримання звіту report = response.choices[0].message.content logger.info(f"Успішно згенеровано звіт у форматі {output_format}") return report elif self.model_type == "gemini": # TODO: Реалізувати для Gemini return "Gemini API для генерації звітів ще не реалізовано" except Exception as e: logger.error(f"Помилка при генерації звіту: {e}") return f"Помилка при генерації звіту: {str(e)}" class AIAgentManager: """ Менеджер AI агентів для аналізу даних Jira """ def __init__(self, api_key=None, model_type="openai"): """ Ініціалізація менеджера AI агентів. Args: api_key (str): API ключ для LLM model_type (str): Тип моделі ("openai" або "gemini") """ self.llm_connector = LLMConnector(api_key, model_type) self.agents = { "analytics_engineer": self._create_analytics_engineer(), "project_manager": self._create_project_manager(), "communication_specialist": self._create_communication_specialist() } def _create_analytics_engineer(self): """ Створення агента "Аналітичний інженер" для обробки даних. Returns: dict: Конфігурація агента """ return { "name": "Аналітичний інженер", "role": "Обробка та аналіз даних Jira", "system_prompt": """Ви досвідчений аналітичний інженер, експерт з обробки та аналізу даних Jira. Ваша задача - отримувати, обробляти та аналізувати дані з Jira, виявляти паттерни та готувати інформацію для подальшого аналізу іншими спеціалістами. Фокусуйтеся на виявленні аномалій, трендів та інсайтів у даних.""" } def _create_project_manager(self): """ Створення агента "Проектний менеджер" для аналізу проекту. Returns: dict: Конфігурація агента """ return { "name": "Проектний менеджер", "role": "Аналіз стану проекту та вироблення рекомендацій", "system_prompt": """Ви досвідчений проектний менеджер з глибоким розумінням процесів розробки ПЗ. Ваша задача - аналізувати дані Jira, розуміти поточний стан проекту, виявляти ризики та проблеми, та надавати конкретні дієві рекомендації для покращення процесу. Вас цікавлять дедлайни, блокуючі фактори, неактивні тікети та ефективність процесу.""" } def _create_communication_specialist(self): """ Створення агента "Комунікаційний спеціаліст" для формування повідомлень. Returns: dict: Конфігурація агента """ return { "name": "Комунікаційний спеціаліст", "role": "Формування повідомлень для стейкхолдерів", "system_prompt": """Ви досвідчений комунікаційний спеціаліст, експерт з формування чітких, інформативних та професійних повідомлень для стейкхолдерів проекту. Ваша задача - перетворювати технічну інформацію та аналітику в зрозумілі, структуровані повідомлення, які допоможуть стейкхолдерам приймати рішення. Фокусуйтеся на ключових інсайтах, використовуйте професійний, але дружній тон.""" } def run_agent(self, agent_name, task, context=None, temperature=0.3): """ Запуск конкретного агента для виконання задачі. Args: agent_name (str): Назва агента ("analytics_engineer", "project_manager" або "communication_specialist") task (str): Задача для агента context (str): Контекст для виконання задачі temperature (float): Параметр температури Returns: str: Результат виконання задачі """ try: if agent_name not in self.agents: return f"Помилка: агент '{agent_name}' не знайдений" agent = self.agents[agent_name] # Формування запиту для LLM messages = [ {"role": "system", "content": agent["system_prompt"]} ] if context: messages.append({"role": "user", "content": f"Контекст: {context}"}) messages.append({"role": "user", "content": task}) # Відправлення запиту до LLM if self.llm_connector.model_type == "openai": response = self.llm_connector.client.chat.completions.create( model=self.llm_connector.model, messages=messages, temperature=temperature, ) # Отримання результату result = response.choices[0].message.content logger.info(f"Успішно отримано результат від агента '{agent_name}'") return result elif self.llm_connector.model_type == "gemini": # Реалізуємо запит до Gemini API через прямий HTTP запит url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.llm_connector.model}:generateContent?key={self.llm_connector.api_key}" # Формування системного промпта та запиту користувача system_prompt = agent["system_prompt"] user_content = "" if context: user_content += f"Контекст: {context}\n\n" user_content += task # Формування повного запиту payload = { "contents": [ { "parts": [ { "text": f"{system_prompt}\n\n{user_content}" } ] } ], "generationConfig": { "temperature": temperature, "maxOutputTokens": 2048 } } # Відправлення запиту headers = {"Content-Type": "application/json"} response = requests.post(url, headers=headers, json=payload) # Обробка відповіді if response.status_code == 200: response_data = response.json() # Отримання тексту відповіді if 'candidates' in response_data and len(response_data['candidates']) > 0: if 'content' in response_data['candidates'][0]: content = response_data['candidates'][0]['content'] if 'parts' in content and len(content['parts']) > 0: result = content['parts'][0].get('text', '') logger.info(f"Успішно отримано результат від агента '{agent_name}' (Gemini)") return result logger.error(f"Помилка при запиті до Gemini API: {response.text}") return f"Помилка при запиті до Gemini API: статус {response.status_code}" except Exception as e: logger.error(f"Помилка при виконанні задачі агентом '{agent_name}': {e}") return f"Помилка при виконанні задачі: {str(e)}"