Crie um chatbot com API de LLM: backend FastAPI e frontend com streaming

Primeiro, um diagrama: três papéis distintos

Pense no chatbot como uma lanchonete: o navegador é o cliente, seu backend é o balcão e a API de LLM é a cozinha. O cliente não entra na cozinha; o balcão gerencia tudo. É por isso que a chave não deve ficar no frontend.

Fluxo: o usuário envia uma mensagem, o backend salva, recupera o histórico recente, adiciona o system prompt, chama a API, encaminha o streaming e salva a resposta final.

Por que usar um backend intermediário? Além da segurança da chave, você precisa adicionar histórico, limitar requisições e filtrar dados no backend. Além disso, ao trocar de modelo ou provedor, basta alterar o backend; o frontend permanece intacto.

Backend: FastAPI com SQLite

Instale as dependências e inicie o servidor:

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

Em seguida, crie o arquivo server.py. Observe quatro pontos importantes no código:

复制# 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() retorna apenas as 20 mensagens mais recentes para evitar estourar a janela de contexto de 100.000 tokens.
2. gen() é um gerador assíncrono que faz yield de cada fragmento recebido, permitindo exibição caractere a caractere no frontend.
3. No final do streaming, há um bloco de dados que contém apenas o uso. Como ele não possui choices, você deve verificar se está vazio e pular.
4. No bloco finally, salve a resposta completa. Assim, mesmo em caso de erro, o texto gerado não será perdido.

Se quiser evitar o banco de dados inicialmente, substitua load e save por operações em dicionário. Para escalar, substitua o SQLite por qualquer banco de sua preferência, mantendo a interface das funções. O exemplo usa sqlite3 síncrono, que é rápido o suficiente para protótipos; use drivers assíncronos para alta concorrência.

Frontend: leitura de streaming com fetch

O navegador não precisa de bibliotecas. Use resp.body.getReader() para obter um leitor, faça loop em read() e decodifique os bytes com TextDecoder para exibir na página. Salve o código abaixo como index.html na mesma pasta do 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>

Dois detalhes: inclua { stream: true } no decoder para evitar caracteres corrompidos (bytes de caracteres multibyte podem ser divididos). Além disso, desabilite o botão de envio durante a requisição para evitar cliques duplos; o exemplo omite isso para ser curto, mas implemente na versão final.

Acesse localhost:8000, confirme a idade e envie uma mensagem. Se o texto aparecer de uma vez e não em streaming, pode haver buffering de proxy. Teste a conexão direta localmente. Ao implantar no Nginx, desative o buffering de resposta para esse endpoint.

Histórico de conversas: por que armazenar localmente

A interface de chat é "sem estado", como um atendente que só sabe o que está no bilhete que você entrega a cada interação: ele só sabe o que está no bilhete. Portanto, "lembrar o contexto" é sua responsabilidade, e existem três níveis para isso:

1. Mínimo:armazena apenas em memória usando um dicionário. Os dados são perdidos na reinicialização, sendo adequado para depuração.
2. Comum:armazena em SQLite conforme o exemplo, com cada mensagem em uma linha, e consulta as últimas N mensagens por session_id.
3. Avançado:quando o histórico está muito longo, peça ao modelo para resumir as mensagens mais antigas em um parágrafo. Coloque esse resumo após o system; as mensagens recentes permanecem no texto original.

Outro benefício de armazenar no seu próprio backend é que você tem controle total sobre o período de retenção. Recomenda-se fornecer um botão "Limpar conversa" para realmente excluir os registros da sessão correspondente e deixar claro na sua política de privacidade o que você armazena.

A condição de acionamento do método de resumo pode ser simples: quando o número estimado de tokens do histórico ultrapassar 10.000, envie a metade mais antiga das mensagens ao modelo para gerar um resumo de no máximo 300 caracteres, salve-o no armazenamento e exclua as mensagens originais. Isso preserva o contexto e mantém o volume de cada requisição estável e controlável. Observe que o próprio resumo é gerado pelo modelo e pode omitir detalhes; configurações importantes (como o apelido do usuário ou tópicos proibidos) devem ser fixadas no system, em vez de depender do resumo para serem mantidas.

Produtos para adultos: entrada e limites

Este serviço destina-se apenas a usuários adultos com 18 anos ou mais, e seu aplicativo deve seguir o mesmo padrão. O frontend de exemplo inclui uma entrada de confirmação simples; produtos mais rigorosos podem implementar um fluxo completo de verificação de idade. Algumas recomendações práticas:

• Informe claramente a restrição de idade na página de entrada; não exiba a interface de chat até que a confirmação seja feita.
• Não promova o produto para estudantes ou menores de idade, nem o inclua em cenários voltados a esse público.
• O conteúdo sexual envolvendo menores de idade será bloqueado pela API e retornará 403, independentemente de ser fictício ou não. Seu backend deve identificar esse status e exibir uma mensagem amigável ao usuário, em vez de mostrar o erro bruto.

Do ponto de vista do design do produto, você também pode permitir que o usuário defina um apelido e preferências de tom nas configurações. Basta incluir essas informações no system para personalizar a conversa sem exigir que o modelo adivinhe.

Reforço e extensões antes do lançamento

• Limite de entrada:o exemplo já limita cada mensagem a 4.000 caracteres; ajuste conforme necessário.
• Limite de frequência:limite a 300 requisições por minuto por chave. Implemente o rate limit no backend antes de encaminhar, por usuário ou IP.
• Exibição de erros:o backend deve capturar exceções e retornar uma mensagem resumida ao frontend, sem expor o stack trace.
• Monitoramento de uso: Para contabilizar tokens, você pode ler o campo usage em requisições não-streaming ou ler o último bloco do streaming. Para mais detalhes, consulteTokens e cobrança.
• Troca de framework:ao migrar o frontend para Vue ou React, a lógica de leitura do streaming permanece idêntica. Ao trocar o backend para Express, basta substituir StreamingResponse pela escrita de streaming correspondente.

Para uma descrição completa dos parâmetros, consulteDetalhes dos parâmetros da API; para mais dúvidas, vejaPerguntas frequentes.

Antes do lançamento, faça uma autoverificação final: a chave está armazenada apenas nas variáveis de ambiente do servidor? As entradas individuais e o histórico têm limites? Os erros exibem mensagens amigáveis? O botão "Limpar conversa" realmente exclui os registros? A confirmação de idade está presente na entrada? Se todos os cinco critérios forem atendidos, este protótipo está pronto para ser testado por usuários reais.