Production Best Practices

From Proof of Concept to Deployment

You now master every parameter of the Chat Completions API. The trickiest part remains: bridging the gap between a working notebook and a service others use without your intervention. This final lesson consolidates the entire course into an actionable production checklist, then points the way to follow-up courses.

Authentication and Model Versioning

Two decisions come before the first line of business logic. The first concerns the API key: it lives in an environment variable, never in code, and a startup assertion beats a None that propagates three layers down into an incomprehensible error message. Plan for rotation too—create the new key, deploy it, then revoke the old one, in that order, if you want to avoid taking the service offline.

import os

# ALWAYS: API key in environment variable
api_key = os.getenv("MISTRAL_API_KEY")
assert api_key, "MISTRAL_API_KEY not defined"

# NEVER: hardcoded key in code
# api_key = "sk-abc123..."  # FORBIDDEN in production

# ALWAYS: regular key rotation
# Create a new key, deploy it, then revoke the old one

The second decision concerns the model identifier. In development, -latest gives you improvements without effort. In production, that same alias means a version change on the provider side can alter your application’s behavior without warning, on a Tuesday morning, with no commit trace to explain it. Pin the version.

# DEVELOPMENT — latest version
model = "mistral-large-latest"

# PRODUCTION — fixed version for reproducibility
model = "mistral-large-2512"

# Why? A "latest" change can alter your application's behavior
# without warning.

A Client That Handles Incidents

The heart of a production integration isn’t the API call but everything around it. The class below bundles three responsibilities into a single choke point: retry with exponential backoff on 429s, a short pause on server errors, and immediate escalation of client errors that no retry will fix. It also normalizes the response into a stable dictionary—content, finish_reason, usage, model, ID—so the rest of your code no longer depends on the exact shape of the object returned by the SDK. The logger.warning on truncations alone justifies the encapsulation: it’s the only way to discover a max_tokens setting is too tight before users report it.

from mistralai.client import Mistral
from mistralai.client.errors import SDKError
import time
import logging

logger = logging.getLogger(__name__)

class MistralClient:
    """Production client with retry, timeout, and logging."""

    def __init__(self, api_key: str, max_retries: int = 3):
        self.client = Mistral(api_key=api_key)
        self.max_retries = max_retries

    def complete(self, messages: list, **kwargs) -> dict:
        for attempt in range(self.max_retries):
            try:
                response = self.client.chat.complete(
                    messages=messages,
                    **kwargs
                )

                result = {
                    "content": response.choices[0].message.content,
                    "finish_reason": response.choices[0].finish_reason,
                    "usage": {
                        "prompt": response.usage.prompt_tokens,
                        "completion": response.usage.completion_tokens,
                        "total": response.usage.total_tokens,
                    },
                    "model": response.model,
                    "id": response.id,
                }

                # Alert if response truncated
                if result["finish_reason"] == "length":
                    logger.warning(f"Response truncated — req_id={response.id}")

                return result

            except SDKError as e:
                if e.status_code == 429:
                    wait = 2 ** attempt
                    logger.warning(f"Rate limit, retry {attempt+1}/{self.max_retries} in {wait}s")
                    time.sleep(wait)
                elif e.status_code >= 500:
                    logger.error(f"Server error {e.status_code}, retry {attempt+1}")
                    time.sleep(1)
                else:
                    logger.error(f"Client error {e.status_code}: {e.message}")
                    raise

        raise Exception(f"Failed after {self.max_retries} attempts")

Track Costs as They Happen

Discovering your consumption on the monthly bill is a luxury few teams can afford. The tracker below writes one JSONL line per call—timestamp, model, input and output tokens, finish_reason, request ID—and maintains a session total you can query at any time. The JSONL format isn’t a detail: it aggregates with any analytics tool without a parsing step, and the presence of the request ID lets you tie a cost anomaly to a specific call.

import json
from datetime import datetime, timezone

class CostTracker:
    """Real-time API cost tracking."""

    def __init__(self, log_path: str = "api_costs.jsonl"):
        self.log_path = log_path
        self.session_tokens = {"prompt": 0, "completion": 0}

    def log(self, response_data: dict):
        usage = response_data["usage"]
        self.session_tokens["prompt"] += usage["prompt"]
        self.session_tokens["completion"] += usage["completion"]

        entry = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "model": response_data["model"],
            "prompt_tokens": usage["prompt"],
            "completion_tokens": usage["completion"],
            "finish_reason": response_data["finish_reason"],
            "id": response_data["id"],
        }

        with open(self.log_path, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def report(self) -> str:
        total = self.session_tokens["prompt"] + self.session_tokens["completion"]
        return (
            f"Session — Prompt: {self.session_tokens['prompt']:,} | "
            f"Completion: {self.session_tokens['completion']:,} | "
            f"Total: {total:,} tokens"
        )

Validate at Both Ends

A robust application trusts neither what goes in nor what comes out. On input, reject empty messages, cap length at 10,000 characters to prevent an unfortunate copy-paste from ballooning your bill, and strip control characters that disrupt tokenization. On output, don’t reject anything but raise warnings: potentially truncated response, abnormally short response under five characters, abnormally long response over 50,000. These three signals catch the vast majority of silent failures.

def validate_input(user_message: str) -> str:
    """Validate and clean user input."""
    if not user_message or not user_message.strip():
        raise ValueError("Empty message")

    if len(user_message) > 10_000:
        raise ValueError("Message too long (max 10,000 characters)")

    # Clean control characters
    cleaned = "".join(c for c in user_message if c.isprintable() or c in "\n\t")
    return cleaned.strip()

def validate_output(content: str, finish_reason: str) -> dict:
    """Validate API output."""
    warnings = []

    if finish_reason == "length":
        warnings.append("Response potentially truncated")

    if not content or len(content.strip()) < 5:
        warnings.append("Abnormally short response")

    if len(content) > 50_000:
        warnings.append("Abnormally long response")

    return {"content": content, "warnings": warnings, "valid": len(warnings) == 0}

Hold the Context Window

One last issue: the slow drift of long conversations. The function below estimates volume at one token per four characters in French—a rough approximation but sufficient as a guardrail—and lets requests pass as long as they stay under 80% of the limit. Beyond that, it preserves the system prompt and removes the oldest messages until dropping back under 70%. This gap between the two thresholds is deliberate: cutting only to the trigger threshold would push the function back into the red on the next message.

def manage_context(messages: list, max_tokens: int = 100_000) -> list:
    """Keep messages within the model's token limit."""
    # Estimate: 1 token ~ 4 characters in French
    total_chars = sum(len(m["content"]) for m in messages)
    estimated_tokens = total_chars // 4

    if estimated_tokens < max_tokens * 0.8:
        return messages  # Everything fits

    # Keep system + first and last messages
    system = [m for m in messages if m["role"] == "system"]
    history = [m for m in messages if m["role"] != "system"]

    # Remove oldest messages
    while estimated_tokens > max_tokens * 0.7 and len(history) > 2:
        removed = history.pop(0)
        estimated_tokens -= len(removed["content"]) // 4

    return system + history

What You Take Away from This Course

The journey followed four steps. First, the fundamentals: the endpoint, roles, basic parameters, and response structure. Next, streaming, with the SSE protocol, its Python implementation, stop sequences, and hardening via safe_prompt and guardrails. Prompting occupied the third phase—production system prompts, few-shot, XML and Markdown structuring, anti-patterns. Advanced parameters closed the loop: temperature and Top P in detail, presence and frequency penalties, N completions, and this production checklist.

The configuration below distills these choices into a defensible starting point for a first production deployment: fixed version, token ceiling suited to the use case, low temperature for factual consistency, top_p untouched since temperature is adjusted, neutral penalties unless a specific need justifies them, and safe_prompt enabled.

# Typical configuration for a production application
PRODUCTION_CONFIG = {
    "model": "mistral-large-2512",      # Fixed version
    "max_tokens": 2000,                  # Suited to use case
    "temperature": 0.3,                  # Consistent and factual
    "top_p": 1.0,                        # Untouched if temperature adjusted
    "presence_penalty": 0.0,             # Default unless specific need
    "frequency_penalty": 0.0,            # Default unless specific need
    "safe_prompt": True,                 # Safety enabled
}

To go further with the Mistral API, four complementary courses follow on Corsen Academy:

• Embeddings and RAG — Semantic search with mistral-embed and vector databases
• Function Calling — Connect the model to your tools and external APIs
• Fine-tuning — Adapt a model to your specific domain
• Production and Evaluation — Metrics, automated evals, cost optimization

Key Takeaways

• Use versioned models in production, never -latest
• Implement retry + backoff + logging from the first deployment
• Validate both inputs AND outputs of every API call
• Track costs in real time with a token tracker
• Manage the context window for long conversations
• Enable safe_prompt as a minimal safety net
• Test your prompts with edge cases before every production release

Test Your Knowledge

No more secrets in the chat/completions endpoint? Let’s verify.

1. Which roles structure the message list?

Answer: system (the frame), user (requests), assistant (model responses)—the full history restarts with each call, as the API is stateless.

2. What do you find in the usage field of the response?

Answer: Tokens consumed on input and output—the basis for billing and the metric to watch for sizing and costs.

3. How does SSE streaming work?

Answer: The response arrives as Server-Sent Events, fragment by fragment: you display as it comes and concatenate to get the final message.

4. What are stop sequences for?

Answer: To cut generation as soon as a given string appears: control length, stop at a delimiter, guarantee clean formats.

5. Temperature, Top P, penalties: what usage in production?

Answer: Temperature/Top P set determinism per task; presence/frequency penalties limit repetition—settings to fix per use case and document alongside your prompts.

Roles, usage, streaming, stops, tuning: the fundamentals hold—the production best practices above put them in order.