Erstellen Sie einen KI-Chatbot mit nur einem API-Schlüssel: Von Null bis zur Produktion in 30 Minuten
Dieses Tutorial zeigt, wie man ein produktionsbereites KI-Chatbot-Backend mit Streaming-Antworten, Gesprächshistorie, Modellwechsel und ordentlicher Fehlerbehandlung erstellt. Wir verwenden Python, FastAPI und das OpenAI SDK, das auf einen API-Aggregator zeigt, sodass Sie jedes Modell nutzen können.
Voraussetzungen
pip install fastapi uvicorn openai
Schritt 1: Basis-Chat-Endpunkt
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel
app = FastAPI()
client = OpenAI(
api_key="sk-lemon-xxx",
base_url="https://api.lemondata.cc/v1"
)
class ChatRequest(BaseModel):
message: str
model: str = "gpt-4.1-mini"
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}
Das funktioniert, bietet aber kein Streaming, keine Historie und keine Fehlerbehandlung. Das beheben wir jetzt.
Schritt 2: Streaming hinzufügen
Streaming sendet Tokens, sobald sie generiert werden, anstatt auf die vollständige Antwort zu warten. Nutzer sehen die Antwort in Echtzeit entstehen.
@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"
)
Schritt 3: Gesprächshistorie
Speichern Sie die Gesprächshistorie im Speicher (in der Produktion durch Redis oder eine Datenbank ersetzen).
from collections import defaultdict
import uuid
conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "Du bist ein hilfreicher Assistent. Sei prägnant und direkt."
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
conv_id = req.conversation_id or str(uuid.uuid4())
# Gesprächshistorie aufbauen
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages.extend(conversations[conv_id])
messages.append({"role": "user", "content": req.message})
# Benutzer-Nachricht speichern
conversations[conv_id].append(
{"role": "user", "content": req.message}
)
def generate():
full_response = []
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"
# Assistenten-Antwort speichern
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
yield f"data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"X-Conversation-ID": conv_id}
)
Schritt 4: Fehlerbehandlung
Aufrufe der KI-API können aus verschiedenen Gründen fehlschlagen: Rate Limits, unzureichendes Guthaben, Modell nicht verfügbar. Behandeln Sie jeden Fall:
from openai import (
APIError,
RateLimitError,
APIConnectionError
)
@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():
try:
full_response = []
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"
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
except RateLimitError as e:
yield f"data: [ERROR] Rate limit erreicht. Bitte warten Sie einen Moment.\n\n"
except APIConnectionError:
yield f"data: [ERROR] Verbindung fehlgeschlagen. Versuche es erneut...\n\n"
except APIError as e:
yield f"data: [ERROR] {e.message}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
def build_messages(conv_id: str, user_msg: str) -> list:
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
# Behalte die letzten 10 Runden, um die Kontextlänge zu verwalten
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
Schritt 5: Modellwechsel
Ermöglichen Sie den Nutzern, Modelle mitten im Gespräch zu wechseln. Verschiedene Modelle für unterschiedliche Anforderungen:
AVAILABLE_MODELS = {
"fast": "gpt-4.1-mini",
"smart": "claude-sonnet-4-6",
"reasoning": "o3",
"budget": "deepseek-chat",
"creative": "claude-sonnet-4-6",
}
@app.get("/models")
async def list_models():
return {"models": AVAILABLE_MODELS}
Das Frontend kann diese als Optionen darstellen. Da alle Modelle über den Aggregator das gleiche OpenAI-kompatible Format verwenden, ist der Wechsel einfach ein Ändern des model-Parameters.
Schritt 6: Verwaltung des Kontextfensters
Lange Gespräche überschreiten die Kontextgrenzen des Modells. Implementieren Sie ein gleitendes Fenster:
def trim_history(messages: list, max_tokens: int = 8000) -> list:
"""Behalte System-Prompt + aktuelle Nachrichten innerhalb des Token-Budgets."""
# Grobe Schätzung: 1 Token ≈ 4 Zeichen
system = messages[0] # System-Prompt immer behalten
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
Komplette Anwendung
# Starten mit: uvicorn main:app --reload --port 8000
# Testen: curl -N -X POST http://localhost:8000/chat/stream \
# -H "Content-Type: application/json" \
# -d '{"message": "Hallo!", "model": "gpt-4.1-mini"}'
Der gesamte Code umfasst weniger als 100 Zeilen. Von hier aus können Sie hinzufügen:
- Authentifizierung (API-Schlüssel oder JWT)
- Persistente Speicherung (PostgreSQL oder Redis für Gespräche)
- Rate Limiting pro Nutzer
- Nutzungsüberwachung und Abrechnung
- WebSocket-Unterstützung für bidirektionales Streaming
- Frontend (React, Vue oder reines JS mit EventSource)
Kostenschätzung
Für einen Chatbot, der 1.000 Gespräche/Tag mit durchschnittlich 5 Runden pro Gespräch verarbeitet:
| Modell | Tägliche Kosten | Monatliche Kosten |
|---|---|---|
| GPT-4.1-mini | ~2,40 $ | ~72 $ |
| GPT-4.1 | ~12,00 $ | ~360 $ |
| Claude Sonnet 4.6 | ~18,00 $ | ~540 $ |
| DeepSeek V3 | ~1,68 $ | ~50 $ |
Die Nutzung von GPT-4.1-mini für die meisten Gespräche und ein Upgrade auf Claude Sonnet 4.6 nur auf Nutzerwunsch hält die Kosten für die meisten Anwendungen unter 100 $/Monat.
Holen Sie sich Ihren API-Schlüssel: lemondata.cc bietet über einen Endpunkt 300+ Modelle. 1 $ Gratisguthaben zum Start.