Entwicklung eines AI-Chat-SaaS mit Next.js, FastAPI, Credits und SSE
Erlernen Sie eine vollständige AI-Chat-Architektur mit einem FastAPI-Gatekeeper, einer Kreditbuchhaltung, Redis-Rate-Limits sowie SSE-Streaming – zusammen mit den ersten Schritten zur Beseitigung der offenen Lücken.
Quellen: Build a Production-Ready AI SaaS with React 19 + FastAPI: The Complete Architecture (2026)
Eine Chat-Oberfläche rund um eine Modell-API ist einfach zu erstellen; ein Produkt muss jedoch auch die Schlüssel vom Client fernhalten, die Nutzung messen, Missbrauch einschränken, Stream-Tokens sauber verwalten und reproduzierbar bereitstellen. In dieser Anleitung werden diese Komponenten mit Next.js auf React 19, FastAPI, PostgreSQL mit pgvector sowie Redis entwickelt. Betrachten Sie den Code als Skizze: Jeder Abschnitt weist darauf hin, was zuerst gesichert werden sollte.
Der Backend ist der Wächter
Der Browser ruft niemals den Modellanbieter auf. Ihre API authentifiziert sich, überprüft Limits und Kreditpunkte, ruft das Modell auf und protokolliert die Kosten.
┌─────────────────────────────────────────────────────────────┐
│ 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│ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────────────┘
Server-Komponenten laden den Historieverlauf ohne zusätzliches JavaScript, async FastAPI handhabt langsame Modellaufrufe ohne Thread-Blockaden, pgvector bietet Raum für Abfragen, Redis speichert Rate-Limit-Zähler sowie gekachelte Embeddings, und ein Worker übernimmt die PDF-Parser- und Embedding-Aufgaben vom Anfragepfad.
Datenmodell: Ergänzung zu einem Buchhaltungsbuch
Das Schema definiert Benutzer, Konversationen, Nachrichten sowie ein Kreditbuch.
# 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)
Anstelle einer Änderung des Saldo werden jede Kauf- oder Nutzungshandlung sowie der Summenwert dieser Handlungen aufgezeichnet, wodurch ein Nachvollziehbarkeitsverlauf entsteht und keine Updates verloren gehen. Bevor es verwendet wird, müssen einige Dinge korrigiert werden: Message.metadata kollidiert mit einem von SQLAlchemy’s deklarativer Basis vorgesehenen Attribut, daher sollte es umbenannt werden (zum Beispiel in meta); fehlende Importe wie Boolean und declarative_base; außerdem stammt der Spaltentyp von pgvector aus dem pgvector-Paket und nicht aus SQLAlchemy’s PostgreSQL-Dialekt.
FastAPI-Dienste und -Routen
Ein herkömmliches Layout trennt Authentifizierung, KI-Logik, Router und Middleware voneinander:
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
Ein Streaming-Dienst, der die Nutzung misst
Diese Dienstleistung erlaubt pro Benutzer über Redis 20 Anfragen pro Minute, prüft den Saldo, streamt Server-Sent Events und speichert anschließend die Assistentennachricht zusammen mit einem negativen Buchungseintrag.
# 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
Hinweise: Ein Token pro Datenblock ist nur eine Schätzung – verwenden Sie daher die vom Anbieter gemeldete Nutzung oder einen Tokenisierer. Der festgelegte Preis dient nur zur Veranschaulichung und ist nicht aktuell; laden Sie die Preise aus der Konfiguration. Zudem ist das Prüfen des Saldo vor dem Streamen und das Abziehen danach nicht atomar – reservieren Sie die Credits im Voraus oder sperren Sie den Benutzerdatensatz.
Die Nachrichten-Endpunkt
Der Router überprüft das Eigentumsrecht (404 statt 403, um IDs nicht preiszugeben), speichert die Benutzernachricht und gibt eine StreamingResponse zurück:
# 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"
)
Mit asynchronem SQLAlchemy führt das verspätete Zugriff auf conversation.messages zu einem Fehler – laden Sie daher die Beziehung frühzeitig.
Eine globale Geschwindigkeitsbeschränkung
Eine zweite, grobere Beschränkung von 100 Anfragen pro Minute wird als Middleware implementiert:
# 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
Ein in BaseHTTPMiddleware ausgelöster HTTPException wird von FastAPIs Handlern nicht in einen sauberen Status 429 umgewandelt, daher wird stattdessen eine JSONResponse mit Status 429 zurückgegeben. Feste Zeitfenster ermöglichen außerdem kurzfristige Anstiege der Anfragenanzahl.
Die Frontend-Plattform
Der App Router verbindet Server-Komponentenseiten mit Client-Komponenten für den Live-Chat:
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
Laden des Historiens auf dem Server
Die Seite liest die Cookie-Daten und lädt gleichzeitig den Chatverlauf sowie die Nachrichten herunter, sodass der Chat bereits bei der ersten Anzeige sichtbar ist:
// 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>
);
}
In Next.js 15 und neueren Versionen sind cookies() und params asynchron; warten Sie mit ihnen ab.
Streaming in die Benutzeroberfläche
Der Client fügt optimistisch die Benachrichtigung des Benutzers sowie einen leeren Platzhalter für den Assistenten hinzu, sendet eine POST-Anfrage und füllt den Platzhalter aus dem Antwortstrom:
// 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>
);
}
Trotz des Kommentars wird hier fetch mit einem Stream-Reader verwendet und nicht EventSource, da letzteres keine POST-Körper senden kann. Puffern Sie Teilzeilen, da Datenblöcke diese trennen können, und fügen Sie einen AbortController hinzu, um bei Entfernung abzubrechen. Die relative URL /api/... setzt außerdem voraus, dass es einen Route-Handler oder eine Umleitung gibt, die auf FastAPI verweist.
Ein Axios-Client für alles Weitere
Mit httpOnly-Cookies sendet der Client Anmeldeinformationen, fügt eine Tracking-ID hinzu und versucht nach einem 401-Fehler beim Erneuern noch einmal:
// 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);
}
);
Jeder parallele 401-Fehler löst einen Neuladen aus; siehe die Lösung für ein fehlerfreies Neuladen für eine version, bei der es zu keinen Konflikten kommt.
Durchsetzung von Krediten als Abhängigkeit
# 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
):
...
Auch hier gilt die oben genannte nicht-atomare Einschränkung.
Testen ohne Aufruf des Modells
MSW 2.x gibt einen künstlichen Ereignisstrom zurück, und ein Test mit der Testing Library überprüft den optimistischen sowie den gestreamten Text:
// 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();
});
});
Der Handler simuliert die URL des Anbieters, doch das Komponente ruft Ihre API auf – simulieren Sie daher stattdessen Ihren Endpunkt. Der Backend-Test patcht den Client und überprüft, ob genau eine Zeile im Kontostand vorhanden ist:
# 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
Deployment mit Containern
Ein mehrstufiger Dockerfile baut Next.js im Standalone-Modus und führt es als Nicht-Root-Benutzer aus:
# 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 startet pgvector Postgres mit einem Health-Check, Redis, die API, das Frontend sowie 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_-Werte werden zur Zeit der Build-Vorgänge für den Browser festgelegt, wodurch backend:8000 nicht auflösbar ist; verwenden Sie stattdessen die öffentliche nginx-Route.
Wichtige Erkenntnisse
- Halten Sie Schlüssel, Preise und Limits auf Serverseite – das Modell ist eine Abhängigkeit hinter Ihrer API.
- Erfassen Sie die Nutzung in einem nur zum Hinzufügen geeigneten Protokoll und machen Sie den Prüf- sowie Abzugsschritt atomar.
- Streamen Sie mit expliziter Abbruchmöglichkeit und Zeilenpufferung.
- Mimikrierten Sie in Tests Ihren eigenen API-Vertrag, nicht den des Anbieters.