Integrating Large Language Models into production applications requires careful consideration of API design, error handling, cost management, and performance. The naive approach—wrap the provider SDK in a function and call it from anywhere—fails the moment a real user base arrives, because LLM APIs behave unlike most services you have integrated before. They are slow, non-deterministic, and priced per token, and they fail in ways that ordinary REST clients do not prepare you for: rate limits, transient server errors, and sudden timeouts on long generations. The patterns in this guide are the difference between a feature that occasionally breaks and one that degrades gracefully under load.
This guide covers the practical patterns that turn an LLM API from a demo into a reliable feature. We start with the simplest possible integration and then layer on conversation state, streaming for responsiveness, retry and timeout handling, cost controls, caching, and the structured-output techniques such as function calling that let models drive application logic. Each pattern is shown with working Python code against the OpenAI SDK, but the underlying principles—context management, backoff, budgeting, and graceful degradation—apply to any provider.
Basic LLM Integration
Before worrying about streaming, caching, or retries, you need the simplest thing that works: a single API call that turns a prompt into a response. Modern LLM providers expose a chat-completions endpoint where you send a list of messages and receive a model-generated completion. Even this basic call involves design decisions that will shape everything downstream. You must choose a model that balances quality against cost and latency, set a temperature that matches the determinism your feature needs, and decide how many tokens you are willing to spend per request. These parameters are not boilerplate; they are the knobs that determine whether your feature is fast, cheap, and consistent. Also decide early where the integration layer lives: a small module that wraps the SDK keeps the rest of your codebase ignorant of LLM specifics and makes the calls easy to mock in tests. Keep the API key out of source code from day one—environment variables or a secret manager are the minimum, and the key should never be logged or exposed in client-side code.
Simple API Calls
The example below is the canonical minimal integration.
It initializes a client with an API key, builds a message list containing an optional system instruction and the user’s prompt, and returns the assistant’s text from the first completion choice.
Note how the model and temperature are parameters rather than hardcoded values: that makes it trivial to A/B test models or tighten determinism for tasks like data extraction.
The system message deserves special attention—it is how you set the assistant’s behavior without repeating instructions in every user message.
A well-written system prompt often matters more than the user prompt itself, so it is worth treating it as a first-class artifact.
Notice also that a hard max_tokens cap prevents a runaway response from draining your quota, which is a cheap safeguard worth adding from day one.
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
def generate_response(prompt, model="gpt-4", temperature=0.7):
"""Generate a response from an LLM."""
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": prompt}
],
temperature=temperature,
max_tokens=500
)
return response.choices[0].message.content
# Usage
response = generate_response("Explain quantum computing in simple terms")
print(response)
This single-shot pattern is fine for one-off queries, but most real features—chatbots, coding assistants, support agents—require multiple turns of context. If you send only the latest user message, the model has no memory of earlier exchanges, so follow-up questions lose their meaning. The standard solution is to maintain a message list that grows with each turn and send the entire history to the API on every request. The trade-off is cost and latency: because the full history is reprocessed each time, long conversations become expensive and slow. This is the fundamental tension in LLM integration—context is expensive, so you must manage it deliberately.
Conversation Management
The class below encapsulates that bookkeeping. It appends each user and assistant message to an in-memory list, injects a system prompt at the front of every request, and stores each assistant reply so the context stays coherent. One detail to notice is that only the assistant’s final text is appended, not the whole response object, which keeps the message structure clean and compatible with the API. This naive version has no size limit yet—we will add context-window management in the pitfalls section, since unbounded history is a common production bug. For interactive products, consider persisting the message list per user rather than per process, so sessions survive restarts and horizontal scaling keeps the context intact. Also be aware that a shared in-memory list is not thread-safe; if one instance is used across concurrent requests, create an instance per user session instead.
class ConversationManager:
"""Manage multi-turn conversations with LLMs."""
def __init__(self, system_prompt="You are a helpful assistant."):
self.client = OpenAI()
self.system_prompt = system_prompt
self.messages = []
def add_message(self, role, content):
"""Add message to conversation history."""
self.messages.append({"role": role, "content": content})
def get_response(self, user_input, model="gpt-4"):
"""Get response while maintaining conversation context."""
self.add_message("user", user_input)
response = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": self.system_prompt},
*self.messages
],
temperature=0.7
)
assistant_message = response.choices[0].message.content
self.add_message("assistant", assistant_message)
return assistant_message
def clear_history(self):
"""Clear conversation history."""
self.messages = []
# Usage
manager = ConversationManager()
response1 = manager.get_response("What is Python?")
response2 = manager.get_response("How is it different from Java?")
Streaming Responses
Users judge an AI feature by how quickly it appears to respond. Waiting for a full response to generate can take tens of seconds, and a silent loading spinner for that long feels broken even when the request is proceeding normally. Streaming solves this by delivering tokens as they are generated, so the user sees text appear word by word, just as a human typist would produce it. Beyond perceived performance, streaming also lets you cancel a generation early, display partial results, or even stop a runaway response mid-sentence. From an architecture perspective, streaming changes your API shape: instead of a single response body, you stream bytes over time, which has implications for timeouts, buffering, and client libraries. The streaming pattern also affects your infrastructure: proxies and load balancers must not buffer the response, or the entire benefit of streaming is lost. The client, in turn, must be tolerant of partial reads, because a connection can drop mid-stream and the user should still see the tokens that were already delivered. Streaming is essential for real-time user feedback and reducing perceived latency.
Basic Streaming
The streaming implementation is deceptively simple: set stream=True, and the SDK returns an iterator of chunks instead of a single response.
Each chunk carries a delta containing the next piece of text, which you print immediately and accumulate into the full response.
The flush=True flag forces the terminal to render each token as it arrives, which is what creates the typewriter effect.
The same principle maps directly to the web: with Server-Sent Events or a streaming HTTP client, you forward each chunk to the browser as it arrives instead of buffering the whole completion.
One practical detail is that streaming and non-streaming requests should be split into separate functions, because the error handling and return shapes are genuinely different.
def stream_response(prompt):
"""Stream response token by token."""
client = OpenAI()
stream = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
stream=True
)
full_response = ""
for chunk in stream:
if chunk.choices[0].delta.content:
token = chunk.choices[0].delta.content
print(token, end="", flush=True)
full_response += token
print() # Newline
return full_response
# Usage
response = stream_response("Write a short poem about Python")
Printing to the terminal works for a demo, but production code needs to route tokens into whatever consumer is appropriate—a websocket, a UI component, or a log stream. The callback pattern decouples the streaming logic from the display logic. The streaming function accepts a callable that is invoked for every token, so the same generator can drive a terminal, a web page, or a telemetry feed without changing the core code. This decoupling also makes the streaming path easy to test and reuse across features.
Streaming with Callbacks
The class below generalizes basic streaming by accepting an on_token callback, defaulting to print, and accumulating the full response alongside it.
This separation of concerns makes the streaming path testable: you can inject a callback that collects tokens into a list and assert on the sequence, rather than capturing stdout.
In a web framework like FastAPI, you would pass an async callback that yields each token into an SSE response, giving you real-time UI updates with no extra infrastructure.
The callback is also the natural place to accumulate usage metadata, so you can record token counts for cost accounting as the stream progresses.
from typing import Callable
class StreamingCallback:
"""Handle streaming responses with callbacks."""
def __init__(self, on_token: Callable[[str], None] = None):
self.on_token = on_token or print
self.full_response = ""
def stream(self, prompt, model="gpt-4"):
"""Stream response with callback."""
client = OpenAI()
stream = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
token = chunk.choices[0].delta.content
self.on_token(token)
self.full_response += token
return self.full_response
# Usage with custom callback
def my_callback(token):
print(f"[{token}]", end="", flush=True)
callback = StreamingCallback(on_token=my_callback)
response = callback.stream("Explain machine learning")
Error Handling and Resilience
LLM APIs are reliable most of the time, but “most of the time” is not good enough for production. Provider APIs experience rate limits, transient 5xx errors, and network timeouts that would crash a naive integration at the worst possible moment—during a user-facing request. Unlike typical databases, where a single retry often succeeds, LLM failures frequently happen under load, which is exactly when you can least afford to hammer the provider again. A robust integration treats every API call as potentially failing and designs the retry behavior explicitly. Before you write any retry code, look at what your provider SDK already offers, because vendor retry policies are typically tuned to the provider’s own error semantics. Whichever path you choose, log every retry so that a sustained outage shows up in your metrics as a spike in retries rather than a silent wave of timeouts.
Robust Error Handling
The function below implements retry with exponential backoff: each failed attempt waits backoff_factor ** attempt seconds, so the first retry waits 2 seconds, the second 4, and so on.
Rate limits and transient server errors are retried, while other errors are raised immediately so a genuine bug is not masked by silent retries.
Note that the retries have a hard cap—after max_retries attempts the function raises, because retrying forever only amplifies an outage.
In production, the sleep should be replaced by jittered backoff and the policy tuned per error type, but the shape of this loop is the pattern you will build on.
Most provider SDKs now ship their own retry configuration, which is often the better choice because the vendor knows which errors are actually safe to retry.
from openai import OpenAI, RateLimitError, APIError
import time
from typing import Optional
def call_llm_with_retry(
prompt: str,
max_retries: int = 3,
backoff_factor: float = 2.0
) -> Optional[str]:
"""Call LLM with exponential backoff retry logic."""
client = OpenAI()
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
timeout=30
)
return response.choices[0].message.content
except RateLimitError as e:
wait_time = backoff_factor ** attempt
print(f"Rate limited. Waiting {wait_time}s before retry...")
time.sleep(wait_time)
except APIError as e:
if e.status_code == 500:
wait_time = backoff_factor ** attempt
print(f"Server error. Waiting {wait_time}s before retry...")
time.sleep(wait_time)
else:
raise
except Exception as e:
print(f"Unexpected error: {e}")
raise
raise Exception(f"Failed after {max_retries} retries")
# Usage
try:
response = call_llm_with_retry("What is AI?")
print(response)
except Exception as e:
print(f"Error: {e}")
Retries handle failures that are transient, but a request can also hang indefinitely—a slow model, a stalled network connection, or a provider that accepts the request but never responds. A call with no timeout pins an async worker and, under load, silently consumes all your concurrency slots. Timeouts convert an unbounded wait into a bounded one so that your system can fail fast and move on. Choosing a good timeout requires knowing your typical latency: set it high enough to accommodate legitimate slow generations, but low enough that a wedged request is released quickly.
Timeout Handling
The example uses asyncio.wait_for to wrap the API call with a deadline.
If the call exceeds timeout_seconds, the coroutine is cancelled and the function returns None, which the caller can treat as a graceful failure.
The async client matters here: a synchronous SDK blocks a thread while waiting, whereas the async client releases the event loop, so timeouts and concurrent requests compose cleanly.
Returning None rather than raising makes the timeout a first-class control-flow outcome that your application can handle with a friendly “please try again” message.
In a web request, also add an HTTP-level timeout as a last line of defense, so a hung upstream connection cannot pin a request forever even if the SDK defaults are overridden.
import asyncio
from openai import AsyncOpenAI
async def call_with_timeout(prompt, timeout_seconds=30):
"""Call LLM with timeout."""
client = AsyncOpenAI()
try:
response = await asyncio.wait_for(
client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
),
timeout=timeout_seconds
)
return response.choices[0].message.content
except asyncio.TimeoutError:
print(f"Request timed out after {timeout_seconds}s")
return None
# Usage
response = asyncio.run(call_with_timeout("Explain quantum computing"))
Cost Optimization
LLM providers bill by token, which makes cost a first-class engineering concern rather than an afterthought. A single chat request can cost fractions of a cent, but at scale—a few thousand requests per minute across a team—those fractions become a serious line item, and long context histories multiply the cost of every interaction. You cannot control what you cannot measure, so the first step is instrumentation: counting the tokens in every request and response and tracking what each call costs. From there, the two big levers are caching repeatable prompts and reducing the size of what you send. Before optimizing, spend a day measuring: most teams are surprised by where their tokens actually go, and the fix is usually a single hot path rather than a global change. A useful habit is to record a per-feature cost estimate at design time, then compare it against reality after launch so that drift becomes visible.
Token Counting
The example below uses tiktoken, the tokenizer library that mirrors the provider’s own counting, to measure the input and output of a call and apply per-model pricing.
Notice that input and output tokens are priced differently—output is typically several times more expensive—so a naive “total tokens” metric hides the real cost structure.
The pricing table is embedded for illustration but should be kept in configuration, because model prices change.
Instrumenting calls this way lets you set budgets, alert on cost spikes, and justify caching investment with real numbers.
Tracking cost per user is also useful: it lets you detect abusive usage patterns and enforce per-user budgets before a single user’s session becomes a surprise line item.
import tiktoken
def count_tokens(text, model="gpt-4"):
"""Count tokens in text."""
encoding = tiktoken.encoding_for_model(model)
tokens = encoding.encode(text)
return len(tokens)
def estimate_cost(prompt, response, model="gpt-4"):
"""Estimate API cost for a request."""
# Pricing as of 2025 (update as needed)
pricing = {
"gpt-4": {"input": 0.03, "output": 0.06},
"gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
"gpt-4-turbo": {"input": 0.01, "output": 0.03}
}
input_tokens = count_tokens(prompt, model)
output_tokens = count_tokens(response, model)
rates = pricing.get(model, pricing["gpt-4"])
input_cost = (input_tokens / 1000) * rates["input"]
output_cost = (output_tokens / 1000) * rates["output"]
return {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"input_cost": input_cost,
"output_cost": output_cost,
"total_cost": input_cost + output_cost
}
# Usage
prompt = "Explain machine learning"
response = "Machine learning is..."
cost = estimate_cost(prompt, response)
print(f"Total cost: ${cost['total_cost']:.6f}")
Caching Responses
Many real workloads are highly repetitive: the same documentation lookup, product description, or template filled with the same inputs is requested over and over. Each duplicate call pays full price and adds latency to a response you have already computed. A cache keyed on the prompt and model can eliminate most of that spend, particularly for deterministic tasks with low temperature settings. The trade-offs are that cached responses go stale, and a cache key must capture everything that affects the output—model, temperature, system prompt, and message history—or you risk serving the wrong answer.
The class below implements a simple on-disk cache keyed by an MD5 hash of the model and prompt. Before calling the API, it checks for a cached result and returns it immediately; on a miss, it calls the provider and stores the response. This is a deliberately simple design that is easy to reason about and swap out for Redis or a database in production. The important production considerations are a TTL to expire stale entries, keying on the full request context, and a strategy for invalidating entries when your prompts change. For high-volume traffic, a distributed cache with a shared key namespace is essential so that every instance of your application benefits from the same cache.
import hashlib
import json
from pathlib import Path
class LLMCache:
"""Cache LLM responses to reduce API calls."""
def __init__(self, cache_dir=".llm_cache"):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def _get_cache_key(self, prompt, model):
"""Generate cache key from prompt and model."""
key = f"{model}:{prompt}"
return hashlib.md5(key.encode()).hexdigest()
def get(self, prompt, model):
"""Get cached response if available."""
cache_key = self._get_cache_key(prompt, model)
cache_file = self.cache_dir / f"{cache_key}.json"
if cache_file.exists():
with open(cache_file) as f:
return json.load(f)
return None
def set(self, prompt, model, response):
"""Cache response."""
cache_key = self._get_cache_key(prompt, model)
cache_file = self.cache_dir / f"{cache_key}.json"
with open(cache_file, 'w') as f:
json.dump(response, f)
def call_with_cache(self, prompt, model="gpt-4"):
"""Call LLM with caching."""
cached = self.get(prompt, model)
if cached:
print("Using cached response")
return cached
client = OpenAI()
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
result = response.choices[0].message.content
self.set(prompt, model, result)
return result
# Usage
cache = LLMCache()
response = cache.call_with_cache("What is Python?")
Advanced Integration Patterns
So far the model has only produced free-form text, which is difficult to consume reliably in application logic. Advanced patterns change the relationship between the model and your code. Function calling lets the model request that your application execute a specific function with typed arguments, turning an LLM into an orchestrator that can query databases, call APIs, or trigger workflows. Prompt templates bring discipline to how prompts are constructed, and batch processing shows how to make many requests efficiently when your workload is not interactive. These three patterns are often combined: a function-calling agent uses templates to build its prompts and batches the tool results it needs to summarize. Approach each one as a building block rather than a finished feature, since real applications almost always need several of them working together.
Function Calling
Function calling, also known as tool calling, is the mechanism behind modern agentic features.
You declare a set of functions with JSON Schema descriptions, and the model responds either with text or with a structured request to invoke one of those functions, including the exact arguments.
Your application then executes the function and feeds the result back for another model turn.
The example below declares a get_weather tool and lets the model decide, based on the user’s query, whether to call it.
The key design decision is that you remain in control of execution—the model proposes the call, but your code validates the arguments and performs the side effect.
This is what keeps agentic behavior safe: the model can never execute anything directly, only suggest a function invocation that your trusted code then runs.
import json
def process_with_function_calling(user_query):
"""Use LLM function calling for structured outputs."""
client = OpenAI()
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
}
]
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": user_query}],
tools=tools,
tool_choice="auto"
)
if response.choices[0].message.tool_calls:
for tool_call in response.choices[0].message.tool_calls:
print(f"Function: {tool_call.function.name}")
print(f"Args: {tool_call.function.arguments}")
return response
# Usage
response = process_with_function_calling("What's the weather in New York?")
Prompt Templates
Inline f-strings are fine for a prototype, but prompts grow into a maintenance problem in production: they are scattered through code, drift out of sync, and make it hard to experiment with wording. Prompt templates centralize the prompt text in one place and substitute variables at call time. This makes prompts reviewable, versionable, and testable like any other code artifact, and it enables A/B testing of different instructions without touching the calling logic.
The example uses Python’s string.Template for straightforward variable substitution.
Note that templates should only interpolate data, never executable logic, and the substitution must be strict—a missing variable should fail loudly rather than silently producing a broken prompt.
For larger teams, a dedicated prompt management system gives you versioning, per-environment overrides, and usage analytics, but even this lightweight pattern prevents the worst kind of drift: prompts hardcoded in fifteen different places.
Whatever approach you choose, keep the template text out of the data path so that user input cannot be interpreted as template instructions.
from string import Template
class PromptTemplate:
"""Manage prompt templates with variable substitution."""
def __init__(self, template_str):
self.template = Template(template_str)
def format(self, **kwargs):
"""Format template with variables."""
return self.template.substitute(**kwargs)
# Usage
template = PromptTemplate("""
You are a $role.
Task: $task
Context: $context
""")
prompt = template.format(
role="Python expert",
task="Explain decorators",
context="For beginners"
)
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
Batch Processing
Some workloads are not interactive: classifying thousands of documents, summarizing a backlog of tickets, or generating metadata for a large catalog. Calling the API once per item is slow and expensive, and issuing all requests at once will hit rate limits. Batching the work into chunks gives you a predictable pace that respects provider limits while still making progress. It also makes progress tracking and resumability straightforward—if a batch fails, you can resume from the checkpoint rather than restarting the whole job.
The example processes items in fixed-size batches, making one API call per item but pacing the work so the provider is never overwhelmed. In practice you would parallelize within each batch with a thread pool or async tasks, since each call is independent and latency-bound. For genuinely huge jobs, every major provider also offers an asynchronous batch API with a significant discount in exchange for deferred results—if your workload tolerates an hours-long turnaround, the synchronous loop below is the wrong tool and the batch endpoint is a large cost win. Aim for idempotent processing: if your output records can be safely written twice, retrying a failed batch becomes trivial and safe. Finally, expose a simple progress counter or status endpoint so operators can see where a long-running batch stands at a glance.
from typing import List
def batch_process_with_llm(items: List[str], batch_size: int = 10):
"""Process multiple items with LLM."""
client = OpenAI()
results = []
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
for item in batch:
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": f"Process: {item}"}]
)
results.append(response.choices[0].message.content)
print(f"Processed {min(i + batch_size, len(items))}/{len(items)}")
return results
# Usage
items = [f"Item {i}" for i in range(100)]
results = batch_process_with_llm(items)
Common Pitfalls and Best Practices
The patterns above are the positive path; this section shows the failure modes that account for most production incidents. Each pair of examples contrasts a naive implementation with the corrected version. The recurring theme is that LLM code fails in ways that are both more frequent and more varied than ordinary HTTP services, so the defaults you choose—unhandled exceptions, unbounded state, missing timeouts—determine whether your feature degrades gracefully or falls over in production. As you review each pair, notice that the corrections add no dependencies or complex machinery; they simply take failure seriously. Adopting these habits early is far cheaper than retrofitting them after an incident.
❌ Bad: No Error Handling
# DON'T: Assume API calls always succeed
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
✅ Good: Comprehensive Error Handling
The naive call above has no try/except and no timeout: a rate limit, a server error, or a slow response either crashes the request or hangs the worker indefinitely.
Any of these surfaces to the user as a 500 error with no path to recovery.
The corrected version catches the specific error types the provider SDK defines.
RateLimitError tells you to back off, APIError covers transport and server problems, and a broad except handles anything unexpected.
A timeout guarantees the call cannot hang forever.
Together these three elements—typed exceptions, a timeout, and an explicit retry policy—are the minimum bar for a production integration.
# DO: Handle errors gracefully
try:
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
timeout=30
)
except RateLimitError:
# Handle rate limiting
pass
except APIError as e:
# Handle API errors
pass
❌ Bad: Unbounded Conversation History
The second most common production bug is letting conversation history grow without bound. Every message in the list is sent with every request, and each new turn appends more. In a long session the context window fills, the request becomes slower and more expensive, and eventually the API rejects it because the token count exceeds the model’s limit—usually at the worst possible time.
# DON'T: Keep growing conversation history indefinitely
for user_input in user_inputs:
messages.append({"role": "user", "content": user_input})
# Messages keep growing!
✅ Good: Manage Context Window
The fix is to bound the history you send, trading a little context for predictable cost and reliability.
Keeping only the most recent MAX_MESSAGES turns preserves the conversational thread while capping cost and latency.
More sophisticated versions summarize older messages into a compact summary before they are dropped, so long conversations keep their essential context.
The principle is the same everywhere: decide explicitly what you are willing to spend, and let that decision bound what you send.
# DO: Limit conversation history
MAX_MESSAGES = 20
def add_message_with_limit(messages, role, content):
messages.append({"role": role, "content": content})
if len(messages) > MAX_MESSAGES:
messages = messages[-MAX_MESSAGES:]
return messages
Production Deployment
Reliability patterns matter only if you can see whether they are working. In production you need to know how often calls fail, how long they take, what they cost, and whether the model’s behavior is drifting from what your prompts intend. The final pattern in this guide is observability: logging every call with enough context to debug a bad response, measure latency, and spot cost anomalies before they become bills. An LLM integration without monitoring is a black box, and debugging a bad model response without request logs is nearly impossible.
Monitoring and Logging
The example instruments a single call with start and end timestamps and structured log lines. In a real deployment these logs feed a central system where you can query per-model latency, error rates by error type, and token usage per feature. Because LLM calls are non-deterministic, the log should capture the request that produced a bad response—prompt, model, temperature, and a trace ID—so you can reproduce and fix it. This is also the hook for alerting: a spike in retries or cost should page someone before users notice a problem. Many teams additionally record a hash of each prompt and response so they can later audit for drift or reproduce a specific production incident. Keep the instrumentation structured rather than free-form, so it can be queried and aggregated; latency, tokens, and cost belong in a time-series store, while full request and response bodies belong in a searchable log store with retention limits.
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def call_llm_with_logging(prompt, model="gpt-4"):
"""Call LLM with comprehensive logging."""
client = OpenAI()
start_time = datetime.now()
logger.info(f"Starting LLM call with model: {model}")
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
duration = (datetime.now() - start_time).total_seconds()
logger.info(f"LLM call completed in {duration:.2f}s")
return response.choices[0].message.content
except Exception as e:
logger.error(f"LLM call failed: {e}")
raise
With monitoring in place, the full pattern is complete: simple calls for the happy path, conversation management for context, streaming for responsiveness, retries and timeouts for resilience, token counting and caching for cost control, function calling for structure, and observability for operations. Apply these together and LLM-powered features move from fragile demos to dependable production services.
Summary
Integrating LLMs into production applications requires:
- Robust error handling with retry logic and timeouts
- Cost optimization through token counting and caching
- Streaming for better user experience
- Conversation management with context window limits
- Monitoring and logging for production visibility
- Function calling for structured outputs
- Batch processing for efficiency
These patterns ensure reliable, cost-effective, and performant LLM-powered applications.
As a final checklist before shipping: every LLM call should have a timeout, an explicit retry policy, a cap on tokens, and a log line that records what was sent and received. Where the same prompt is repeated across users, add caching; where responses are interactive, stream them. These few defaults will prevent the majority of production incidents and make the remaining ones diagnosable in minutes rather than days.
Comments