Dieses Tutorial zeigt, wie man einen kleinen, aber produktionsreifen Chatbot-Service mit FastAPI, SSE-Streaming, Konversationsspeicher und Modellwechsel erstellt. Das Ziel ist kein Spielzeug-Demo. Das Ziel ist ein Backend, das Sie hinter einer echten Produktoberfläche einsetzen und sicher weiterentwickeln können, ohne Ihre Integration jedes Mal neu schreiben zu müssen, wenn Sie das Modell wechseln.
Wenn Sie bereits ein OpenAI-kompatibles SDK auf TokenLab ausgerichtet haben, knüpft dieser Artikel genau dort an. Falls Sie den Basis-URL-Wechsel noch nicht durchgeführt haben, lesen Sie zuerst den Migrationsleitfaden. Wenn Ihr Hauptanliegen die Formung von Anfragen und Backoff unter Last ist, kombinieren Sie diesen Leitfaden mit dem Leitfaden zum Rate Limiting der AI API.
Wichtige Erkenntnisse
- Ein produktionsreifer Chatbot benötigt sechs Komponenten: einen synchronen Endpunkt, einen Streaming-Endpunkt, serverseitigen Konversationsstatus, eine Modell-Allowlist, echte Fehlerbehandlung und einen klaren Pfad für Speicher-Upgrades.
- Testen Sie Ihren Key, die Basis-URL und das Routing mit einem kleinen Chat-Endpunkt, bevor Sie Streaming, Speicher oder Tool-Calls hinzufügen.
- SSE-Streaming deckt die meisten Chat-Produkte ab und verursacht weniger operativen Overhead als WebSockets.
- Stellen Sie Modelle über eine Backend-Allowlist bereit, nicht über ein freies Textfeld, damit das Frontend keine beliebigen oder veralteten Modell-IDs anfordern kann.
- Die Verfügbarkeit und das Angebot von Modellen ändern sich häufig. Überprüfen Sie das Modellverzeichnis von TokenLab (Stand 07.07.2026), bevor Sie Ihre Allowlist für die Produktion festlegen.
Was wir bauen
Der fertige Service besteht aus sechs beweglichen Teilen:
- Ein synchroner
/chat-Endpunkt für Smoke-Tests. - Ein Streaming-
/chat/stream-Endpunkt für die echte UI. - Konversationsstatus, identifiziert durch
conversation_id. - Eine Modell-Allowlist, damit das Frontend keine beliebigen IDs anfordern kann.
- Fehlerbehandlung, die nicht beim ersten 429-Fehler zusammenbricht.
- Ein klarer Pfad vom In-Memory-Prototyp zu Redis oder PostgreSQL.
Das reicht aus, um einen Support-Bot, einen internen Assistenten oder die erste Version eines eingebetteten Chat-Widgets zu betreiben.
Installieren Sie den minimalen Stack
pip install fastapi uvicorn openai pydantic redis
Sie können redis für den ersten Durchlauf überspringen, aber das Einbinden des Imports macht das spätere Upgrade zu einer Kleinigkeit statt zu einem Refactoring.
Schritt 1: Starten Sie mit einem kleinen, langweiligen Chat-Endpunkt
Der schnellste Weg, sich bei der Arbeit an Chatbots zu verlieren, ist der Start mit WebSockets, Tool-Nutzung und Agenten-Orchestrierung, bevor der grundlegende Anforderungspfad stabil ist. Beginnen Sie mit einem kleinen Endpunkt, der beweist, dass Ihr Key, die Basis-URL und das Modell-Routing funktionieren.
from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel
app = FastAPI()
client = OpenAI(
api_key="sk-tokenlab-xxx",
base_url="https://api.tokenlab.sh/v1"
)
class ChatRequest(BaseModel):
message: str
model: str = "deepseek-v4-flash"
conversation_id: str | None = None
@app.post("/chat")
async def chat(req: ChatRequest):
response = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}]
)
return {"reply": response.choices[0].message.content}
Führen Sie einen Smoke-Test durch. Wenn dieser fehlschlägt, bauen Sie nicht darauf weiter auf.
Schritt 2: Fügen Sie Streaming hinzu, da Benutzer Latenz spüren, bevor sie sie messen
Die meisten Chatbot-Produkte fühlen sich nicht langsam an, weil das Modell langsam ist, sondern weil die UI leer bleibt, bis die vollständige Antwort eintrifft. SSE reicht für die meisten Chat-Produkte aus und hat einen geringeren operativen Aufwand als WebSockets.
from fastapi.responses import StreamingResponse
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
def generate():
stream = client.chat.completions.create(
model=req.model,
messages=[{"role": "user", "content": req.message}],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
yield f"data: {delta.content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
Auf dem Frontend reicht ein einfacher Fetch-Reader aus:
async function sendMessage(payload) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value, { stream: true });
console.log(chunk);
}
}
Wenn Ihr Produkt bereits einen Browser-Client über Standard-HTTP betreibt, hält SSE die Architektur einfacher als eine WebSocket-Schicht.
Schritt 3: Verschieben Sie den Konversationsstatus aus dem Request Body
Die erste Chatbot-Demo behält das vollständige Transkript meist im Browser und sendet es bei jedem Turn erneut. Das funktioniert für Prototypen. Es wird jedoch schnell unübersichtlich, wenn Sie Wiederholungsversuche, fortsetzbare Sitzungen oder serverseitige Tools benötigen.
Ein In-Memory-Speicher ist für den Anfang in Ordnung:
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
def build_messages(conv_id: str, user_msg: str) -> list:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
history = conversations[conv_id][-20:]
messages.extend(history)
messages.append({"role": "user", "content": user_msg})
conversations[conv_id].append({"role": "user", "content": user_msg})
return messages
Der Upgrade-Pfad zu Redis ist hauptsächlich Speicher-Infrastruktur, keine Logikänderung:
import json
import redis
redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def load_history(conv_id: str) -> list:
raw = redis_client.get(f"chat:{conv_id}")
return json.loads(raw) if raw else []
def save_history(conv_id: str, history: list) -> None:
redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))
Greifen Sie zu Redis, wenn Konversationen TTL, Wiederaufnehmbarkeit oder Multi-Instanz-Deployment benötigen. Greifen Sie zu PostgreSQL, wenn das Transkript selbst Produktdaten sind, die Sie später abfragen, prüfen oder auswerten müssen.
Schritt 4: Behandeln Sie Fehler als Produktverhalten, nicht nur als Exceptions
Wenn Ihr Chatbot kundenorientiert ist, ist der Fehlerpfad genauso wichtig wie der Erfolgsfall. Einem Benutzer ist es egal, ob der Fehler durch Rate Limiting, ein erschöpftes Guthaben oder einen Modellausfall beim Anbieter verursacht wurde. Wichtig ist, ob die UI einfriert.
from openai import APIConnectionError, APIError, RateLimitError
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
messages = build_messages(conv_id, req.message)
def generate():
full_response = []
try:
stream = client.chat.completions.create(
model=req.model,
messages=messages,
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
full_response.append(delta.content)
yield f"data: {delta.content}\n\n"
except RateLimitError:
yield "data: [error: rate limited, please retry shortly]\n\n"
except APIConnectionError:
yield "data: [error: connection issue, please retry]\n\n"
except APIError:
yield "data: [error: something went wrong on our end]\n\n"
finally:
if full_response:
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
Schritt 5: Sperren Sie, welche Modelle das Frontend anfordern darf
Lassen Sie das Frontend niemals einen beliebigen Modell-String direkt an die API übergeben. Ein freies Textfeld lädt dazu ein, veraltete Modelle, Tippfehler oder Modelle anzufordern, die Sie nie bereitstellen wollten. Routen Sie stattdessen über eine Backend-Allowlist.
ALLOWED_MODELS = {
"default": "deepseek-v4-flash",
"flagship": "gpt-5.5",
"balanced": "claude-sonnet-5",
}
class ChatRequest(BaseModel):
message: str
tier: str = "default"
conversation_id: str | None = None
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
model = ALLOWED_MODELS.get(req.tier, ALLOWED_MODELS["default"])
# Der Rest der Streaming-Logik verwendet `model` statt eines vom Client gelieferten Strings
Dies gibt Ihnen einen zentralen Ort, um Modelle auszutauschen, wenn ein Anbieter eines veraltet, ohne den Frontend-Code anzufassen oder ein Client-Update auszuliefern.
Schritt 6: Behandeln Sie den Rest der Produktion, nicht nur den Erfolgsfall
Ein Chatbot-Backend gilt als produktionsreif, wenn die Randbedingungen abgedeckt sind, nicht wenn der Kern-Chat-Aufruf besonders clever ist.
Die Checkliste ist kurz:
- Fügen Sie Request-IDs hinzu, damit Sie Frontend-Fehler mit Backend-Logs verknüpfen können.
- Begrenzen Sie die Nebenläufigkeit pro Benutzer und die Anforderungsgröße.
- Kürzen Sie lange Historien, bevor sie Ihr Token-Budget sprengen.
- Loggen Sie Modell, Latenz, Eingabegröße und den Grund für den Abschluss.
- Trennen Sie benutzerseitige Fehlermeldungen von internen Fehlerdetails.
- Testen Sie ein alternatives Modell, damit Sie wissen, dass das Fallback funktioniert, bevor der erste echte Ausfall auftritt.
Das Kürzen der Historie kann einfach bleiben:
def trim_history(messages: list, max_tokens: int = 8000) -> list:
system = messages[0]
history = messages[1:]
total_chars = len(system["content"])
trimmed = []
for msg in reversed(history):
msg_chars = len(msg["content"])
if total_chars + msg_chars > max_tokens * 4:
break
trimmed.insert(0, msg)
total_chars += msg_chars
return [system] + trimmed
Es geht nicht um token-perfekte Abrechnung. Es geht darum, offensichtliche Kontext-Explosionen zu stoppen, bevor sie Ihre Rechnung oder Ihre Latenz belasten.
Von der Demo zum Produkt
Sobald dieses Backend stabil ist, ist das nächste Upgrade selten „mehr KI“. Meistens ist es langweilige Infrastruktur:
- Authentifizierung, damit ein Benutzer nicht die Konversation eines anderen lesen kann.
- Persistenz, damit Sitzungen Deploys überleben.
- Rate Limiting, damit ein „lauter“ Benutzer nicht Ihr Kontingent aufbraucht.
- Abrechnung oder Nutzungszuordnung, wenn der Chatbot kundenorientiert ist.
- Hintergrund-Zusammenfassungen, wenn Konversationen ein Langzeitgedächtnis benötigen.
Ein einheitliches Gateway hilft bei den meisten dieser Punkte. Sobald die Basis-URL-Migration hinter Ihnen liegt, werden Modelländerungen von einem Plattform-Rewrite zu einer einfachen Konfigurationsänderung.
Smoke-Test
uvicorn main:app --reload --port 8000
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "model": "deepseek-v4-flash"}'
Wenn Sie einen Turn streamen, eine Konversation beibehalten und bei einem erzwungenen Fehler eine saubere Fehlermeldung zurückgeben können, haben Sie das richtige Fundament.
Kostenschätzung
Erstellen Sie einen API-Key bei TokenLab, richten Sie Ihr OpenAI-SDK auf https://api.tokenlab.sh/v1 aus, und Sie können die erste Produktionsversion Ihres Chatbots ausliefern, ohne separate Konten bei verschiedenen Anbietern verwalten zu müssen.
| Modell | Typische Stufe | Hinweise |
|---|---|---|
| DeepSeek V4 Flash | Schnell / Standard | Guter Standard für hohes Volumen und niedrige Latenz |
| GPT-5.5 | Flaggschiff | Für Turns, die stärkeres Reasoning erfordern |
| Claude Sonnet 5 | Ausgewogen | Starke Wahl für Coding und Review-Antworten |
| Gemini 3.5 Flash | Budget / schnelle Alt. | Kostengünstige, schnelle Alternative für hohes Volumen |
Die genauen Preise pro Token ändern sich häufig bei den Anbietern und werden hier nicht als feste Zahlen wiedergegeben. Überprüfen Sie die aktuellen Raten im Modellverzeichnis (Stand 07.07.2026), bevor Sie Ihr Budget planen. In der Praxis hält das Routing der meisten Konversationen auf eine schnelle/Standard-Stufe wie DeepSeek V4 Flash oder Gemini 3.5 Flash und die Reservierung von GPT-5.5 oder Claude Sonnet 5 für anspruchsvollere Aufgaben die monatlichen Kosten niedrig. Bestätigen Sie jedoch die tatsächlichen Raten pro Million Token für Ihr Konto, bevor Sie sich auf ein Budget festlegen.
FAQ
Brauche ich WebSockets, um einen KI-Chatbot zu bauen? Nein. SSE-Streaming, wie in Schritt 2 gezeigt, deckt die überwiegende Mehrheit der Chat-Produkte ab. WebSockets bieten echten Mehrwert, wenn Sie bidirektionales Push außerhalb von Request/Response benötigen, wie bei Live-Kollaboration oder serverinitiierten Ereignissen. Für eine Standard-Chat-UI ist SSE einfacher bereitzustellen, zu debuggen und zu skalieren.
Woher weiß ich, welches Modell ich als Standard verwenden soll? Starten Sie mit einem schnellen, kostengünstigen Modell wie DeepSeek V4 Flash oder Gemini 3.5 Flash für die Standard-Stufe und fügen Sie eine ausgewogene oder Reasoning-Stufe mit Claude Sonnet 5 oder GPT-5.5 hinter der in Schritt 5 gezeigten Allowlist hinzu. Überprüfen Sie das Modellverzeichnis (Stand 07.07.2026) auf aktuelle Optionen, da neue Modelle erscheinen und ältere nach einem Zeitplan außerhalb Ihrer Kontrolle veraltet sind.
Was geht zuerst kaputt, wenn ein Chatbot von der Demo in den echten Betrieb geht? Fast immer der Fehlerpfad, nicht der Erfolgsfall. Unbegrenzte Wiederholungsversuche, fehlende Limits für die Nebenläufigkeit pro Benutzer und eine unbegrenzte Konversationshistorie sind die drei häufigsten Ursachen dafür, dass ein Chatbot-Backend unter echter Last zusammenbricht. Die Schritte 4 und 6 oben adressieren alle drei direkt.
Starten Sie mit Ihrem API-Key: tokenlab.sh bietet 300+ Modelle über einen Endpunkt. $1 Startguthaben für den Aufbau.
Quellen
Preis geprüft am 2026-07-07
- TokenLab model directoryGeprüft am 2026-07-07
