想让自己的产品接入 AI，但不想被 OpenAI 的域名访问和高价格绑死？本文带你用 FastAPI 搭建一套自己的 AI 中台 API，支持多模型切换、流式输出、对话历史管理，生产可用。
为什么要自建 AI API？
很多开发者直接让前端调用 OpenAI/通义千问的官方 API，这有几个问题：
| 问题 | 说明 |
|---|---|
| API Key 暴露 | 前端代码可逆向，Key 泄漏即被盗刷 |
| 无法切换模型 | 前端写死了模型，换模型要改所有端 |
| 无访问控制 | 任何人都能调，没有鉴权和频率限制 |
| 无日志记录 | 不知道用户问了什么，无法改进 |
| 无法缓存 | 同样的问题反复调用 API，浪费 Token |
自建 AI API 中台能解决所有这些问题：统一鉴权、统一模型管理、统一日志、统一缓存，前端只需要调你自己的 API。
技术选型
为什么选 FastAPI？
Django REST → 适合大型项目，功能全但重
Flask → 简单但异步支持弱
FastAPI → 异步原生、自动文档、类型检查、高性能
FastAPI 在 AI 场景下的优势：
– 原生异步：完美支持流式输出（SSE）
– 自动文档：访问 /docs 即有 Swagger UI
– Pydantic 验证：请求参数自动校验，省去大量 if 判断
– 性能：基于 Starlette，接近 Node.js 水平
项目结构
ai-api/
├── main.py # FastAPI 主入口
├── routers/
│ ├── chat.py # 对话接口
│ └── models.py # 模型列表接口
├── services/
│ ├── base.py # 模型基类
│ ├── qwen.py # 通义千问
│ ├── deepseek.py # DeepSeek
│ └── openai_svc.py # OpenAI/兼容接口
├── middleware/
│ ├── auth.py # API Key 鉴权
│ └── rate_limit.py # 频率限制
├── models/
│ └── schemas.py # 请求/响应数据模型
├── requirements.txt
└── .env
快速开始
安装依赖
pip install fastapi uvicorn[standard] httpx python-dotenv pydantic
requirements.txt
fastapi>=0.110.0
uvicorn[standard]>=0.27.0
httpx>=0.27.0
python-dotenv>=1.0.0
pydantic>=2.0.0
核心代码实现
1. 数据模型（models/schemas.py）
from pydantic import BaseModel, Field
from typing import Literal, Optional
from enum import Enum
class ModelName(str, Enum):
qwen_turbo = "qwen-turbo"
qwen_plus = "qwen-plus"
deepseek_v3 = "deepseek-chat"
gpt35 = "gpt-3.5-turbo"
class Message(BaseModel):
role: Literal["system", "user", "assistant"]
content: str
class ChatRequest(BaseModel):
model: ModelName = Field(default=ModelName.qwen_turbo, description="模型名称")
messages: list[Message] = Field(..., min_length=1, description="对话历史")
stream: bool = Field(default=False, description="是否流式输出")
temperature: float = Field(default=0.7, ge=0, le=2)
max_tokens: Optional[int] = Field(default=None, ge=1, le=4096)
system_prompt: Optional[str] = Field(default=None, description="系统提示词")
class ChatResponse(BaseModel):
id: str
model: str
content: str
usage: dict
finish_reason: str
2. 模型服务基类（services/base.py）
from abc import ABC, abstractmethod
from typing import AsyncIterator
from models.schemas import ChatRequest, ChatResponse
class BaseModelService(ABC):
"""所有模型服务的基类，统一接口"""
@abstractmethod
async def chat(self, request: ChatRequest) -> ChatResponse:
"""非流式对话"""
pass
@abstractmethod
async def chat_stream(self, request: ChatRequest) -> AsyncIterator[str]:
"""流式对话，逐 token 返回"""
pass
3. 通义千问服务（services/qwen.py）
import httpx
import json
import uuid
from typing import AsyncIterator
from models.schemas import ChatRequest, ChatResponse
from services.base import BaseModelService
DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
class QwenService(BaseModelService):
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def _build_messages(self, request: ChatRequest) -> list:
messages = []
if request.system_prompt:
messages.append({"role": "system", "content": request.system_prompt})
messages.extend([m.model_dump() for m in request.messages])
return messages
async def chat(self, request: ChatRequest) -> ChatResponse:
payload = {
"model": request.model.value,
"input": {"messages": self._build_messages(request)},
"parameters": {
"result_format": "message",
"temperature": request.temperature
}
}
if request.max_tokens:
payload["parameters"]["max_tokens"] = request.max_tokens
async with httpx.AsyncClient(timeout=60) as client:
resp = await client.post(DASHSCOPE_BASE_URL, headers=self.headers, json=payload)
resp.raise_for_status()
data = resp.json()
choice = data["output"]["choices"][0]
return ChatResponse(
id=str(uuid.uuid4()),
model=request.model.value,
content=choice["message"]["content"],
usage=data.get("usage", {}),
finish_reason=choice.get("finish_reason", "stop")
)
async def chat_stream(self, request: ChatRequest) -> AsyncIterator[str]:
payload = {
"model": request.model.value,
"input": {"messages": self._build_messages(request)},
"parameters": {
"result_format": "message",
"temperature": request.temperature,
"incremental_output": True # 通义千问流式增量输出
}
}
headers = {**self.headers, "X-DashScope-SSE": "enable"}
async with httpx.AsyncClient(timeout=120) as client:
async with client.stream("POST", DASHSCOPE_BASE_URL,
headers=headers, json=payload) as resp:
async for line in resp.aiter_lines():
if line.startswith("data:"):
data_str = line[5:].strip()
if data_str == "[DONE]":
break
try:
data = json.loads(data_str)
content = data["output"]["choices"][0]["message"]["content"]
if content:
yield content
except (json.JSONDecodeError, KeyError):
continue
4. DeepSeek 服务（services/deepseek.py）
DeepSeek 兼容 OpenAI 格式，代码更简单：
import httpx
import json
import uuid
from typing import AsyncIterator
from models.schemas import ChatRequest, ChatResponse
from services.base import BaseModelService
DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1/chat/completions"
class DeepSeekService(BaseModelService):
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def _build_payload(self, request: ChatRequest, stream: bool = False) -> dict:
messages = []
if request.system_prompt:
messages.append({"role": "system", "content": request.system_prompt})
messages.extend([m.model_dump() for m in request.messages])
payload = {
"model": "deepseek-chat",
"messages": messages,
"temperature": request.temperature,
"stream": stream
}
if request.max_tokens:
payload["max_tokens"] = request.max_tokens
return payload
async def chat(self, request: ChatRequest) -> ChatResponse:
async with httpx.AsyncClient(timeout=60) as client:
resp = await client.post(DEEPSEEK_BASE_URL,
headers=self.headers,
json=self._build_payload(request))
resp.raise_for_status()
data = resp.json()
return ChatResponse(
id=data["id"],
model=data["model"],
content=data["choices"][0]["message"]["content"],
usage=data.get("usage", {}),
finish_reason=data["choices"][0]["finish_reason"]
)
async def chat_stream(self, request: ChatRequest) -> AsyncIterator[str]:
async with httpx.AsyncClient(timeout=120) as client:
async with client.stream("POST", DEEPSEEK_BASE_URL,
headers=self.headers,
json=self._build_payload(request, stream=True)) as resp:
async for line in resp.aiter_lines():
if line.startswith("data: "):
data_str = line[6:].strip()
if data_str == "[DONE]":
break
try:
data = json.loads(data_str)
delta = data["choices"][0]["delta"].get("content", "")
if delta:
yield delta
except (json.JSONDecodeError, KeyError):
continue
5. 鉴权中间件（middleware/auth.py）
from fastapi import Request, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import os
VALID_API_KEYS = set(os.getenv("VALID_API_KEYS", "test-key-123").split(","))
security = HTTPBearer()
async def verify_api_key(credentials: HTTPAuthorizationCredentials = None):
"""依赖注入形式的 API Key 验证"""
if credentials is None or credentials.credentials not in VALID_API_KEYS:
raise HTTPException(
status_code=401,
detail="Invalid or missing API key",
headers={"WWW-Authenticate": "Bearer"}
)
return credentials.credentials
6. 对话路由（routers/chat.py）
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
import json
from models.schemas import ChatRequest, ChatResponse, ModelName
from services.qwen import QwenService
from services.deepseek import DeepSeekService
from middleware.auth import verify_api_key
import os
router = APIRouter(prefix="/v1", tags=["chat"])
# 根据环境变量初始化模型服务
def get_service(model: ModelName):
if model in (ModelName.qwen_turbo, ModelName.qwen_plus):
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
raise HTTPException(500, "Qwen API Key 未配置")
return QwenService(api_key)
elif model == ModelName.deepseek_v3:
api_key = os.getenv("DEEPSEEK_API_KEY")
if not api_key:
raise HTTPException(500, "DeepSeek API Key 未配置")
return DeepSeekService(api_key)
else:
raise HTTPException(400, f"不支持的模型: {model}")
@router.post("/chat/completions")
async def chat_completions(
request: ChatRequest,
api_key: str = Depends(verify_api_key)
):
service = get_service(request.model)
# 流式响应
if request.stream:
async def event_stream():
async for chunk in service.chat_stream(request):
data = json.dumps({"choices": [{"delta": {"content": chunk}}]})
yield f"data: {data}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
}
)
# 普通响应
response = await service.chat(request)
return response
7. 主入口（main.py）
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv
import uvicorn
from routers import chat
load_dotenv()
app = FastAPI(
title="AI API 中台",
description="统一管理多个大模型的 API 中台服务",
version="1.0.0"
)
# CORS 配置（根据实际需求限制 origins）
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["POST", "GET"],
allow_headers=["*"],
)
app.include_router(chat.router)
@app.get("/health")
async def health():
return {"status": "ok", "version": "1.0.0"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
.env
DASHSCOPE_API_KEY=your_qwen_key
DEEPSEEK_API_KEY=your_deepseek_key
VALID_API_KEYS=your-app-key-1,your-app-key-2
启动与测试
# 开发模式
python main.py
# 生产模式（多进程）
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
访问自动生成的文档：http://localhost:8000/docs
测试接口（curl）
# 普通请求
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Authorization: Bearer your-app-key-1" \
-H "Content-Type: application/json" \
-d '{
"model": "qwen-turbo",
"messages": [{"role": "user", "content": "用一句话解释FastAPI"}],
"stream": false
}'
# 流式请求
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Authorization: Bearer your-app-key-1" \
-H "Content-Type: application/json" \
-d '{
"model": "deepseek-chat",
"messages": [{"role": "user", "content": "写一首关于云计算的五言绝句"}],
"stream": true
}'
用 Docker 部署到云服务器
Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]
# 构建镜像
docker build -t ai-api .
# 运行（从 .env 文件注入环境变量）
docker run -d \
--name ai-api \
-p 8000:8000 \
--env-file .env \
--restart unless-stopped \
ai-api
可扩展的功能方向
| 功能 | 实现思路 |
|---|---|
| Token 用量统计 | 记录每个 Key 的使用量到 Redis |
| 请求频率限制 | slowapi 库 + Redis |
| 对话历史持久化 | 存 Redis，按 session_id 管理 |
| 响应缓存 | 相同问题 hash 后查缓存，节省 Token |
| 多租户管理 | 每个 Key 绑定不同的模型配置 |
| 监控告警 | Prometheus + Grafana 接入 |
总结
一个生产可用的 AI API 中台需要具备：
- 统一鉴权：前端不直接持有大模型 Key
- 多模型路由：一套接口，多模型切换
- 流式输出：用户体验更好
- 错误处理：优雅降级，不让异常暴露给用户
FastAPI 在这个场景下是最合适的选择——异步原生、文档自动生成、代码清晰。一旦这套基础设施搭好，接入新模型只需新增一个 Service 类，10 分钟搞定。
完整代码已整理，欢迎参考和改进。
关于作者
长期关注大模型应用落地与云服务器实战，专注技术在企业场景中的落地实践。
个人博客：y1cloud.cn —— 持续更新云计算、AI大模型实战教程，欢迎访问交流。