""" Этот файл содержит классы сервисов для выполнения различных задач: - Получение субтитров, - Анализ изображений, - Суммаризация текста, - Генерация изображений. """ from llm_client import CustomLLM from typing import Dict, Any, List, Optional import requests # --- Сервисы для внешних операций --- class SubtitleService: """ Сервис для получения текущих субтитров. Заглушки. """ def get_google_meet_subtitles(self) -> Optional[str]: """ Получает текущие, не финальные субтитры из Google Meet. (Реализация требует интеграции с Google Workspace API или другим методом.) """ print("Заглушка: Получение субтитров из Google Meet...") # TODO: Реальная интеграция с Google Meet API return "Это текущие субтитры из Google Meet: 'Привет, как дела?'" def get_ms_teams_subtitles(self) -> Optional[str]: """ Получает текущие, не финальные субтитры из Microsoft Teams. (Реализация требует интеграции с Microsoft Graph API или другим методом.) """ print("Заглушка: Получение субтитров из Microsoft Teams...") # TODO: Реальная интеграция с Microsoft Teams API return "Это текущие субтитры из MS Teams: 'Отлично, спасибо!'" class ImageAnalysisService: """ Сервис для анализа изображений с использованием Gemini Vision API. """ def __init__(self, llm: CustomLLM): self.llm = llm # Ожидаем, что это будет экземпляр Gemini LLM def analyze_image(self, image_url_or_bytes: Any, prompt: str) -> str: """ Анализирует изображение с помощью Gemini Vision. `image_url_or_bytes` может быть URL, путем к файлу или байтами изображения. LangChain Gemini Vision принимает это как часть сообщения. """ if "vision" not in self.llm.config.get("capabilities", []): raise ValueError( "Для анализа изображений требуется LLM с поддержкой 'vision' (например, Gemini)." ) print(f"Анализ изображения: {image_url_or_bytes} с промптом: {prompt}") # LangChain GeminiVision ожидает контент в определенном формате # Для простоты, здесь заглушка, имитирующая обработку изображения. # В реальной реализации, image_url_or_bytes нужно будет соответствующим образом загрузить/передать. # Пример: [HumanMessage(content=[{"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": image_url_or_bytes}}])] # Или для локальных файлов: [{"type": "image_bytes", "image_bytes": {"data": base64_encoded_bytes}}] # Поскольку у нас нет реального механизма загрузки изображения здесь, # имитируем ответ LLM для демо-целей. # В реальном приложении: # messages = [ # HumanMessage( # content=[ # {"type": "text", "text": prompt}, # {"type": "image_url", "image_url": {"url": image_url_or_bytes}} # Или image_bytes # ] # ) # ] # return self.llm.invoke(messages) return f"Заглушка анализа изображения: Изображение '{image_url_or_bytes}' содержит (по результатам анализа Gemini): 'яркий пейзаж с горами и рекой' по запросу '{prompt}'." class SummarizationService: """ Сервис для суммаризации истории диалога с использованием Gemini LLM. """ def __init__(self, llm: CustomLLM): self.llm = llm # Ожидаем, что это будет экземпляр Gemini LLM def summarize_history(self, history: List[Dict[str, str]], num_recent_messages: int = 5) -> str: """ Суммирует старую историю диалога с использованием LLM. Комбинация: последние N сообщений + суммаризация старой истории. """ if self.llm.provider != "google": print( "Предупреждение: Суммаризация рекомендуется с Gemini для лучшего качества." ) # Последние N сообщений recent_messages = history[-num_recent_messages:] old_history = history[:-num_recent_messages] summarized_old_history = "" if old_history: # Формируем промпт для суммаризации старой истории old_messages_text = "\n".join( [f"{msg['role']}: {msg['content']}" for msg in old_history]) summary_prompt = f"Суммируй следующий диалог максимально кратко, сохранив ключевые моменты и темы:\n---\n{old_messages_text}\n---" try: summarized_old_history = self.llm.invoke(summary_prompt) except Exception as e: print(f"Ошибка при суммаризации старой истории: {e}") summarized_old_history = "Ошибка суммаризации старой истории." # Объединяем суммированную старую историю с последними сообщениями full_context = [] if summarized_old_history: full_context.append( f"Предшествующая история (краткое содержание): {summarized_old_history}" ) for msg in recent_messages: full_context.append(f"{msg['role']}: {msg['content']}") return "\n".join(full_context) class ImageGenerationService: """ Сервис для генерации изображений через удаленный эндпоинт (bingart). """ def __init__(self, base_url: str): self.base_url = base_url def generate_images(self, prompt: str) -> List[str]: """ Отправляет запрос на генерацию изображений. """ url = f"{self.base_url}/generate-images" try: # Для простоты используем GET, как указано в ТЗ, хотя POST с body более типичен для prompt response = requests.get(url, params={"prompt": prompt}) response.raise_for_status() data = response.json() if "error" in data: raise Exception( f"Ошибка генерации изображений: {data['error']}") return data.get("image_urls", []) except requests.exceptions.RequestException as e: print(f"Ошибка при вызове сервиса генерации изображений: {e}") raise except Exception as e: print(f"Не удалось получить изображения: {e}") raise