使用 Next.js、FastAPI、Credits 和 SSE 构建 AI 聊天 SaaS 平台
了解包含 FastAPI 网关、信用账本、Redis 速率限制及 SSE 流式传输的完整栈 AI 聊天架构，以及需要首先弥补的缺陷。
来源: Build a Production-Ready AI SaaS with React 19 + FastAPI: The Complete Architecture (2026)
围绕模型API构建聊天界面相对简单；但产品还必须确保密钥不会泄露到客户端、合理计量使用量、限制滥用行为、清晰地管理流式令牌，并实现可复现的部署。本指南将使用React 19的Next.js、FastAPI、带有pgvector的PostgreSQL以及Redis来构建这些功能。可将代码视为草图：每个部分都会指出首先需要加强的安全措施。
后端是控制枢纽
浏览器永远不会直接调用模型提供方。您的API负责身份验证、检查使用限制和额度、调用模型，并记录相关成本。
┌─────────────────────────────────────────────────────────────┐
│ React 19 App │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Server │ │ Client │ │ Chat Interface │ │
│ │ Components │ │ Components │ │ (Streaming) │ │
│ │ (Initial │ │ (Interactivity│ │ │ │
│ │ data fetch)│ │ + State) │ │ │ │
│ └──────┬───────┘ └──────┬───────┘ └────────┬─────────┘ │
│ │ │ │ │
│ └─────────────────┴────────────────────┘ │
│ │ │
│ TanStack Query │
│ (Auth + Sync) │
└───────────────────────────┬─────────────────────────────────┘
│ HTTPS / JSON
┌───────────────────────────┴─────────────────────────────────┐
│ FastAPI Backend │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Auth │ │ Rate │ │ OpenAI │ │
│ │ Middleware │ │ Limiter │ │ Service │ │
│ │ (JWT + │ │ (Redis) │ │ (Structured │ │
│ │ Credits) │ │ │ │ Output) │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Background │ │ PostgreSQL │ │ pgvector │ │
│ │ Worker │ │ (Users + │ │ (Embeddings) │ │
│ │ (Celery) │ │ Conversations│ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────────────┘
服务器组件无需额外JavaScript即可加载历史数据，异步的FastAPI能在不阻塞线程的情况下处理耗时的模型调用，pgvector为数据检索提供了便利，Redis用于存储速率限制计数器和缓存过的嵌入向量，而工作进程则可将PDF解析和嵌入任务从请求处理流程中分离出来。
数据模型：记录在账本上
该架构定义了用户、对话、消息以及信用账本。
# backend/models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text, Numeric, Index
from sqlalchemy.dialects.postgresql import UUID, JSONB, VECTOR
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
email = Column(String, unique=True, nullable=False)
hashed_password = Column(String, nullable=False)
credits_balance = Column(Numeric(10, 2), default=0.00)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
conversations = relationship("Conversation", back_populates="user")
__table_args__ = (
Index('ix_users_email', 'email'),
)
class Conversation(Base):
__tablename__ = "conversations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
title = Column(String, nullable=True)
model = Column(String, default="gpt-4o")
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
user = relationship("User", back_populates="conversations")
messages = relationship("Message", back_populates="conversation", order_by="Message.created_at")
class Message(Base):
__tablename__ = "messages"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversations.id"), nullable=False)
role = Column(String, nullable=False) # 'user', 'assistant', 'system'
content = Column(Text, nullable=False)
token_count = Column(Integer, default=0)
metadata = Column(JSONB, default={}) # Store model, latency, finish_reason
created_at = Column(DateTime, default=datetime.utcnow)
conversation = relationship("Conversation", back_populates="messages")
class CreditLedger(Base):
__tablename__ = "credit_ledger"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
amount = Column(Numeric(10, 2), nullable=False) # Negative for usage
description = Column(String, nullable=False)
reference_id = Column(String, nullable=True) # conversation_id or message_id
created_at = Column(DateTime, default=datetime.utcnow)
不会直接修改余额，而是将每次购买或使用记录追加进去，余额即为这些记录的总和，从而形成审计追踪且不会丢失任何更新。在使用之前需要修复几个问题：Message.metadata与SQLAlchemy声明式基类预留的属性名冲突，需为其重命名（例如改为meta）；缺少诸如Boolean和declarative_base之类的导入；此外pgvector的列类型来自pgvector包，而非SQLAlchemy的PostgreSQL方言。
FastAPI服务与路由
传统的架构会将认证、AI逻辑、路由器以及中间件分开处理：
backend/
├── main.py
├── config.py
├── database.py
├── models.py
├── schemas.py
├── auth/
│ ├── utils.py
│ ├── dependencies.py
│ └── router.py
├── ai/
│ ├── service.py
│ └── schemas.py
├── api/
│ ├── conversations.py
│ └── credits.py
├── middleware/
│ └── rate_limit.py
└── worker.py
可计量使用量的流式服务
该服务允许每位用户每分钟通过 Redis 发送20次请求，先检查余额，再流式传输服务器发送的事件，最后将助手回复信息与负向账目记录一起提交。
# backend/ai/service.py
from openai import AsyncOpenAI
from typing import AsyncGenerator
import json
from sqlalchemy.ext.asyncio import AsyncSession
from models import Message, CreditLedger
from schemas import ChatCompletionRequest, ChatStreamChunk
import redis.asyncio as redis
client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
redis_client = redis.from_url(settings.REDIS_URL)
class AIService:
def __init__(self, db: AsyncSession):
self.db = db
async def stream_chat(
self,
user_id: str,
conversation_id: str,
messages: list[dict],
model: str = "gpt-4o"
) -> AsyncGenerator[str, None]:
"""
Streams chat completion and deducts credits.
Yields SSE-formatted strings.
"""
# 1. Check rate limit
key = f"ratelimit:chat:{user_id}"
current = await redis_client.incr(key)
if current == 1:
await redis_client.expire(key, 60)
if current > 20: # 20 requests per minute
yield f"data: {json.dumps({'error': 'Rate limit exceeded'})}\n\n"
return
# 2. Check credits (simplified; use ledger sum in production)
balance = await self.get_credit_balance(user_id)
if balance < 0.01:
yield f"data: {json.dumps({'error': 'Insufficient credits'})}\n\n"
return
# 3. Stream from OpenAI
try:
stream = await client.chat.completions.create(
model=model,
messages=messages,
stream=True,
max_tokens=4096,
)
full_content = ""
total_tokens = 0
async for chunk in stream:
delta = chunk.choices[0].delta.content or ""
full_content += delta
total_tokens += 1 # Approximate; use tiktoken for accuracy
yield f"data: {json.dumps({'chunk': delta, 'done': False})}\n\n"
# 4. Persist assistant message
assistant_msg = Message(
conversation_id=conversation_id,
role="assistant",
content=full_content,
token_count=total_tokens,
metadata={"model": model, "finish_reason": "stop"}
)
self.db.add(assistant_msg)
# 5. Deduct credits (approximate $0.002 per 1K tokens for GPT-4o)
cost = (total_tokens / 1000) * 0.002
ledger_entry = CreditLedger(
user_id=user_id,
amount=-cost,
description=f"Chat completion: {conversation_id}",
reference_id=str(conversation_id)
)
self.db.add(ledger_entry)
await self.db.commit()
yield f"data: {json.dumps({'done': True, 'tokens': total_tokens})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
async def get_credit_balance(self, user_id: str) -> float:
# In production, use a SQL SUM query
result = await self.db.execute(
select(func.sum(CreditLedger.amount)).where(CreditLedger.user_id == user_id)
)
return result.scalar() or 0.00
注意事项：每段数据一个令牌仅是近似值，应使用提供商报告的用量或分词器。硬编码的价格仅为示例，并非实时价格，需从配置中加载实际价格。此外，在流式传输前检查余额、之后再扣款并非原子操作；应提前预留额度或锁定用户记录。
消息端点
路由器会验证用户所有权（返回404而非403，以避免泄露ID），保存用户的消息，然后返回一个 StreamingResponse：
# backend/api/conversations.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from auth.dependencies import get_current_user
from database import get_db
from ai.service import AIService
from models import Conversation, Message
from schemas import CreateMessageRequest
router = APIRouter(prefix="/conversations", tags=["conversations"])
@router.post("/{conversation_id}/messages")
async def create_message(
conversation_id: str,
request: CreateMessageRequest,
current_user = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
):
# Verify ownership
conversation = await db.get(Conversation, conversation_id)
if not conversation or conversation.user_id != current_user.id:
raise HTTPException(status_code=404, detail="Conversation not found")
# Persist user message
user_msg = Message(
conversation_id=conversation_id,
role="user",
content=request.content
)
db.add(user_msg)
await db.commit()
# Build message history
messages = [
{"role": m.role, "content": m.content}
for m in conversation.messages
]
messages.append({"role": "user", "content": request.content})
# Stream response
ai_service = AIService(db)
return StreamingResponse(
ai_service.stream_chat(
user_id=str(current_user.id),
conversation_id=conversation_id,
messages=messages,
model=conversation.model
),
media_type="text/event-stream"
)
使用异步 SQLAlchemy 时，若延迟访问 conversation.messages 会引发错误，因此需提前加载该关联数据。
全局速率限制
另一个更严格的限制为每分钟100次请求，该限制通过中间件实现：
# backend/middleware/rate_limit.py
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis
redis_client = redis.from_url("redis://localhost:6379")
class RateLimitMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Skip rate limiting for auth routes
if request.url.path.startswith("/auth"):
return await call_next(request)
# Identify user (from JWT cookie or header)
user_id = request.state.user_id if hasattr(request.state, "user_id") else "anonymous"
key = f"ratelimit:global:{user_id}"
current = await redis_client.incr(key)
if current == 1:
await redis_client.expire(key, 60)
if current > 100: # 100 requests per minute globally
raise HTTPException(status_code=429, detail="Too many requests")
response = await call_next(request)
response.headers["X-RateLimit-Remaining"] = str(100 - current)
return response
BaseHTTPMiddleware内部抛出的HTTPException不会被FastAPI的处理程序转换为标准的429状态码，因此应返回状态为429的JSONResponse。固定时间窗口也无法避免突发流量。
前端部分
App Router架构将服务器组件页面与用于实时聊天的客户端组件配对：
src/
├── app/
│ ├── layout.tsx
│ ├── page.tsx
│ ├── (dashboard)/
│ │ ├── layout.tsx
│ │ ├── conversations/
│ │ │ ├── page.tsx # Server Component: list conversations
│ │ │ └── [id]/
│ │ │ ├── page.tsx # Server Component: load initial messages
│ │ │ └── ChatClient.tsx # Client Component: streaming + input
│ │ └── settings/
│ │ └── page.tsx
│ └── api/ # Route handlers if needed
├── components/
│ ├── ui/ # shadcn/ui components
│ └── chat/
│ ├── MessageList.tsx
│ ├── MessageItem.tsx
│ └── StreamingText.tsx
├── lib/
│ ├── api.ts # Axios instance with interceptors
│ └── auth.ts
├── hooks/
│ ├── useChatStream.ts # Custom hook for SSE
│ └── useConversations.ts
└── providers.tsx
在服务器端加载历史记录
页面会读取Cookie并同时获取对话内容和消息，这样首次渲染时就能显示聊天内容：
// app/(dashboard)/conversations/[id]/page.tsx
import { notFound } from "next/navigation";
import { cookies } from "next/headers";
import { ChatClient } from "./ChatClient";
async function getConversation(id: string, token: string) {
const res = await fetch(`${process.env.API_URL}/conversations/${id}`, {
headers: { Authorization: `Bearer ${token}` },
next: { revalidate: 0 } // Dynamic data
});
if (!res.ok) return null;
return res.json();
}async function getMessages(id: string, token: string) {
const res = await fetch(`${process.env.API_URL}/conversations/${id}/messages`, {
headers: { Authorization: `Bearer ${token}` },
});
if (!res.ok) return [];
return res.json();
}export default async function ConversationPage({
params
}: {
params: { id: string }
}) {
const cookieStore = cookies();
const token = cookieStore.get("access_token")?.value;
if (!token) return notFound();
// Parallel data fetching on the server
const [conversation, messages] = await Promise.all([
getConversation(params.id, token),
getMessages(params.id, token)
]);
if (!conversation) return notFound();
return (
<div className="flex h-[calc(100vh-4rem)] flex-col">
<header className="border-b px-6 py-4">
<h1 className="text-lg font-semibold">{conversation.title || "New Chat"}</h1>
</header>
{/* Client component handles interactivity and streaming */}
<ChatClient
conversationId={params.id}
initialMessages={messages}
/>
</div>
);
}
在Next.js 15及更高版本中，cookies()和params都是异步操作，需要使用await来等待其执行结果。
将数据流式传输到用户界面
客户端乐观地添加用户消息和一个空的助手占位符，然后发送POST请求，并从响应流中填充该占位符：
// app/(dashboard)/conversations/[id]/ChatClient.tsx
"use client";import { useState, useRef, useCallback } from "react";
import { useChatStream } from "@/hooks/useChatStream";
import { MessageList } from "@/components/chat/MessageList";
import { ChatInput } from "@/components/chat/ChatInput";
interface Message {
id: string;
role: "user" | "assistant";
content: string;
isStreaming?: boolean;
}
export function ChatClient({
conversationId,
initialMessages
}: {
conversationId: string;
initialMessages: Message[];
}) {
const [messages, setMessages] = useState<Message[]>(initialMessages);
const [input, setInput] = useState("");
const [isLoading, setIsLoading] = useState(false);
const handleSubmit = useCallback(async (content: string) => {
if (!content.trim() || isLoading) return;
// 1. Optimistic UI: Add user message immediately
const userMessage: Message = {
id: `temp-${Date.now()}`,
role: "user",
content
};
const assistantMessage: Message = {
id: `temp-assistant-${Date.now()}`,
role: "assistant",
content: "",
isStreaming: true
};
setMessages(prev => [...prev, userMessage, assistantMessage]);
setInput("");
setIsLoading(true);
// 2. Open SSE connection
const response = await fetch(`/api/conversations/${conversationId}/messages`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ content })
});
if (!response.body) {
setIsLoading(false);
return;
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let assistantContent = "";
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const lines = chunk.split("\n");
for (const line of lines) {
if (line.startsWith("data: ")) {
const data = JSON.parse(line.slice(6));
if (data.error) {
throw new Error(data.error);
}
if (data.chunk) {
assistantContent += data.chunk;
setMessages(prev =>
prev.map(msg =>
msg.id === assistantMessage.id
? { ...msg, content: assistantContent }
: msg
)
);
}
if (data.done) {
setMessages(prev =>
prev.map(msg =>
msg.id === assistantMessage.id
? { ...msg, isStreaming: false, tokenCount: data.tokens }
: msg
)
);
}
}
}
}
} catch (error) {
setMessages(prev =>
prev.map(msg =>
msg.id === assistantMessage.id
? { ...msg, content: "Error: Failed to generate response", isStreaming: false }
: msg
)
);
} finally {
setIsLoading(false);
reader.releaseLock();
}
}, [conversationId, isLoading]);
return (
<div className="flex flex-1 flex-col">
<MessageList messages={messages} />
<ChatInput
value={input}
onChange={setInput}
onSubmit={handleSubmit}
disabled={isLoading}
/>
</div>
);
}
尽管有相关注释，此处实际使用的是带有流读取器的fetch，而非无法发送POST请求体的EventSource。由于数据块可能会拆分内容，需对部分行进行缓冲处理，并添加AbortController以便在组件卸载时取消请求。此外，相对路径/api/...也假定存在能够将请求代理到FastAPI的路由处理程序或重写规则。
其他场景可使用Axios客户端
对于httpOnly Cookie，客户端会发送凭证、添加追踪ID，并在遇到401响应时刷新后重试一次：
// lib/api.ts
import axios from "axios";
export const api = axios.create({
baseURL: process.env.NEXT_PUBLIC_API_URL,
withCredentials: true, // Send cookies
});
// Request interceptor: add auth token
api.interceptors.request.use((config) => {
// Token is in httpOnly cookie, sent automatically
// Add request ID for tracing
config.headers["X-Request-ID"] = crypto.randomUUID();
return config;
});
// Response interceptor: handle 401 and token refresh
api.interceptors.response.use(
(response) => response,
async (error) => {
const originalRequest = error.config;
if (error.response?.status === 401 && !originalRequest._retry) {
originalRequest._retry = true;
try {
await axios.post(
`${process.env.NEXT_PUBLIC_API_URL}/auth/refresh`,
{},
{ withCredentials: true }
);
return api(originalRequest);
} catch (refreshError) {
window.location.href = "/login";
return Promise.reject(refreshError);
}
}
return Promise.reject(error);
}
);
每个并发的401错误都会触发刷新操作；如需无竞争问题的解决方案，请参阅单次刷新修复方案。
将信用额度作为依赖项进行校验
FastAPI的依赖组件会在执行任何模型调用之前，先汇总账本信息，若余额过低则返回402“需要支付”错误以拒绝请求：
# backend/api/deps.py
from fastapi import Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from auth.dependencies import get_current_user
from database import get_db
from models import CreditLedger
from sqlalchemy import func, select
async def require_credits(
min_credits: float = 0.01,
current_user = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
):
result = await db.execute(
select(func.sum(CreditLedger.amount)).where(
CreditLedger.user_id == current_user.id
)
)
balance = result.scalar() or 0.00
if balance < min_credits:
raise HTTPException(
status_code=402, # Payment Required
detail=f"Insufficient credits. Balance: ${balance:.4f}"
)
return current_user
# Usage
@router.post("/conversations/{id}/messages")
async def create_message(
...,
user = Depends(require_credits) # Enforces payment before AI call
):
...
它也存在上述非原子操作的问题。
在不调用模型的情况下进行测试
MSW 2.x会返回模拟的事件流，而Testing Library测试则会检查乐观锁相关的文本信息：
// tests/mocks/openai.ts
import { http, HttpResponse } from "msw";
export const openaiHandlers = [
http.post("https://api.openai.com/v1/chat/completions", async () => {
const encoder = new TextEncoder();
const stream = new ReadableStream({
start(controller) {
const chunks = [
'data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n',
'data: {"choices":[{"delta":{"content":" world"}}]}\n\n',
'data: {"choices":[{"delta":{}}],"done":true}\n\n',
];
chunks.forEach(chunk => {
controller.enqueue(encoder.encode(chunk));
});
controller.close();
}
});
return new HttpResponse(stream, {
headers: {
"Content-Type": "text/event-stream",
},
});
}),
];
// tests/ChatClient.test.tsx
import { render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { ChatClient } from "@/app/(dashboard)/conversations/[id]/ChatClient";
it("streams assistant response", async () => {
const user = userEvent.setup();
render(
<ChatClient
conversationId="test-id"
initialMessages={[]}
/>
);
await user.type(screen.getByPlaceholderText("Ask anything..."), "Hello");
await user.click(screen.getByRole("button", { name: /send/i }));
// Optimistic user message appears immediately
expect(screen.getByText("Hello")).toBeInTheDocument();
// Streaming response appears
await waitFor(() => {
expect(screen.getByText("Hello world")).toBeInTheDocument();
});
});
处理程序会模拟提供者URL，但组件实际上会调用你的API，因此应直接模拟端点。后端测试则会拦截客户端请求并验证是否存在一条账本记录：
# tests/test_chat.py
import pytest
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_chat_deducts_credits(client, auth_headers, db_session):
with patch("ai.service.client.chat.completions.create") as mock_create:
mock_create.return_value = AsyncMock()
mock_create.return_value.__aiter__.return_value = [
MockChunk(delta=MockContent("Hi"))
]
response = await client.post(
"/conversations/test-id/messages",
json={"content": "Hello"},
headers=auth_headers
)
assert response.status_code == 200
# Verify credit deduction
ledger = await db_session.execute(
select(CreditLedger).where(CreditLedger.user_id == test_user_id)
)
assert len(ledger.scalars().all()) == 1
使用容器进行部署
多阶段 Dockerfile 以独立模式构建 Next.js，并以非根用户身份运行它：
# frontend/Dockerfile
FROM node:20-alpine AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build
FROM node:20-alpine AS runner
WORKDIR /app
ENV NODE_ENV=production
RUN addgroup --system --gid 1001 nodejs
RUN adduser --system --uid 1001 nextjs
COPY --from=builder /app/public ./public
COPY --from=builder --chown=nextjs:nodejs /app/.next/standalone ./
COPY --from=builder --chown=nextjs:nodejs /app/.next/static ./.next/static
USER nextjs
EXPOSE 3000
ENV PORT=3000
CMD ["node", "server.js"]
Compose 同时启动带有健康检查功能的 pgvector Postgres、Redis、API、前端应用以及 nginx：
version: "3.8"
services:
postgres:
image: pgvector/pgvector:pg16
environment:
POSTGRES_USER: app
POSTGRES_PASSWORD: ${DB_PASSWORD}
POSTGRES_DB: saas_db
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U app"]
interval: 5s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
backend:
build: ./backend
environment:
DATABASE_URL: postgresql+asyncpg://app:${DB_PASSWORD}@postgres/saas_db
REDIS_URL: redis://redis:6379/0
OPENAI_API_KEY: ${OPENAI_API_KEY}
SECRET_KEY: ${SECRET_KEY}
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_started
command: uvicorn main:app --host 0.0.0.0 --port 8000
frontend:
build: ./frontend
environment:
NEXT_PUBLIC_API_URL: http://backend:8000
depends_on:
- backend
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
depends_on:
- frontend
- backend
volumes:
postgres_data:
redis_data:
NEXT_PUBLIC_ 变量的值会在构建时固定下来供浏览器使用，因此浏览器无法解析 backend:8000；应使用公共的 nginx 路由。
关键要点
- 将密钥、定价信息及使用限制保存在服务器端；模型则是 API 后面的依赖项。
- 将使用情况记录在只读账本中，并确保检查与扣款操作是原子性的。
- 使用显式取消机制和行缓冲进行流处理。
- 在测试中模拟自己的 API 接口，而非供应商的接口。