0%
💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.
概述
现代 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 /app/.next/standalone ./
COPY /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. 小测验：核心概念自查
Q1. 为什么在 AI 流式传输中使用 SSE（Server-Sent Events）？
答案：为了让 LLM 一生成 token 就立即传给客户端，使文本在用户等待响应期间能够实时显示出来。
解释：比 WebSocket 更简单，最适合通过 HTTP 实现从服务端到客户端的单向流式传输。FastAPI 的 StreamingResponse 与前端的 EventSource API 正好支持这种模式。
Q2. RAG（检索增强生成）的核心优势是什么？
答案：可以克服 LLM 训练数据的局限，提供最新或特定领域的信息。
解释：无需重新训练模型，只需检索外部文档并纳入上下文即可。通过向量相似度检索找到相关文档，再将其注入提示词中生成准确的回答。这也有助于减少幻觉（hallucination）现象。
Q3. 在 FastAPI 中使用 Pydantic 模型的主要原因是什么？
答案：为了实现请求与响应数据的自动校验、序列化，以及自动生成 API 文档。
解释：Pydantic 基于 Python 类型提示在运行时校验数据。FastAPI 借此自动生成 OpenAPI（Swagger）文档，并针对错误输入返回清晰的错误信息。
Q4. 为什么向量数据库要使用 HNSW 索引？
答案：为了在高维向量空间中快速执行近似最近邻（ANN）检索。
解释：以暴力方式比较数百万条向量太慢。HNSW（Hierarchical Navigable Small World）通过分层图结构大幅提升检索速度，同时保持较高的准确率。pgvector、Chroma、Weaviate 等均支持该索引。
Q5. LangChain 的 ConversationBufferWindowMemory 中 k 参数的作用是什么？
答案：指定要在上下文窗口中保留的最近对话轮数（turn）。
解释：LLM 存在 token 上限，无法发送全部对话历史。设为 k=10 时，会保留最近 10 轮用户与 AI 的交互，更早的内容会被删除。这是在内存成本与上下文保持之间取得平衡的重要参数。
参考资料
현재 단락 (1/450)
现代 AI 应用开发早已超出简单的 API 调用范畴，需要整合流式传输、多模态、RAG（检索增强生成）等复杂技术。本指南将带你走完以 FastAPI 后端与 Next.js 前端为基础，构建生产级 L...
작성 글자: 0자원문 글자: 13,194자작성 단락: 0/450