대형 모델 API로 채팅봇 구축: FastAPI 백엔드 및 스트리밍 프론트엔드

먼저 그림을 그려 세 역할이 각자 맡은 일을 하도록 합니다

채팅 봇을 외식점으로 비유해 봅시다: 브라우저는 고객, 백엔드는 접수대, 대형 모델 인터페이스는 주방입니다. 고객은 주방에 직접 들어가지 않으며 모든 주문은 접수대를 통해 전달됩니다. 이것이 API 키를 프론트엔드에 두면 안 되는 이유입니다.

대화 흐름: 웹에서 메시지 전송 → 백엔드 저장 → 최근 기록 조회 → system 추가 → 인터페이스 요청 → 스트리밍 수신 및 웹 전달 → 완료 후 전체 응답 저장.

프론트엔드가 직접 연결하지 않고 백엔드가 전달하는 이유는 키 보안 외에도 두 가지 현실적인 이유가 있습니다. 첫째, 전달 시 기록을 추가하고 속도 제한 및 필터링을 해야 하는데 이는 백엔드에서만 가능합니다. 둘째, 향후 모델이나 서비스를 변경할 때 백엔드만 수정하면 되므로 프론트엔드는 변경할 필요가 없습니다.

백엔드: FastAPI 및 SQLite

먼저 의존성을 설치하고 실행합니다:

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

그런 다음 server.py를 작성합니다. 코드에는 다음 네 가지 포인트를 주목해야 합니다:

复制# 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()는 최근 20개만 가져와 대화 길이가 100,000 토큰 컨텍스트를 초과하지 않도록 합니다.
2. gen()는 비동기 생성기로, 작은 단위의 내용을 받을 때마다 yield하여 프론트엔드가 글자를 하나씩 표시할 수 있게 합니다.
3. 스트리밍 끝부분에는 사용량만 포함된 데이터 블록이 있으므로 choices가 없어 빈 것으로 처리하고 건너뜁니다.
4. finally 블록에서 전체 응답을 저장하므로 중간에 오류가 발생해도 생성된 부분이 손실되지 않습니다.

데이터베이스를 아직 사용하지 않으려면 load와 save를 딕셔너리 읽기/쓰기로 교체하면 됩니다. 트래픽이 증가하면 SQLite 대신 익숙한 데이터베이스로 교체해도 인터페이스는 동일하게 유지됩니다. 예제에서는 동기 sqlite3를 사용하지만, 동시 요청이 많은 환경에서는 비동기 드라이버를 고려하세요.

프론트엔드: fetch로 스트림 읽기

브라우저에는 라이브러리가 필요 없습니다. resp.body.getReader()로 리더를 얻고, read()를 반복 호출하여 바이트 블록을 가져온 후 TextDecoder로 디코딩해 페이지에 추가합니다. 아래 내용을 index.html로 저장하고 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>

두 가지 세부 사항: 디코딩 시 { stream: true }를 추가하세요. 한 글자의 바이트가 두 블록으로 나뉠 수 있으므로 추가하지 않으면 깨질 수 있습니다. 또한 입력란의 전송 로직은 요청 중 버튼 비활성화를 처리해야 합니다. 예제에서는 간결함을 위해 생략했지만, 출시 시에는 반드시 추가하세요.

실행 후 브라우저에서 8000 포트로 접속하고 연령 확인을 체크한 후 메시지를 보내세요. 글자가 한 번에 나타나지 않고 하나씩 출력되지 않는다면 중간에 응답을 버퍼링하는 프록시가 있을 수 있으므로 로컬 포트에 직접 연결하여 확인하세요. Nginx 뒤에 배포할 때는 해당 경로에 대한 응답 버퍼링을 비활성화해야 합니다.

세션 기록: 왜 내 백엔드에 저장해야 하나요

채팅 API는 '무상태'입니다. 전달된 종이에 적힌 내용만 아는 안내원처럼 생각하세요. 따라서 '컨텍스트 창 기억'은 당신의 책임이며, 구체적인 방법은 세 가지 수준으로 나뉩니다:

1. 간단함:메모리 내에서 사전으로만 저장합니다. 재시작 시 데이터가 손실되므로 디버깅에 적합합니다.
2. 일반적:예시와 같이 SQLite를 사용하여 저장합니다. 각 메시지 행을 session_id 기준으로 조회하여 최근 N개의 메시지를 가져옵니다.
3. 고급:역사 기록이 너무 길어지면, 이전 내용을 모델에 요약시켜 system 프롬프트 뒤에 배치합니다. 최근 메시지는 원문 그대로 유지합니다.

자사 백엔드에 저장하면 보존 기간을 완전히 통제할 수 있다는 또 다른 이점이 있습니다. '대화 지우기' 버튼을 제공하여 해당 세션의 기록을 실제로 삭제하고, 개인정보 처리방침에 무엇을 저장하는지 명확히 명시하는 것이 좋습니다.

요약 방식의 조건은 간단합니다: 추산된 토큰이 1만 개를 넘으면 가장 오래된 메시지 절반을 모델에 요약시켜 300자 이내로 만들고 저장소에 저장한 뒤 원본을 삭제합니다. 이렇게 하면 맥락은 유지되지만 매 요청의 요청량이 통제 가능한 범위로 안정화됩니다. 요약은 모델이 생성하므로 세부 정보가 누락될 수 있으며, 사용자 이름이나 금지된 주제 같은 중요한 설정은 요약에 의존하기보다 system에 고정하는 것이 좋습니다.

성인 대상 제품: 진입로와 경계

이 서비스는 18세 이상 성인만 이용 가능합니다. 애플리케이션도 마찬가지여야 합니다. 예제 프론트엔드에는 가장 간단한 확인 진입구가 있습니다. 더 엄격한 제품이라면 더 완전한 연령 확인 절차를 추가할 수 있습니다. 몇 가지 실전 팁:

• 진입 페이지에 연령 제한을 명확히 안내하고, 확인 전까지 채팅 인터페이스를 표시하지 않습니다.
• 제품을 학생이나 미성년자 층에 홍보하지 말고, 그들을 위한 사용 시나리오에 포함하지 마십시오.
• 미성년자가 관련된 성 콘텐츠는 실제 여부(가상 여부)와 관계없이 API에서 차단되며 403 상태 코드를 반환합니다. 백엔드는 이 상태를 식별하고 원본 오류를 표시하기보다 사용자에게 친절한 안내 메시지를 표시해야 합니다.

제품 설계 측면에서 사용자가 설정에서 별명과 말투 선호도를 직접 설정할 수 있도록 할 수 있습니다. 이를 system 프롬프트에 반영하면 대화가 더 개인화되지만, 모델이 추측할 필요가 없어집니다.

출시 전 보안 강화 및 확장

• 입력 제한:예시에서는 단일 메시지 길이를 4,000자로 제한했습니다. 필요에 따라 조정하십시오.
• 빈도 제한:키당 분당 300회 요청으로 제한합니다. 백엔드에서 사용자 또는 IP를 기준으로 먼저 속도 제한을 적용하십시오.
• 오류 표시:백엔드에서 예외를 캐치하여 프론트엔드에 짧은 안내 메시지를 반환하고, 스택 트레이스를 노출하지 마십시오.
• 사용량 모니터링:토큰 사용량을 집계하려면 스트리밍이 아닌 요청에서 usage 필드를 읽거나, 스트림의 마지막 청크에서 읽으십시오. 자세한 내용은 토큰과 요금을 참조하십시오.
• 프레임워크 변경:프론트엔드를 Vue나 React로 변경해도 스트림 읽기 로직은 동일합니다. 백엔드를 Express로 변경하면 StreamingResponse를 해당 프레임워크의 스트림 작성 방식으로 교체하기만 하면 됩니다.

더 완전한 파라미터 설명이 필요하면 API 파라미터 상세 설명을 참조하십시오. 추가 질문은 자주 묻는 질문을 확인하십시오.

출시 전 마지막으로 자체 점검을 수행하십시오: 키가 백엔드 환경 변수에만 있는지 확인하십시오. 단일 입력 및 역사 길이에 상한이 있는지 확인하십시오. 오류에 친절한 안내 메시지가 있는지 확인하십시오. '대화 지우기'가 실제로 기록을 삭제하는지 확인하십시오. 진입로에 연령 확인이 있는지 확인하십시오. 이 5가지 항목이 모두 충족되면 이 프로토타입을 실제 사용자에게 테스트용으로 제공할 수 있습니다.