Best Practices für die Produktion
Auf dieser Seite
Vom Prototyp zur Bereitstellung
Sie beherrschen nun jeden Parameter der Chat-Completions-API. Der schwierigste Teil bleibt: Die Lücke zwischen einem funktionierenden Notebook und einem Dienst für andere Nutzer zu überbrücken, ohne dass Sie ständig eingreifen müssen. Diese letzte Lektion fasst den gesamten Kurs in einer handlungsorientierten Produktions-Checkliste zusammen und weist den Weg zu weiteren Kursen.
Authentifizierung und Modellversionierung
Zwei Entscheidungen stehen vor der ersten Zeile geschäftskritischer Logik. Die erste betrifft den API-Schlüssel: Er lebt in einer Umgebungsvariablen, niemals im Code, und eine Start-Prüfung (assertion) ist besser als ein None, das sich drei Ebenen tief in eine unverständliche Fehlermeldung fortsetzt. Planen Sie auch Rotationen vor: Erstellen Sie den neuen Schlüssel, setzen Sie ihn bereit, und widerrufen Sie den alten erst danach, wenn Sie vermeiden wollen, dass der Dienst offline geht.
import os
# IMMER: API-Schlüssel in einer Umgebungsvariablen
api_key = os.getenv("MISTRAL_API_KEY")
assert api_key, "MISTRAL_API_KEY nicht definiert"
# NIEMALS: Hardcodierter Schlüssel im Code
# api_key = "sk-abc123..." # VERBOTEN in der Produktion
# IMMER: Regelmäßige Schlüsselrotation
# Erstellen Sie einen neuen Schlüssel, setzen Sie ihn bereit, und widerrufen Sie den altenDie zweite Entscheidung betrifft die Modell-ID. In der Entwicklung liefert -latest Verbesserungen ohne Aufwand. In der Produktion bedeutet derselbe Alias jedoch, dass eine Änderung auf Seiten des Anbieters das Verhalten Ihrer Anwendung ohne Warnung verändern kann – am Dienstagmorgen, ohne Commit-Nachverfolgung, die den Grund erklärt. Fixieren Sie die Version.
# ENTWICKLUNG — neueste Version
model = "mistral-large-latest"
# PRODUKTION — feste Version für Reproduzierbarkeit
model = "mistral-large-2512"
# Warum? Eine Änderung bei "latest" kann das Verhalten der
# Anwendung ohne Warnung verändern.Ein Client, der Vorfälle bewältigt
Das Herzstück einer produktionsreifen Integration ist nicht der API-Aufruf selbst, sondern alles drumherum. Die Klasse unten bündelt drei Verantwortlichkeiten in einer einzigen Engstelle: Wiederholung mit exponentiellem Backoff bei Statuscode 429, eine kurze Pause bei Serverfehlern und die sofortige Eskalation von Clientfehlern, die sich nicht durch Wiederholung beheben lassen. Zudem normalisiert sie die Antwort in ein stabiles Wörterbuch – Inhalt, finish_reason, Nutzung, Modell, ID –, sodass der Rest Ihres Codes nicht mehr von der genauen Struktur des vom SDK zurückgegebenen Objekts abhängt. Nur der logger.warning bei Trunkierungen rechtfertigt bereits die Kapselung: Es ist der einzige Weg, um festzustellen, dass eine max_tokens-Einstellung zu eng ist, bevor Nutzer davon berichten.
from mistralai.client import Mistral
from mistralai.client.errors import SDKError
import time
import logging
logger = logging.getLogger(__name__)
class MistralClient:
"""Produktions-Client mit Wiederholung, Timeout und Protokollierung."""
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,
}
# Warnung, wenn die Antwort abgeschnitten wurde
if result["finish_reason"] == "length":
logger.warning(f"Antwort abgeschnitten — req_id={response.id}")
return result
except SDKError as e:
if e.status_code == 429:
wait = 2 ** attempt
logger.warning(f"Ratenlimit, Wiederholung {attempt+1}/{self.max_retries} in {wait}s")
time.sleep(wait)
elif e.status_code >= 500:
logger.error(f"Serverfehler {e.status_code}, Wiederholung {attempt+1}")
time.sleep(1)
else:
logger.error(f"Clientfehler {e.status_code}: {e.message}")
raise
raise Exception(f"Fehlschlag nach {self.max_retries} Versuchen")Kosten in Echtzeit verfolgen
Das Entdecken des Verbrauchs auf der monatlichen Rechnung ist ein Luxus, den sich nur wenige Teams leisten können. Der Tracker unten schreibt pro Aufruf eine JSONL-Zeile – Zeitstempel, Modell, Eingabe- und Ausgabetokens, finish_reason, Anforderungs-ID – und hält eine Sitzungssumme, die Sie jederzeit abfragen können. Das JSONL-Format ist kein Detail: Es aggregiert mit jedem Analysetool ohne einen zusätzlichen Parsing-Schritt, und das Vorhandensein der Anforderungs-ID ermöglicht es Ihnen, eine Kostenanomalie mit einem spezifischen Aufruf zu verknüpfen.
import json
from datetime import datetime, timezone
class CostTracker:
"""Echtzeit-Verfolgung der API-Kosten."""
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"Sitzung — Eingabe: {self.session_tokens['prompt']:,} | "
f"Ausgabe: {self.session_tokens['completion']:,} | "
f"Gesamt: {total:,} Tokens"
)Validierung an beiden Enden
Eine robuste Anwendung vertraut weder dem, was hineingeht, noch dem, was herauskommt. Bei der Eingabe lehnen Sie leere Nachrichten ab, begrenzen die Länge auf 10.000 Zeichen, um zu verhindern, dass ein unglückliches Kopieren und Einfügen Ihre Rechnung in die Höhe treibt, und entfernen Steuerzeichen, die die Tokenisierung stören. Bei der Ausgabe lehnen Sie nichts ab, sondern heben Warnungen hervor: potenziell abgeschnittene Antwort, ungewöhnlich kurze Antwort unter fünf Zeichen, ungewöhnlich lange Antwort über 50.000. Diese drei Signale erfassen den Großteil der stillen Fehler.
def validate_input(user_message: str) -> str:
"""Validierung und Bereinigung der Benutzereingabe."""
if not user_message or not user_message.strip():
raise ValueError("Leere Nachricht")
if len(user_message) > 10_000:
raise ValueError("Nachricht zu lang (max. 10.000 Zeichen)")
# Steuerzeichen entfernen
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:
"""Validierung der API-Ausgabe."""
warnings = []
if finish_reason == "length":
warnings.append("Antwort potenziell abgeschnitten")
if not content or len(content.strip()) < 5:
warnings.append("Ungewöhnlich kurze Antwort")
if len(content) > 50_000:
warnings.append("Ungewöhnlich lange Antwort")
return {"content": content, "warnings": warnings, "valid": len(warnings) == 0}Das Kontextfenster halten
Ein letztes Problem: der langsame Drift bei langen Gesprächen. Die Funktion unten schätzt das Volumen auf ein Token pro vier Zeichen im Französischen – eine grobe Annäherung, aber ausreichend als Schutzzaun – und lässt Anfragen passieren, solange sie unter 80 % des Limits bleiben. Darüber hinaus bewahrt sie den System-Prompt und entfernt die ältesten Nachrichten, bis sie wieder unter 70 % fallen. Dieser Abstand zwischen den beiden Schwellenwerten ist absichtlich: Ein Schnitt nur an der Auslöseschwelle würde die Funktion im nächsten Schritt wieder in den roten Bereich drängen.
def manage_context(messages: list, max_tokens: int = 100_000) -> list:
"""Nachrichten innerhalb des Token-Limits des Modells halten."""
# Schätzung: 1 Token ~ 4 Zeichen im Französischen
total_chars = sum(len(m["content"]) for m in messages)
estimated_tokens = total_chars // 4
if estimated_tokens < max_tokens * 0.8:
return messages # Alles passt
# System-Prompt + erste und letzte Nachrichten behalten
system = [m for m in messages if m["role"] == "system"]
history = [m for m in messages if m["role"] != "system"]
# Älteste Nachrichten entfernen
while estimated_tokens > max_tokens * 0.7 and len(history) > 2:
removed = history.pop(0)
estimated_tokens -= len(removed["content"]) // 4
return system + historyWas Sie aus diesem Kurs mitnehmen
Die Reise folgte vier Schritten. Erstens, die Grundlagen: der Endpunkt, Rollen, grundlegende Parameter und die Antwortstruktur. Zweitens, Streaming mit dem SSE-Protokoll, seiner Python-Implementierung, Stop-Sequenzen und der Absicherung durch safe_prompt und Guardrails. Drittens, Prompting – Produktions-System-Prompts, Few-Shot, XML- und Markdown-Strukturierung, Anti-Muster. Viertens, Erweiterte Parameter, schlossen den Kreis: Temperatur und Top P im Detail, Präsenz- und Häufigkeitsstrafen, N-Vervollständigungen und diese Produktions-Checkliste.
Die folgende Konfiguration fasst diese Entscheidungen zu einem verteidigbaren Ausgangspunkt für eine erste produktionsreife Bereitstellung zusammen: feste Version, Token-Obergrenze, die dem Anwendungsfall entspricht, niedrige Temperatur für faktische Konsistenz, top_p unverändert, da die Temperatur angepasst wird, neutrale Strafen, es sei denn, ein spezifischer Bedarf rechtfertigt sie, und safe_prompt aktiviert.
# Typische Konfiguration für eine produktionsreife Anwendung
PRODUCTION_CONFIG = {
"model": "mistral-large-2512", # Feste Version
"max_tokens": 2000, # Entspricht dem Anwendungsfall
"temperature": 0.3, # Konsistent und faktisch
"top_p": 1.0, # Unverändert, wenn Temperatur angepasst
"presence_penalty": 0.0, # Standard, es sei denn, spezifischer Bedarf
"frequency_penalty": 0.0, # Standard, es sei denn, spezifischer Bedarf
"safe_prompt": True, # Sicherheit aktiviert
}Um mit der Mistral-API weiterzumachen, folgen auf der Corsen Academy vier ergänzende Kurse:
- Embeddings und RAG – Semantische Suche mit
mistral-embedund Vektordatenbanken - Function Calling – Verbinden Sie das Modell mit Ihren Tools und externen APIs
- Feinabstimmung – Anpassen eines Modells an Ihren spezifischen Bereich
- Produktion und Evaluation – Metriken, automatisierte Evaluierungen, Kostenoptimierung
Wichtige Erkenntnisse
- Verwenden Sie in der Produktion versionierte Modelle, niemals
-latest - Implementieren Sie von der ersten Bereitstellung an Wiederholung + Backoff + Protokollierung
- Validieren Sie sowohl Eingaben als auch Ausgaben jeder API-Anfrage
- Verfolgen Sie Kosten in Echtzeit mit einem Token-Tracker
- Verwalten Sie das Kontextfenster für lange Gespräche
- Aktivieren Sie
safe_promptals minimales Sicherheitsnetz - Testen Sie Ihre Prompts mit Grenzfällen vor jeder produktionsreifen Veröffentlichung
Wissen überprüfen
Keine Geheimnisse mehr im Endpunkt für Chat/Completions? Lassen Sie es uns überprüfen.
1. Welche Rollen strukturieren die Nachrichtenliste?
Antwort: system (der Rahmen), user (Anfragen), assistant (Modellantworten) – die vollständige Historie startet bei jedem Aufruf neu, da die API zustandslos ist.
2. Was finden Sie im usage-Feld der Antwort?
Antwort: Verwendete Tokens bei Eingabe und Ausgabe – die Grundlage für die Abrechnung und die Metrik zur Überwachung für Dimensionierung und Kosten.
3. Wie funktioniert SSE-Streaming?
Antwort: Die Antwort kommt als Server-Sent Events, Stück für Stück an: Sie anzeigen, was kommt, und concatenieren, um die endgültige Nachricht zu erhalten.
4. Wozu dienen Stop-Sequenzen?
Antwort: Um die Generierung abzuschneiden, sobald eine gegebene Zeichenfolge erscheint: Länge steuern, bei einem Delimiter stoppen, saubere Formate garantieren.
5. Temperatur, Top P, Strafen: Wie in der Produktion verwenden?
Antwort: Temperatur/Top P setzen Determinismus je Aufgabe; Präsenz-/Häufigkeitsstrafen begrenzen Wiederholungen – Einstellungen, die je Anwendungsfall festgelegt und zusammen mit den Prompts dokumentiert werden müssen.
Rollen, Nutzung, Streaming, Stops, Tuning: Die Grundlagen halten – die oben genannten Best Practices für die Produktion bringen sie in Ordnung.