LLM API integration: robust production process

LLM API integration: robust production process

Connecting an LLM API through an HTTP request is a five-minute job. But a month into production, you risk uncontrolled cost growth, timeouts, degraded response quality, and even attacks via prompt injection. For example: one client deployed a chatbot on GPT-4 without caching — their first-month bill exceeded the budget by 8×. Another project faced a system prompt leak due to insufficient input sanitization. Production-ready integration requires thoughtful architecture: retry logic, fallback between providers, input protection, and token control. We take on all these tasks — from provider selection to monitoring costs. We pay special attention to response time: with proper setup, average latency does not exceed 1–2 seconds.

One of the most serious threats is prompt injection (Wikipedia). Without proper protection, an attacker can force the model to ignore system instructions. In our practice, this is a key check before launch.

How to choose an LLM provider?

For most tasks, GPT-4o-mini or Claude Haiku cover 90% of use cases at 5–10× lower cost than flagship models. Semantic caching can reduce API requests by 3–5× compared to conventional caching, directly lowering costs.

Contact us for help choosing the optimal combination — we'll consider your load and budget.

How to protect the backend from prompt injection?

User input should never be inserted directly into the system prompt. Isolate it:

def build_safe_messages(system_prompt: str, user_input: str) -> list[dict]: return [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_input} # never format user_input into system ] def sanitize_user_input(text: str) -> str: # Remove role-switching attempts 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] # limit length 

Our approach to production-ready integration

We build the client with retry and fallback between providers. This code has been battle-tested in dozens of projects.

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}") 

Prompt management and cost control

Prompts are stored in code, versioned via git. We use templates:

from string import Template PROMPTS = { "product_description": Template(""" Write a persuasive product description for an online store. Category: $category Specifications: $specs Target audience: $audience Length: 150–200 words. Tone: $tone Avoid clichés like "innovative", "unique", "best". """), "review_response": Template(""" Write a reply to a customer review on behalf of the store. Rating: $rating/5 Review text: $review Tone: polite, specific, no canned phrases. """) } def get_prompt(name: str, **kwargs) -> str: return PROMPTS[name].substitute(**kwargs) 

We count tokens before sending via tiktoken and log every request. We set daily limits at the provider's dashboard. For cost savings, we use semantic caching:

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 hour return result # Analytics and monitoring 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 

Semantic caching can pay for the integration within the first month of use. Instead of exact match, we compare input embeddings. If a similar query exists, we return the cached response. This reduces costs by 30–70% without quality loss.

What's included in the work

• Architecture and provider selection — analysis of your tasks and recommendation of optimal models.
• Client development with retry and fallback — support for multiple APIs with automatic switching.
• Prompt injection protection — input isolation, sanitization, limits.
• Caching and cost control — semantic cache, limits, logging.
• Documentation and training — integration description, operational manual, team training.
• Post-launch support — 30-day warranty, maintenance.

Process and timelines

1. Analysis — discuss scenarios, select providers, estimate load.
2. Design — client architecture, caching and logging scheme.
3. Implementation — coding, CI/CD setup, backend integration.
4. Testing — load testing, security checks, edge case debugging.
5. Deployment — deploy on your server or cloud, monitoring.

Estimated timelines: basic single API integration — 1–2 days, multi-provider client with fallback — 4–5 days, full infrastructure — 7–8 days. Pricing is determined individually after project evaluation.

Guarantees and experience

Our engineers hold certifications from OpenAI and Anthropic, have over 5 years of backend development experience, and have delivered 100+ successful projects. We guarantee stable integration operation and provide documentation in Russian. For complex cases, we implement custom solutions (e.g., semantic cache based on GPTCache). Get a consultation — we'll evaluate your project and propose the optimal solution.