Crea un chatbot con la API de LLM: backend FastAPI y frontend streaming

Primero un diagrama: tres roles, cada uno con su tarea

Piensa en el chatbot como una tienda de comida: el navegador es el cliente, tu backend es el mostrador y la API de LLM es la cocina. El cliente no entra a la cocina; todos los pedidos pasan por el mostrador, por eso la clave no debe estar en el frontend.

El flujo es: el navegador envía un mensaje, el backend lo guarda, extrae el historial reciente, lo une con el system, pide al endpoint y reenvía en tiempo real al navegador; al terminar, guarda la respuesta completa.

¿Por qué reenviar desde el backend en lugar de conectar directamente? Además de la seguridad de la clave, hay dos razones prácticas: primero, necesitas añadir historial, aplicar límites y filtrar, cosas que solo el backend puede hacer; segundo, si cambias de modelo o proveedor, solo modificas el backend y el frontend queda intacto.

Backend: FastAPI con SQLite

Instala las dependencias e inicia:

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

Luego escribe server.py. El código tiene cuatro puntos importantes:

复制# 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() solo toma las últimas 20 entradas para evitar que el historial crezca y supere la ventana de contexto de 100,000 tokens.
2. gen() es un generador asíncrono que emite un fragmento cada vez que recibe contenido, permitiendo que el frontend lo muestre palabra por palabra.
3. Al final del stream hay un bloque de datos que solo contiene el uso; como no tiene choices, se debe saltar si está vacío.
4. En finally se guarda la respuesta completa; incluso si hay un error, lo generado no se pierde.

Si prefieres no usar base de datos, reemplaza load y save por lectura y escritura en un diccionario; el resto del código no cambia. Si el tráfico aumenta, puedes cambiar SQLite por cualquier base de datos que conozcas, manteniendo la forma de estas dos funciones. Nota: el ejemplo usa sqlite3 sincrónico, que es rápido para prototipos; para alta concurrencia, considera un driver asíncrono.

Frontend: leer streaming con fetch

No necesitas librerías. resp.body.getReader() devuelve un lector. Itera read(), decodifica cada bloque con TextDecoder y añádelo a la página. Guarda esto como index.html junto a 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>

Dos detalles: al decodificar, añade { stream: true } porque los bytes de un carácter pueden dividirse en dos bloques; sin esto aparecerán caracteres extraños. Además, el botón de enviar debe deshabilitarse durante la petición para evitar clics múltiples; el ejemplo lo omite por brevedad, pero añádelo antes del lanzamiento.

Ejecuta y abre el navegador en el puerto 8000, acepta la verificación de edad y envía un mensaje. Si ves todo el texto de golpe y no letra por letra, probablemente un proxy está bufferizando la respuesta; prueba conectarte directamente al puerto local para descartar. Si usas Nginx, desactiva el buffer de respuesta para esta ruta.

Historial de conversaciones: por qué guardarlo en tu backend

La API de chat es “sin estado”, como un recepcionista que solo sabe lo que le escribes en un papel. Por tanto, “recordar el contexto” es tu responsabilidad. Hay tres niveles de implementación:

1. Mínimo:Almacena el historial en memoria usando un diccionario. Se pierde al reiniciar, ideal para depuración.
2. Común:Almacena en SQLite como en el ejemplo, una fila por mensaje, y consulta los últimos N mensajes por session_id.
3. Avanzado:Cuando el historial es muy largo, pide al modelo que resuma el contenido más antiguo en un párrafo y colócalo después del system. Los mensajes recientes se conservan en texto original.

Otro beneficio de almacenar en tu propio backend es que controlas completamente la retención. Se recomienda incluir un botón para "borrar conversación" que elimine realmente los registros de esa session_id, y especificar claramente en tu política de privacidad qué datos conservas.

Las condiciones de activación del método de resumen pueden ser simples: cuando el número estimado de tokens históricos supera diez mil, se le pide al modelo que resuma la mitad más antigua de los mensajes en un párrafo de no más de 300 palabras, se guarda en el almacenamiento y se eliminan los mensajes originales. Así se mantiene el contexto y se estabiliza el volumen de cada petición en un rango controlado. Ten en cuenta que el resumen también lo genera el modelo y puede omitir detalles; los ajustes importantes (como el nombre del usuario o los temas tabú) son más adecuados para fijarlos en el system prompt en lugar de depender de que el resumen los conserve.

Productos para adultos: entrada y límites

Este servicio está dirigido exclusivamente a usuarios mayores de 18 años; tu aplicación debe serlo también. El frontend de ejemplo incluye un mecanismo de confirmación mínimo; los productos más estrictos pueden implementar un flujo de verificación de edad más completo. Algunas recomendaciones prácticas:

• Indica claramente la restricción de edad en la página de entrada; no muestres la interfaz de chat hasta que se confirme.
• No promociones el producto a estudiantes o menores, ni lo integres en entornos dirigidos a ellos.
• El contenido sexual que involucre a menores se interceptará en el endpoint y devolverá un 403, independientemente de si es ficticio. Tu backend debe detectar este estado y mostrar un mensaje amigable al usuario, en lugar de mostrar el error original.

Desde el diseño del producto, también puedes permitir que los usuarios definan su propio nombre y preferencias de tono en la configuración. Al incluir esto en el system, la conversación será más personalizada y no será necesario que el modelo lo adivine.

Refuerzos y extensiones previos al lanzamiento

• Limita la entrada:El ejemplo ya limita cada mensaje a 4000 caracteres; ajústalo según sea necesario.
• Límite de frecuencia: 300 peticiones por minuto por clave, con un límite aplicado en el backend según el usuario o la IP.
• Muestra errores:El backend captura las excepciones y devuelve un mensaje breve al frontend; no expongas las trazas de pila.
• Monitorea el uso:Si deseas registrar los tokens, lee el campo usage en las peticiones no streaming, o lee el último bloque del stream. Consulta Tokens y facturación para más detalles.
• Cambia de framework:Al migrar el frontend a Vue o React, la lógica de lectura del stream es idéntica. Al cambiar el backend a Express, solo necesitas reemplazar StreamingResponse por la escritura de stream correspondiente.

Para una descripción más completa de los parámetros, consulta Detalles de los parámetros de la API; para más preguntas, revisa Preguntas frecuentes.

Realiza una última autoevaluación antes del lanzamiento: ¿la clave solo está en las variables de entorno del servidor? ¿Existen límites para la entrada individual y la longitud del historial? ¿Los errores muestran un mensaje amigable? ¿Borrar la conversación elimina realmente los registros? ¿La confirmación de edad está en la entrada? Si se cumplen estos cinco puntos, este prototipo está listo para que lo prueben usuarios reales.