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.

AI Integration: Chatbots, RAG, Semantic Search, Recommendations

In 8 out of 10 projects, an "AI chatbot" turns out to be an expensive wrapper over GPT-4o with a system prompt. Without access to real company data. The user asks "how much does the Premium plan cost?" — the bot hallucinates a price out of thin air. Asks "when will my order arrive?" — gets a polite "contact support." This is not integration — it's imitation. We have implemented RAG solutions in 30+ projects over 5 years: from e-commerce stores to medical portals. We guarantee: useful AI assistance begins where the model reads your documents, not generic answers.

How do we build RAG systems?

Retrieval-Augmented Generation — standard architecture: query → find relevant fragments in a vector DB → insert found context → model response. But the devil is in the implementation details. Let's break down key components that determine quality.

Chunking. Cutting a document into 500-token pieces without regard for structure is a guarantee of losing meaning. If the cut lands in the middle of a paragraph, context breaks. Solution — recursive RecursiveCharacterTextSplitter with 10–15% overlap for documentation. For contracts and instructions, we use a semantic splitter: extract headings, lists, code blocks — each section becomes an independent chunk. Difference in search quality: on a medical project, precision increased from 0.55 to 0.84 just by proper chunking.

Embedding model. For Russian-language texts, intfloat/multilingual-e5-large gives a noticeable accuracy boost over outdated text-embedding-ada-002. In our measurements, NDCG@10 on a test set of 10,000 query-document pairs is 12% higher. OpenAI text-embedding-3-large is good for English content, but for Russian we recommend BAAI/bge-m3 or the mentioned e5-large.

Vector DB. If you already have PostgreSQL — pgvector saves resources. Install extension CREATE EXTENSION vector, add column vector(1024), create HNSW index. On a project with 80,000 support articles, p95 search time was 12 ms. That's enough. For catalogs with millions of items — Qdrant or Weaviate: native hybrid search and sharding out of the box.

What does hybrid search give?

Vector-only search is blind to exact matches: SKUs like "ABC-123", proper names, abbreviations are lost. Full-text-only search doesn't catch synonyms and paraphrasing. Combining via RRF (Reciprocal Rank Fusion) gives the best of both worlds: BM25 + vector search, results merged. In practice, recall@20 increases from 0.65 to 0.92 — the difference is noticeable to the user.

Reranking — final filter: top-20 candidates from hybrid search are run through a cross-encoder cross-encoder/ms-marco-MiniLM-L-6-v2. It adds 50–100 ms to response time, but relevance improves by another 5–10%. Without reranking, the chatbot may show irrelevant documents.

How to implement semantic search on a site?

A search for "comfortable leather armchairs" should find products described as "soft chairs made of natural leather" — ordinary LIKE search cannot do this. Our architecture: when adding a product/post, automatically generate an embedding via multilingual-e5-large, store it in pgvector. On query, embed it with the same model, search nearest neighbors via cosine distance with HNSW index. For a catalog of 100,000 items, index builds in 3 minutes, memory ~400 MB (1536-dimensional vectors). Average search time: 20 ms.

What about recommendation systems?

Collaborative filtering ("users like you bought X") requires history — at least 2–3 months of data with 1000+ active users. For startups or small projects, we use content-based: embedding of current product → search nearest neighbors by cosine similarity. When enough statistics accumulate (usually 15–20 interactions per user), we switch to a hybrid LightFM model. It combines behavior and product features. In our e-commerce project with 50,000 SKUs, the hybrid model increased conversion in the recommendation block by 18% (A/B test lasted 2 weeks).

How does streaming work?

Users shouldn't wait for the entire text to be generated — it kills UX. Server-Sent Events (SSE) is the protocol for token streaming. OpenAI SDK supports stream: true, returning an AsyncIterator. On frontend — Vercel AI SDK (useChat) or custom EventSource. Typical mistake: using WebSocket for unidirectional streaming — SSE is simpler (less code, built-in reconnect). Stack: Node.js + SSE + React.

How to orchestrate agents?

A simple chatbot answers. An agent performs actions: creates a Jira ticket, checks order status in CRM, books a calendar slot. For orchestration, we use LangGraph: state graph where each node is a model or tool call. Vercel AI SDK useChat + tools for Next.js allows adding integration in 10 lines of code. Main challenge — reliability: the model sometimes calls the wrong tool or passes malformed parameters. Protection — Zod schemas for each tool and structured outputs to guarantee JSON.

What does the work include?

Additionally: we hand over vectorizer source code, monitoring dashboards (Langfuse), admin panel access for knowledge base updates. Post-production support — 1 month free.

What are the timelines?

Pricing is calculated individually after project discovery. We'll evaluate your project in 1 day. Contact us — we'll show how to turn AI from a toy into a profit-driving tool.