llm-agent-backend/app/services.py

"""
Этот файл содержит классы сервисов для выполнения различных задач:
- Получение субтитров,
- Анализ изображений,
- Суммаризация текста,
- Генерация изображений.
"""

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