Интеграция LLM API (OpenAI/Anthropic/Gemini) в бэкенд сайта

Интеграция LLM API (OpenAI/Anthropic/Gemini) в бэкенд сайта

Подключить LLM API в три строки кода — не проблема. Проблема начинается через месяц работы в продакшне: неожиданные расходы, таймауты, деградация качества, prompt injection от пользователей. Эта страница — про production-ready интеграцию.

Выбор провайдера

Для большинства задач GPT-4o-mini или Claude Haiku покрывают 90% случаев при в 5–10 раз меньшей стоимости флагманских моделей.

Структура клиента с retry и fallback

import asyncio
from openai import AsyncOpenAI, APIError, RateLimitError, APITimeoutError
from anthropic import AsyncAnthropic
import time

class LLMClient:
    def __init__(self):
        self.openai = AsyncOpenAI(api_key=OPENAI_API_KEY, timeout=30.0)
        self.anthropic = AsyncAnthropic(api_key=ANTHROPIC_API_KEY, timeout=30.0)

    async def complete(
        self,
        messages: list[dict],
        model: str = "gpt-4o-mini",
        temperature: float = 0.7,
        max_tokens: int = 1000,
        retries: int = 3
    ) -> str:
        last_error = None

        for attempt in range(retries):
            try:
                if model.startswith("gpt") or model.startswith("o1"):
                    response = await self.openai.chat.completions.create(
                        model=model,
                        messages=messages,
                        temperature=temperature,
                        max_tokens=max_tokens
                    )
                    return response.choices[0].message.content

                elif model.startswith("claude"):
                    system = next((m["content"] for m in messages if m["role"] == "system"), None)
                    user_messages = [m for m in messages if m["role"] != "system"]
                    response = await self.anthropic.messages.create(
                        model=model,
                        system=system,
                        messages=user_messages,
                        max_tokens=max_tokens
                    )
                    return response.content[0].text

            except RateLimitError:
                wait = 2 ** attempt
                await asyncio.sleep(wait)
                last_error = "rate_limit"

            except APITimeoutError:
                last_error = "timeout"
                if attempt < retries - 1:
                    await asyncio.sleep(1)

            except APIError as e:
                if e.status_code >= 500:
                    await asyncio.sleep(2 ** attempt)
                    last_error = f"server_error_{e.status_code}"
                else:
                    raise

        raise RuntimeError(f"LLM call failed after {retries} attempts: {last_error}")

Управление промптами

Храните промпты в коде, не в базе данных — версионирование в git. Используйте шаблоны:

from string import Template

PROMPTS = {
    "product_description": Template("""
Напиши продающее описание товара для интернет-магазина.
Категория: $category
Характеристики: $specs
Целевая аудитория: $audience
Объём: 150–200 слов.
Тон: $tone
Не используй клише типа "инновационный", "уникальный", "лучший".
"""),

    "review_response": Template("""
Напиши ответ на отзыв покупателя от имени магазина.
Оценка: $rating/5
Текст отзыва: $review
Тон: вежливый, конкретный, без шаблонных фраз.
""")
}

def get_prompt(name: str, **kwargs) -> str:
    return PROMPTS[name].substitute(**kwargs)

Защита от prompt injection

Пользовательский ввод нельзя вставлять напрямую в системный промпт. Изоляция:

def build_safe_messages(system_prompt: str, user_input: str) -> list[dict]:
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_input}  # никогда не форматировать user_input в system
    ]

def sanitize_user_input(text: str) -> str:
    # Удаляем попытки сменить роль
    dangerous_patterns = [
        r"ignore previous instructions",
        r"you are now",
        r"forget everything",
        r"system:",
        r"<\|im_start\|>"
    ]
    for pattern in dangerous_patterns:
        text = re.sub(pattern, "[filtered]", text, flags=re.IGNORECASE)
    return text[:4000]  # ограничиваем длину

Контроль расходов

Считайте токены до отправки через tiktoken:

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    enc = tiktoken.encoding_for_model(model)
    return len(enc.encode(text))

def estimate_cost(input_tokens: int, output_tokens: int, model: str) -> float:
    pricing = {
        "gpt-4o-mini": {"input": 0.00015, "output": 0.00060},
        "gpt-4o": {"input": 0.0025, "output": 0.010},
        "claude-3-5-haiku-20241022": {"input": 0.0008, "output": 0.004},
    }
    p = pricing.get(model, {"input": 0.001, "output": 0.002})
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1000

Логируйте расходы по каждому типу запроса в базу. Ставьте суточные лимиты на уровне OpenAI/Anthropic dashboard.

Кэширование и дедупликация

Семантическое кэширование — для похожих вопросов возвращать кэшированный ответ:

import hashlib
from redis import Redis

cache = Redis()

def cached_llm_call(messages: list[dict], **kwargs) -> str:
    cache_key = "llm:" + hashlib.sha256(
        json.dumps(messages, sort_keys=True).encode()
    ).hexdigest()

    cached = cache.get(cache_key)
    if cached:
        return cached.decode()

    result = await llm_client.complete(messages, **kwargs)
    cache.setex(cache_key, 3600, result)  # 1 час
    return result

Для semantic caching (похожие, но не идентичные запросы) используйте GPTCache или встройте в собственную RAG-систему.

Мониторинг и логирование

Каждый LLM-запрос должен логироваться:

async def tracked_llm_call(messages, user_id: str, feature: str, **kwargs) -> str:
    start = time.time()
    try:
        result = await llm_client.complete(messages, **kwargs)
        latency = time.time() - start

        await db.llm_logs.insert({
            "user_id": user_id,
            "feature": feature,
            "model": kwargs.get("model"),
            "input_tokens": count_tokens(str(messages)),
            "output_tokens": count_tokens(result),
            "latency_ms": int(latency * 1000),
            "success": True,
            "timestamp": datetime.utcnow()
        })
        return result

    except Exception as e:
        await db.llm_logs.insert({"feature": feature, "error": str(e), "success": False})
        raise

Сроки

Базовая интеграция одного API с retry — 1–2 дня. Мультипровайдерный клиент с fallback + логирование + кэш — 4–5 дней. Полная production-ready инфраструктура с мониторингом расходов — 7–8 дней.