Інтеграція LLM API в бекенд: надійний production-процес

Інтеграція LLM API в бекенд: надійний production-процес

Підключити LLM API через HTTP-запит — справа п'яти хвилин. Але через місяць у продакшні ви ризикуєте зіткнутися з неконтрольованим зростанням витрат, таймаутами, деградацією якості відповідей і навіть атаками через prompt injection. Наприклад: один наш клієнт запустив чат-бота на GPT-4, не передбачивши кешування — рахунок за перший місяць перевищив плановий у 8 разів. Інший проект зіткнувся з витоком системного промпту через недостатню санітизацію введення користувача. Production-ready інтеграція вимагає продуманої архітектури: retry-логіки, fallback між провайдерами, захисту введення та контролю токенів. Ми беремо на себе всі ці завдання — від вибору провайдера до моніторингу витрат. Особливу увагу приділяємо часу відповіді: при правильному налаштуванні середня затримка не перевищує 1–2 секунд.

Однією з найсерйозніших загроз є prompt injection (Wikipedia). Без належного захисту зловмисник може змусити модель ігнорувати системні інструкції. У нашій практиці це одна з ключових перевірок перед запуском.

Як вибрати провайдера LLM?

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

Для підбору оптимальної комбінації зв'яжіться з нами — ми врахуємо ваше навантаження та бюджет.

Як захистити бекенд від 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 днів, супровід.

Процес роботи та терміни

1. Аналітика — обговорюємо сценарії, вибираємо провайдерів, оцінюємо навантаження.
2. Проектування — архітектура клієнта, схема кешування, логування.
3. Реалізація — написання коду, налаштування CI/CD, інтеграція з вашим бекендом.
4. Тестування — навантажувальне тестування, перевірка безпеки, налагодження граничних випадків.
5. Деплой — розгортання на вашому сервері або хмарі, моніторинг.

Орієнтовні терміни: базова інтеграція одного API — 1–2 дні, мультипровайдерний клієнт із fallback — 4–5 днів, повна інфраструктура — 7–8 днів. Вартість розраховується індивідуально після оцінки проекту.

Гарантії та досвід

Наші інженери мають сертифікати з OpenAI та Anthropic, більше 5 років досвіду в розробці бекендів та 100+ успішних проектів. Ми гарантуємо стабільну роботу інтеграції та надаємо документацію українською. Для складних випадків впроваджуємо кастомні рішення (наприклад, семантичний кеш на базі GPTCache). Отримайте консультацію — оцінимо ваш проект та запропонуємо оптимальне рішення.