Chatbot mit LLM-API: FastAPI-Backend mit Streaming-Frontend

Zuerst eine Skizze: drei Rollen, drei Aufgaben

Stell dir den Chatbot wie einen Lieferservice vor: Browser ist Kunde, Backend ist Empfang, LLM ist Küche. Kunde geht nicht direkt in die Küche. Alle Bestellungen laufen über den Empfang – deshalb darf der Schlüssel nicht im Frontend sein.

Ablauf: Nachricht senden, speichern, letzten Verlauf holen, System-Prompt hinzufügen, Anfrage stellen, Stream weiterleiten, Antwort speichern.

Warum Backend-Weiterleitung? Neben Sicherheit: Du brauchst Verlauf, Ratenlimit und Filter – alles Backend. Bei Modellswechsel musst du nur das Backend ändern, nicht das Frontend.

Backend: FastAPI mit SQLite

Installiere Abhängigkeiten und starte:

复制pip install fastapi uvicorn openai
export API_KEY=你的密钥
uvicorn server:app --reload --port 8000

Erstelle server.py. Achte auf vier Punkte:

复制# server.py
import os
import sqlite3
import uuid

from fastapi import FastAPI
from fastapi.responses import FileResponse, StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(base_url="https://api.apidamoxing.com/v1", api_key=os.environ["API_KEY"])

SYSTEM = {"role": "system", "content": "你是『小墨』，一个说话简洁、爱用比喻的聊天伙伴。回答控制在 200 字内。"}
KEEP = 20  # 每次只带最近 20 条消息

db = sqlite3.connect("chat.db", check_same_thread=False)
db.execute("create table if not exists msg(id integer primary key autoincrement, sid text, role text, content text)")

def load(sid, n=KEEP):
    rows = db.execute("select role, content from msg where sid=? order by id desc limit ?", (sid, n)).fetchall()
    return [{"role": r, "content": c} for r, c in reversed(rows)]

def save(sid, role, content):
    db.execute("insert into msg(sid, role, content) values (?,?,?)", (sid, role, content))
    db.commit()

class ChatIn(BaseModel):
    session_id: str
    message: str

@app.get("/")
def index():
    return FileResponse("index.html")

@app.post("/session")
def new_session():
    return {"session_id": uuid.uuid4().hex}

@app.post("/chat")
async def chat(body: ChatIn):
    save(body.session_id, "user", body.message[:4000])
    messages = [SYSTEM] + load(body.session_id)

    async def gen():
        parts = []
        try:
            stream = await client.chat.completions.create(
                model="uncensored", messages=messages,
                stream=True, max_tokens=800, temperature=0.8,
            )
            async for chunk in stream:
                if not chunk.choices:        # 末尾的 usage 块没有 choices
                    continue
                delta = chunk.choices[0].delta.content
                if delta:
                    parts.append(delta)
                    yield delta
        except Exception as e:
            yield f"\n[请求失败：{type(e).__name__}]"
        finally:
            if parts:
                save(body.session_id, "assistant", "".join(parts))

    return StreamingResponse(gen(), media_type="text/plain; charset=utf-8")

1. load() holt nur die letzten 20 Nachrichten, um das 100.000-Token-Kontextfenster nicht zu sprengen.
2. gen() ist ein asynchroner Generator, der jeden empfangenen Chunk sofort yieldet, sodass die Frontend-Anwendung Zeichen für Zeichen anzeigen kann.
3. Am Ende des Streams befindet sich ein Datenblock, der nur die Nutzungsinformationen enthält. Da er keine choices enthält, muss er auf Leere geprüft und übersprungen werden.
4. finally speichert die vollständige Antwort. Auch bei Fehlern geht nichts verloren.

Wenn du die Datenbank zunächst ignorieren willst, ersetze load und save durch Lese- und Schreibzugriffe auf ein Dictionary; der Rest bleibt gleich. Bei steigendem Traffic kannst du SQLite später durch jede beliebige Datenbank ersetzen, solange die Schnittstelle dieser beiden Funktionen erhalten bleibt. Hinweis: Das Beispiel nutzt das synchrone sqlite3, was schnell genug für Prototypen ist; für hohe Parallelität solltest du einen asynchronen Treiber wählen.

Frontend: Streams mit fetch lesen

Keine Bibliotheken nötig. resp.body.getReader() gibt einen Reader; loop über read(), decodiere Bytes mit TextDecoder und füge sie an. Speichere als index.html im selben Verzeichnis wie server.py:

复制<!doctype html>
<meta charset="utf-8">
<title>小墨</title>
<div id="gate">
  <label><input type="checkbox" id="adult"> 我已年满 18 周岁</label>
  <button id="enter">进入</button>
</div>
<div id="app" hidden>
  <div id="log" style="white-space:pre-wrap;min-height:300px"></div>
  <input id="box" placeholder="说点什么"> <button id="send">发送</button>
</div>
<script>
let sid = localStorage.getItem("sid");
const log = document.getElementById("log");

document.getElementById("enter").onclick = async () => {
  if (!document.getElementById("adult").checked) return;
  if (!sid) {
    const r = await fetch("/session", { method: "POST" });
    sid = (await r.json()).session_id;
    localStorage.setItem("sid", sid);
  }
  document.getElementById("gate").hidden = true;
  document.getElementById("app").hidden = false;
};

document.getElementById("send").onclick = async () => {
  const box = document.getElementById("box");
  const text = box.value.trim();
  if (!text) return;
  box.value = "";
  log.textContent += "\n你：" + text + "\n小墨：";
  const resp = await fetch("/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ session_id: sid, message: text }),
  });
  const reader = resp.body.getReader();
  const decoder = new TextDecoder("utf-8");
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    log.textContent += decoder.decode(value, { stream: true });
  }
  log.textContent += "\n";
};
</script>

Zwei Details: Füge { stream: true } hinzu, da Bytes eines Zeichens aufgeteilt werden können und sonst Müll entsteht. Deaktiviere den Button während der Anfrage, um Mehrfachklicks zu verhindern.

Starte die Anwendung, öffne den Browser auf Port 8000, kreuze die Altersbestätigung an und sende eine Nachricht. Wenn der Text als Block erscheint und nicht zeichenweise, puffert wahrscheinlich ein Proxy. Teste direkt am lokalen Port. Bei Nginx-Deployment muss die Response-Pufferung für diesen Pfad deaktiviert werden.

Chat-Verlauf: Warum im eigenen Backend speichern

Die Chat-Schnittstelle ist zustandslos, wie ein Empfangschef, der immer nur den gerade überreichten Zettel liest: Er weiß nur, was darauf steht. Die Kontext-Speicherung liegt in deiner Verantwortung. Es gibt drei Ansätze:

1. Minimal:Speichere nur im Arbeitsspeicher mit einem Dictionary. Nach einem Neustart sind die Daten verloren, ideal zum Testen.
2. Praktisch:Speichere wie im Beispiel in SQLite, jede Nachricht in einer Zeile, und frage die letzten N Nachrichten nach session_id ab.
3. Erweitert: Bei zu langem Kontext den älteren Inhalt vom Modell zusammenfassen lassen und als Zusammenfassung nach dem system-Prompt platzieren, während die neueren Nachrichten im Original erhalten bleiben.

Ein weiterer Vorteil, wenn du die Daten in deinem eigenen Backend speicherst, ist die vollständige Kontrolle über die Aufbewahrungsdauer. Es empfiehlt sich, einen Button „Chat löschen“ anzubieten, der die entsprechenden Session-Einträge tatsächlich löscht, und in deiner Datenschutzerklärung klar zu benennen, was du speicherst.

Die Trigger-Bedingung für die Zusammenfassung kann einfach sein: Wenn die geschätzte Anzahl an Tokens die Grenze von 10.000 überschreitet, gib die älften Nachrichten an das Modell, um sie in einen Absatz von maximal 300 Wörtern zusammenzufassen. Speichere das Ergebnis und lösche die Originalnachrichten. So bleibt der Kontext erhalten und die Anfragegröße bleibt stabil. Beachte: Die Zusammenfassung ist modellgeneriert und kann Details verlieren. Wichtige Einstellungen (z. B. Benutzername, Tabus) sollten im System-Prompt fixiert werden, nicht auf die Zusammenfassung verlassen.

Produkte für Erwachsene: Einstieg und Grenzen

Dieser Dienst richtet sich ausschließlich an erwachsene Nutzer ab 18 Jahren, und deine Anwendung sollte das ebenfalls so handhaben. Das Beispiel-Frontend enthält einen einfachen Bestätigungseingang. Strengere Produkte können einen vollständigeren Altersverifizierungsprozess einbauen. Hier sind einige Empfehlungen:

• Gib auf der Einstiegsseite klar die Altersbeschränkung an. Zeige den Chat-Bereich erst an, nachdem die Bestätigung erfolgt ist.
• Vermarkte das Produkt nicht an Schüler oder Minderjährige, und binde es nicht in Bereiche ein, die sich an sie richten.
• Sexuelle Inhalte mit Minderjährigen werden von der API unabhängig von ihrer Fiktionalität blockiert und mit dem Status 403 zurückgegeben. Dein Backend muss diesen Status erkennen und dem Nutzer eine freundliche Meldung anzeigen, anstatt den rohen Fehler anzuzeigen.

Du kannst Nutzern auch in den Einstellungen die Möglichkeit geben, einen Spitznamen und Stilpräferenzen festzulegen. Schreibe diese Einstellungen in den system-Prompt. Das macht die Konversation persönlicher, und das Modell muss nicht raten.

Absicherung und Erweiterungen vor dem Launch

• Eingaben begrenzen:Das Beispiel schneidet einzelne Nachrichten auf 4000 Zeichen zu. Passe dies bei Bedarf an.
• Ratenlimit: 300 Anfragen pro Minute und Key. Begrenze im Backend pro Benutzer oder IP.
• Fehlermeldungen:Fange Exceptions im Backend ab und gib kurze Hinweise an das Frontend zurück. Gib keine Stack-Traces nach außen weiter.
• Nutzungsüberwachung:Um Tokens zu zählen, lies usage bei nicht-streamenden Anfragen oder im letzten Block des Streams. Details dazu findest du im Abschnitt Tokens und Abrechnung.
• Frameworkwechsel:Wenn du das Frontend auf Vue oder React umstellst, ist die Logik zum Lesen des Streams identisch. Ein Wechsel des Backends zu Express erfordert lediglich, StreamingResponse durch die entsprechende Stream-Schreibweise zu ersetzen.

Für eine vollständige Parameterreferenz sieh dir API-Parameter im Detail an. Bei weiteren Fragen hilft dir Häufig gestellte Fragen weiter.

Führe vor dem Launch eine letzte Selbstkontrolle durch: Wird der Key nur in den Server-Umgebungsvariablen gespeichert? Gibt es Obergrenzen für einzelne Eingaben und die Verlaufslänge? Gibt es benutzerfreundliche Fehlermeldungen? Löscht „Chat löschen“ die Einträge wirklich? Ist die Altersbestätigung am Einstieg vorhanden? Wenn alle fünf Punkte erfüllt sind, kann dieses Prototyp-Modell an echte Nutzer übergeben werden.