AI 应用全栈开发指南：用 FastAPI + Next.js 构建 LLM 服务

概述

现代 AI 应用开发早已超出简单的 API 调用范畴，需要整合流式传输、多模态、RAG（检索增强生成）等复杂技术。本指南将带你走完以 FastAPI 后端与 Next.js 前端为基础，构建生产级 LLM 服务的全过程。

1. AI 应用架构设计

现代 AI 应用的三层结构

现代 AI 应用由三个核心层组成。

• 前端层：Next.js App Router、Vercel AI SDK、流式 UI
• 后端层：FastAPI、LangChain、认证中间件、缓存
• AI/数据层：OpenAI/Claude API、向量数据库、嵌入模型

流式处理 vs 批处理

处理 LLM 响应主要有两种方式。

流式处理是在生成 token 的同时立即发送给客户端的方式。用户感知到的响应速度更快，适合对话式界面。使用 Server-Sent Events（SSE）或 WebSocket。

批处理是等整个响应完成后再一次性返回的方式。适合文档处理、数据分析、后台任务。可利用 Celery + Redis 队列实现。

项目文件夹结构

ai-app/
├── backend/
│   ├── app/
│   │   ├── main.py
│   │   ├── routers/
│   │   │   ├── chat.py
│   │   │   └── documents.py
│   │   ├── services/
│   │   │   ├── llm_service.py
│   │   │   └── vector_service.py
│   │   └── models/
│   │       └── schemas.py
│   ├── requirements.txt
│   └── Dockerfile
├── frontend/
│   ├── app/
│   │   ├── chat/
│   │   │   └── page.tsx
│   │   └── api/
│   │       └── chat/
│   │           └── route.ts
│   ├── components/
│   └── package.json
└── docker-compose.yml

2. FastAPI 后端搭建

安装与基本设置

pip install fastapi uvicorn openai langchain langchain-openai python-dotenv

用 Pydantic 模型做请求/响应校验

# app/models/schemas.py
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class Role(str, Enum):
    user = "user"
    assistant = "assistant"
    system = "system"

class Message(BaseModel):
    role: Role
    content: str

class ChatRequest(BaseModel):
    messages: List[Message]
    model: str = Field(default="gpt-4o-mini")
    temperature: float = Field(default=0.7, ge=0, le=2)
    max_tokens: Optional[int] = Field(default=None)

class ChatResponse(BaseModel):
    content: str
    usage: dict

异步流式接口

使用 FastAPI 的 StreamingResponse，可以把 LLM token 实时发送给客户端。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from openai import AsyncOpenAI
from app.models.schemas import ChatRequest

app = FastAPI(title="AI App Backend")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_methods=["*"],
    allow_headers=["*"],
)

client = AsyncOpenAI()

@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        stream = await client.chat.completions.create(
            model=request.model,
            messages=[m.dict() for m in request.messages],
            stream=True,
            temperature=request.temperature,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f"data: {delta}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

依赖注入模式

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="토큰이 만료되었습니다."
        )

@app.post("/api/chat/secure")
async def secure_chat(request: ChatRequest, user=Depends(verify_token)):
    # 仅限已通过身份验证的用户访问
    pass

3. LangChain 集成

构建对话链

使用 LangChain，可以轻松实现内存管理、链式组合与工具集成。

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.prompts import PromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
memory = ConversationBufferWindowMemory(k=10)

template = """당신은 친절한 AI 어시스턴트입니다.
현재 대화:
{history}
Human: {input}
AI:"""

prompt = PromptTemplate(
    input_variables=["history", "input"],
    template=template
)

chain = ConversationChain(llm=llm, memory=memory, prompt=prompt)

response = chain.predict(input="안녕하세요, 저는 Python 개발자입니다.")

构建 RAG 流水线

RAG（检索增强生成）是通过检索外部文档来提升 LLM 响应质量的模式。

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader

# 加载文档并分块
loader = PyPDFLoader("document.pdf")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(documents)

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 组建 RAG 链
llm = ChatOpenAI(model="gpt-4o")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

answer = qa_chain.invoke({"query": "문서의 주요 내용은?"})

编写自定义工具

from langchain.tools import tool
from langchain.agents import initialize_agent, AgentType

@tool
def search_database(query: str) -> str:
    """데이터베이스에서 정보를 검색합니다. query는 검색할 키워드입니다."""
    # 实际的数据库查询逻辑
    results = db.search(query)
    return str(results)

@tool
def get_weather(city: str) -> str:
    """특정 도시의 현재 날씨를 조회합니다."""
    response = requests.get(f"https://api.weather.com/v1/{city}")
    return response.json()["description"]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_database, get_weather]
agent = initialize_agent(tools, llm, agent=AgentType.OPENAI_FUNCTIONS)

4. Next.js 前端

用 Vercel AI SDK 实现流式聊天

Vercel AI SDK 是在 Next.js 中便捷实现 AI 流式传输的官方库。

npm install ai @ai-sdk/openai react-markdown

// app/api/chat/route.ts
import { openai } from '@ai-sdk/openai'
import { streamText } from 'ai'

export async function POST(req: Request) {
  const { messages } = await req.json()

  const result = await streamText({
    model: openai('gpt-4o-mini'),
    messages,
    system: '당신은 친절하고 도움이 되는 AI 어시스턴트입니다.',
  })

  return result.toDataStreamResponse()
}

聊天界面组件

// app/chat/page.tsx
'use client'
import { useChat } from 'ai/react'
import ReactMarkdown from 'react-markdown'

export default function ChatPage() {
  const { messages, input, handleInputChange, handleSubmit, isLoading } = useChat({
    api: '/api/chat',
  })

  return (
    <div className="flex flex-col h-screen max-w-2xl mx-auto">
      <header className="p-4 border-b font-semibold text-lg">
        AI 어시스턴트
      </header>
      <div className="flex-1 overflow-y-auto p-4 space-y-4">
        {messages.map(m => (
          <div
            key={m.id}
            className={`flex ${m.role === 'user' ? 'justify-end' : 'justify-start'}`}
          >
            <div
              className={`max-w-xs rounded-lg p-3 ${
                m.role === 'user'
                  ? 'bg-blue-500 text-white'
                  : 'bg-gray-100 text-gray-800'
              }`}
            >
              <ReactMarkdown>{m.content}</ReactMarkdown>
            </div>
          </div>
        ))}
        {isLoading && (
          <div className="flex justify-start">
            <div className="bg-gray-100 rounded-lg p-3 text-gray-500">
              답변 생성 중...
            </div>
          </div>
        )}
      </div>
      <form onSubmit={handleSubmit} className="p-4 border-t flex gap-2">
        <input
          value={input}
          onChange={handleInputChange}
          className="flex-1 border rounded-lg px-3 py-2 focus:outline-none focus:ring-2 focus:ring-blue-500"
          placeholder="메시지를 입력하세요..."
          disabled={isLoading}
        />
        <button
          type="submit"
          disabled={isLoading}
          className="bg-blue-500 text-white px-4 py-2 rounded-lg disabled:opacity-50"
        >
          전송
        </button>
      </form>
    </div>
  )
}

处理文件上传

// app/upload/page.tsx
'use client'
import { useState } from 'react'

export default function UploadPage() {
  const [status, setStatus] = useState('')

  async function handleUpload(e: React.FormEvent<HTMLFormElement>) {
    e.preventDefault()
    const formData = new FormData(e.currentTarget)
    setStatus('업로드 중...')

    const response = await fetch('/api/upload', {
      method: 'POST',
      body: formData,
    })

    if (response.ok) {
      const data = await response.json()
      setStatus(`완료: ${data.message}`)
    } else {
      setStatus('업로드 실패')
    }
  }

  return (
    <form onSubmit={handleUpload} className="p-4">
      <input type="file" name="file" accept=".pdf,.txt,.md" />
      <button type="submit" className="mt-2 bg-green-500 text-white px-4 py-2 rounded">
        업로드
      </button>
      {status && <p className="mt-2 text-sm">{status}</p>}
    </form>
  )
}

5. 向量数据库集成

pgvector（PostgreSQL 扩展）

使用 PostgreSQL 的 pgvector 扩展，可以在现有数据库中执行向量检索。

-- 启用 pgvector 扩展
CREATE EXTENSION vector;

-- 创建包含嵌入列的表
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1536),
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 创建 HNSW 索引（快速近似最近邻检索）
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

# pgvector 使用示例
import asyncpg
import numpy as np

async def store_embedding(content: str, embedding: list):
    conn = await asyncpg.connect(DATABASE_URL)
    await conn.execute(
        "INSERT INTO documents (content, embedding) VALUES ($1, $2)",
        content, embedding
    )

async def search_similar(query_embedding: list, k: int = 5):
    conn = await asyncpg.connect(DATABASE_URL)
    results = await conn.fetch(
        """SELECT content, 1 - (embedding <=> $1) as similarity
           FROM documents
           ORDER BY embedding <=> $1
           LIMIT $2""",
        query_embedding, k
    )
    return results

Chroma DB（本地开发用）

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="my_documents",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

# 添加文档
vectorstore.add_texts(
    texts=["Python은 AI 개발에 널리 사용됩니다.", "FastAPI는 고성능 API 프레임워크입니다."],
    metadatas=[{"source": "intro.txt"}, {"source": "framework.txt"}]
)

# 相似度检索
results = vectorstore.similarity_search("API 개발", k=3)

6. 认证与安全

JWT 令牌认证

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    username: str = payload.get("sub")
    if username is None:
        raise HTTPException(status_code=401, detail="유효하지 않은 토큰")
    return username

防御提示词注入

import re

INJECTION_PATTERNS = [
    r"ignore previous instructions",
    r"disregard all prior",
    r"you are now",
    r"act as",
    r"pretend you are",
]

def sanitize_input(user_input: str) -> str:
    lower_input = user_input.lower()
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, lower_input):
            raise HTTPException(
                status_code=400,
                detail="잠재적으로 유해한 입력이 감지되었습니다."
            )
    # 限制最大长度
    if len(user_input) > 4000:
        raise HTTPException(status_code=400, detail="입력이 너무 깁니다.")
    return user_input.strip()

速率限制

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/api/chat")
@limiter.limit("10/minute")
async def chat(request: Request, chat_request: ChatRequest):
    # 每分钟限制 10 次
    pass

7. 多模态输入处理

图像分析（GPT-4o Vision）

import base64
from pathlib import Path

async def analyze_image(image_path: str, question: str) -> str:
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")

    ext = Path(image_path).suffix.lower()
    mime_map = {".jpg": "image/jpeg", ".png": "image/png", ".gif": "image/gif"}
    media_type = mime_map.get(ext, "image/jpeg")

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{media_type};base64,{image_data}"
                        },
                    },
                    {"type": "text", "text": question}
                ],
            }
        ],
    )
    return response.choices[0].message.content

用 Whisper API 处理语音

async def transcribe_audio(audio_file_path: str) -> str:
    with open(audio_file_path, "rb") as audio_file:
        transcript = await client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            language="ko"
        )
    return transcript.text

8. 性能优化

Redis 响应缓存

import redis
import json
import hashlib

redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)

def get_cache_key(messages: list) -> str:
    content = json.dumps(messages, sort_keys=True)
    return hashlib.md5(content.encode()).hexdigest()

async def cached_chat(messages: list) -> str:
    cache_key = get_cache_key(messages)
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages
    )
    result = response.choices[0].message.content

    # 缓存 1 小时 TTL
    redis_client.setex(cache_key, 3600, json.dumps(result))
    return result

连接池

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    echo=False,
)

AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

9. Docker Compose 部署

docker-compose.yml

version: '3.8'
services:
  backend:
    build: ./backend
    ports:
      - '8000:8000'
    environment:
      - OPENAI_API_KEY=your_key
      - DATABASE_URL=postgresql+asyncpg://user:pass@db/aiapp
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    restart: unless-stopped

  frontend:
    build: ./frontend
    ports:
      - '3000:3000'
    environment:
      - NEXT_PUBLIC_API_URL=http://backend:8000
    depends_on:
      - backend
    restart: unless-stopped

  db:
    image: pgvector/pgvector:pg16
    environment:
      - POSTGRES_DB=aiapp
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped

volumes:
  postgres_data:
  redis_data:

backend/Dockerfile

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

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
COPY --from=builder /app/.next/standalone ./
COPY --from=builder /app/.next/static ./.next/static

EXPOSE 3000
CMD ["node", "server.js"]

部署命令

# 构建并启动
docker-compose up --build -d

# 查看日志
docker-compose logs -f backend

# 横向扩容（后端 3 个实例）
docker-compose up --scale backend=3 -d

# 停止
docker-compose down

10. 小测验：核心概念自查

参考资料

• FastAPI 官方文档
• Vercel AI SDK 文档
• LangChain 文档
• Next.js App Router 文档
• pgvector GitHub
• OpenAI API 参考