Інтеграція LLM API в бекенд: надійний production-процес
Підключити LLM API через HTTP-запит — справа п'яти хвилин. Але через місяць у продакшні ви ризикуєте зіткнутися з неконтрольованим зростанням витрат, таймаутами, деградацією якості відповідей і навіть атаками через prompt injection. Наприклад: один наш клієнт запустив чат-бота на GPT-4, не передбачивши кешування — рахунок за перший місяць перевищив плановий у 8 разів. Інший проект зіткнувся з витоком системного промпту через недостатню санітизацію введення користувача. Production-ready інтеграція вимагає продуманої архітектури: retry-логіки, fallback між провайдерами, захисту введення та контролю токенів. Ми беремо на себе всі ці завдання — від вибору провайдера до моніторингу витрат. Особливу увагу приділяємо часу відповіді: при правильному налаштуванні середня затримка не перевищує 1–2 секунд.
Однією з найсерйозніших загроз є prompt injection (Wikipedia). Без належного захисту зловмисник може змусити модель ігнорувати системні інструкції. У нашій практиці це одна з ключових перевірок перед запуском.
Як вибрати провайдера LLM?
| Провайдер | Модель | Сильні сторони | Обмеження |
|---|---|---|---|
| OpenAI | GPT-4o, GPT-4o-mini | Зріле API, найкраща екосистема | Дорожче аналогів |
| Anthropic | Claude 3.5 Sonnet, Claude Haiku | Довгий контекст, точність | Немає embedding API |
| Gemini 1.5 Pro/Flash | Ціна, мультимодальність | Менш стабільне API | |
| Mistral | Mistral Large, Mixtral | Європейський провайдер, GDPR | Менше інструментів |
| Groq | Llama 3, Mixtral | Швидкість (300+ token/s) | Обмежений вибір моделей |
Для більшості завдань GPT-4o-mini або Claude Haiku покривають 90% випадків при в 5–10 разів меншій вартості флагманських моделей. Семантичне кешування скорочує кількість запитів до API в 3–5 разів порівняно зі звичайним кешуванням, що прямо знижує витрати.
| Сценарій | Рекомендована модель | Альтернатива |
|---|---|---|
| Чат-бот підтримки | GPT-4o-mini | Claude Haiku |
| Аналіз документів | Claude 3.5 Sonnet | Gemini 1.5 Pro |
| Генерація контенту | GPT-4o | Mistral Large |
Для підбору оптимальної комбінації зв'яжіться з нами — ми врахуємо ваше навантаження та бюджет.
Як захистити бекенд від 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] # обмежуємо довжину
Наш підхід до production-ready інтеграції
Ми будуємо клієнт із 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)
Рахуємо токени до відправки через tiktoken та логуємо кожен запит. Ставимо добові ліміти на рівні dashboard провайдера. Для економії використовуємо семантичне кешування:
import hashlib
import json
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
# Аналітика та моніторинг
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
Семантичне кешування дозволяє окупити інтеграцію протягом першого місяця використання. Замість точного збігу запитів ми порівнюємо embedding вхідних даних. Якщо схожий запит вже був, повертаємо кешовану відповідь. Це знижує витрати на 30–70% без втрати якості.
Що входить в роботу
- Архітектура та вибір провайдера — аналіз ваших завдань і рекомендація оптимальних моделей.
- Розробка клієнта з retry та fallback — підтримка декількох API з автоматичним перемиканням.
- Захист від prompt injection — ізоляція введення, санітизація, обмеження.
- Кешування та контроль витрат — семантичний кеш, ліміти, логування.
- Документація та навчання — опис інтеграції, інструкція з експлуатації, навчання команди.
- Підтримка після запуску — гарантія 30 днів, супровід.
Процес роботи та терміни
- Аналітика — обговорюємо сценарії, вибираємо провайдерів, оцінюємо навантаження.
- Проектування — архітектура клієнта, схема кешування, логування.
- Реалізація — написання коду, налаштування CI/CD, інтеграція з вашим бекендом.
- Тестування — навантажувальне тестування, перевірка безпеки, налагодження граничних випадків.
- Деплой — розгортання на вашому сервері або хмарі, моніторинг.
Орієнтовні терміни: базова інтеграція одного API — 1–2 дні, мультипровайдерний клієнт із fallback — 4–5 днів, повна інфраструктура — 7–8 днів. Вартість розраховується індивідуально після оцінки проекту.
Гарантії та досвід
Наші інженери мають сертифікати з OpenAI та Anthropic, більше 5 років досвіду в розробці бекендів та 100+ успішних проектів. Ми гарантуємо стабільну роботу інтеграції та надаємо документацію українською. Для складних випадків впроваджуємо кастомні рішення (наприклад, семантичний кеш на базі GPTCache). Отримайте консультацію — оцінимо ваш проект та запропонуємо оптимальне рішення.