import os import json import logging import requests from typing import Dict, List, Any, Optional import openai 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-3.5-turbo" # Стандартна модель 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-pro" # Стандартна модель для 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 response = openai.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": """Ви аналітик Jira з досвідом у процесах розробки ПЗ. Проаналізуйте надані дані про тікети та надайте корисні інсайти та рекомендації для покращення процесу. Будьте конкретними та орієнтованими на дії. Виділіть сильні та слабкі сторони, а також потенційні ризики та можливості. Аналіз повинен бути структурованим і легким для сприйняття менеджерами проекту."""}, {"role": "user", "content": f"Проаналізуйте наступні дані Jira та надайте рекомендації:\n\n{data_summary}"} ], temperature=temperature, ) # Отримання результату 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" # API endpoint url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model}:generateContent?key={self.api_key}" # Формування запиту payload = { "contents": [ { "parts": [ { "text": """Ви аналітик Jira з досвідом у процесах розробки ПЗ. Проаналізуйте надані дані про тікети та надайте корисні інсайти та рекомендації для покращення процесу. Будьте конкретними та орієнтованими на дії. Виділіть сильні та слабкі сторони, а також потенційні ризики та можливості. Аналіз повинен бути структурованим і легким для сприйняття менеджерами проекту.""" } ] }, { "parts": [ { "text": f"Проаналізуйте наступні дані Jira та надайте рекомендації:\n\n{data_summary}" } ] } ], "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("Успішно отримано аналіз від Gemini") return result logger.error(f"Помилка при взаємодії з Gemini: {response.text}") return f"Помилка при взаємодії з Gemini: статус {response.status_code}" except Exception as e: logger.error(f"Помилка при взаємодії з Gemini: {e}") return f"Помилка при взаємодії з Gemini: {str(e)}" 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 = openai.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 = openai.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 = openai.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": # TODO: Реалізувати для Gemini return f"Gemini API для агента '{agent_name}' ще не реалізовано" except Exception as e: logger.error(f"Помилка при виконанні задачі агентом '{agent_name}': {e}") return f"Помилка при виконанні задачі: {str(e)}"