llm.py - main - Soldier327/API_INTERFACE

llm.py
195 строк9 KB
Soldier-327
Initial commit: add shopping list generator with Streamlit UI
27 май 2026, 18:21
27 май 2026, 18:218fa6c09
"""Модуль обработчика API для генерации списка покупок."""

import requests
import json
from typing import Dict, Any, Optional

# Конфигурация API (можно переопределить через переменные окружения)
DEFAULT_API_URL = "https://api.openai.com/v1/chat/completions"


class ShoppingListGenerator:
    """Класс для генерации списка покупок через API модели."""

    def __init__(self, api_key: str, api_url: str = DEFAULT_API_URL):
        """
        Инициализация генератора.

        Args:
            api_key: API ключ для аутентификации
            api_url: URL эндпоинта API
        """
        self.api_key = api_key
        self.api_url = api_url

    def _build_prompt(self, dish: str, people: str, format_type: str) -> Dict[str, Any]:
        """
        Формирование промпта для модели.

        Args:
            dish: Название блюда
            people: Количество человек
            format_type: Формат ответа

        Returns:
            Словарь с промптом для API
        """
        if format_type == "Только список":
            system_prompt = """Ты — помощник по кулинарии. 
Сгенерируй ТОЛЬКО список продуктов для приготовления блюда.
Каждый продукт с новой строки, используй маркер "-" в начале строки.
Не добавляй пояснений, шагов или лишнего текста.
Учитывай количество персон и корректируй количество продуктов.

Пример вывода:
- Курица 500г
- Помидоры 3 шт
- Чеснок 2 зубчика
- Соль по вкусу"""

            user_prompt = f"""Блюдо: {dish}
Количество персон: {people}
Сгенерируй список продуктов."""

        else:  # "Список + шаги"
            system_prompt = """Ты — помощник по кулинарии.
Сначала выведи список продуктов (каждый продукт с новой строки, используй маркер "-"),
затем добавь раздел "🍳 Шаги приготовления:" и 5-8 коротких шагов.
Шаги должны быть краткими, понятными, начинаться с глаголов.
Учитывай количество персон.

Пример вывода:
- Курица 500г
- Помидоры 3 шт
- Чеснок 2 зубчика

🍳 Шаги приготовления:
1. Нарежьте курицу кубиками.
2. Обжарьте курицу до золотистой корочки.
3. Добавьте нарезанные помидоры.
4. Добавьте измельчённый чеснок.
5. Тушите 10 минут на медленном огне.
6. Посолите по вкусу."""

            user_prompt = f"""Блюдо: {dish}
Количество персон: {people}
Сгенерируй список продуктов и краткие шаги приготовления (5-8 пунктов)."""

        return {
            "model": "gpt-3.5-turbo",
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            "temperature": 0.7,
            "max_tokens": 1000
        }

    def generate(self, dish: str, people: str, format_type: str) -> str:
        """
        Генерация списка покупок через API.

        Args:
            dish: Название блюда
            people: Количество человек
            format_type: Формат ответа

        Returns:
            Строка с результатом или сообщение об ошибке
        """
        # Валидация входных данных на уровне модуля
        if not dish or not dish.strip():
            return "Пустой запрос"

        # Валидация формата API ключа
        if not self.api_key or not self.api_key.startswith("sk-"):
            return "🔒 Неверный формат API ключа. Должен начинаться с 'sk-'"

        # Формируем промпт
        payload = self._build_prompt(dish, people, format_type)

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        try:
            # Вызов API с таймаутом 30 секунд
            response = requests.post(
                self.api_url,
                json=payload,
                headers=headers,
                timeout=30
            )

            # Обработка HTTP ошибок по кодам
            if response.status_code == 401:
                return "🔒 Не настроен доступ к сервису."
            elif response.status_code == 429:
                return "📊 Превышен лимит запросов. Попробуйте позже."
            elif 400 <= response.status_code < 500:
                return "⚠️ Ошибка в запросе. Проверьте настройки API."
            elif 500 <= response.status_code < 600:
                return "🔧 Сервис временно недоступен. Попробуйте позже."

            # Проверка успешности ответа
            response.raise_for_status()

            # Парсинг JSON ответа
            data = response.json()

            # Проверка структуры ответа
            if not data.get("choices") or len(data["choices"]) == 0:
                return "❌ Пустой ответ модели. Попробуйте переформулировать вопрос."

            result = data["choices"][0].get("message", {}).get("content", "")

            if not result or len(result.strip()) == 0:
                return "❌ Пустой ответ модели. Попробуйте переформулировать вопрос."

            return result.strip()

        except requests.exceptions.Timeout:
            return "⏰ Не удалось получить ответ. Попробуйте ещё раз."
        except requests.exceptions.ConnectionError:
            return "🌐 Не удалось подключиться к сервису. Проверьте интернет."
        except json.JSONDecodeError:
            return "📄 Неожиданный формат ответа от сервера."
        except Exception:
            # Любые другие ошибки без traceback
            return "❌ Ошибка при вызове API. Попробуйте позже."

    def generate_with_retry(self, dish: str, people: str, format_type: str, max_retries: int = 1) -> str:
        """
        Генерация с повторами при временных ошибках.

        Args:
            dish: Название блюда
            people: Количество человек
            format_type: Формат ответа
            max_retries: Максимальное количество повторов

        Returns:
            Строка с результатом
        """
        result = self.generate(dish, people, format_type)

        # Список ошибок, при которых стоит повторить
        retryable_errors = [
            "⏰ Не удалось получить ответ. Попробуйте ещё раз.",
            "🌐 Не удалось подключиться к сервису. Проверьте интернет.",
            "🔧 Сервис временно недоступен. Попробуйте позже.",
            "📊 Превышен лимит запросов. Попробуйте позже."
        ]

        if max_retries > 0 and result in retryable_errors:
            result = self.generate(dish, people, format_type)

        return result


# Функция-обёртка для простого вызова (для совместимости)
def generate_shopping_list(dish: str, people: str, format_type: str, api_key: str) -> str:
    """Упрощённая функция вызова API."""
    generator = ShoppingListGenerator(api_key)
    return generator.generate_with_retry(dish, people, format_type)