《AI大模型应知应会100篇》第61篇：FastAPI搭建大模型API服务

第61篇：FastAPI搭建大模型API服务

摘要

随着大语言模型（LLM）在各行各业的广泛应用，构建一个高性能、可扩展的大模型 API 服务变得尤为重要。FastAPI 以其异步支持、类型安全、自动生成文档等优势，成为当前最流行的选择之一。

本篇文章将从零开始，手把手教你使用 FastAPI 搭建一个完整的大模型 API 服务，涵盖项目初始化、接口设计、模型集成、安全认证、部署方案及性能优化等全流程内容，并附有完整的代码示例与部署指南。

核心概念与知识点

1. FastAPI基础架构

项目结构设计（推荐目录组织）

fastapi-llm-service/
├── app/
│   ├── main.py               # 主入口
│   ├── api/                  # 接口模块
│   │   ├── chat.py
│   │   ├── models.py
│   │   └── __init__.py
│   ├── services/             # 服务层
│   │   ├── model_service.py
│   │   └── cache.py
│   ├── utils/                # 工具函数
│   │   └── logger.py
│   ├── config.py             # 配置文件
│   └── models/               # Pydantic 数据模型
│       └── request_models.py
├── Dockerfile
├── requirements.txt
└── README.md

✅ 这种分层结构便于后期维护和扩展。

FastAPI 的依赖注入机制

FastAPI 支持强大的依赖注入系统，可以用于管理数据库连接、身份验证、配置加载等。

from fastapi import Depends, FastAPI

app = FastAPI()

def get_db():
    db = "Connected to DB"
    yield db
    print("Closing DB connection")

@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
    return {"db": db}

💡 使用 Depends() 可以实现跨接口复用逻辑，提升代码可维护性。

使用 Pydantic 进行请求/响应验证

Pydantic 是 FastAPI 的核心组件之一，它提供数据校验功能，确保输入输出符合预期格式。

from pydantic import BaseModel

class ChatRequest(BaseModel):
    prompt: str
    max_tokens: int = 50
    temperature: float = 0.7

@app.post("/chat")
async def chat(request: ChatRequest):
    return {"response": f"Received: {request.prompt}"}

🧪 如果请求体不符合 ChatRequest 定义的字段或类型，会自动返回 422 错误。

异步处理与高并发支持

FastAPI 基于 ASGI（如 Uvicorn），天然支持异步编程，适合处理大模型推理这类 I/O 密集型任务。

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/slow")
async def slow_api():
    await asyncio.sleep(3)
    return {"message": "This is a slow endpoint"}

⏱️ 上述接口模拟了一个耗时操作，但不会阻塞主线程，非常适合调用远程模型 API 或本地 GPU 推理。

2. 核心API设计【实战部分】

Chat 接口实现（支持流式响应）

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def generate_stream(prompt: str):
    for word in ["Hello", prompt, "How", "can", "I", "help", "?"]:
        yield f"data: {word}\n\n"
        await asyncio.sleep(0.5)

@app.post("/stream-chat")
async def stream_chat(request: ChatRequest):
    return StreamingResponse(generate_stream(request.prompt), media_type="text/event-stream")

🔁 流式输出能显著提升用户体验，尤其适用于对话类应用。

嵌入生成接口设计

from typing import List

@app.post("/embeddings")
async def embeddings(texts: List[str]):
    # 模拟嵌入生成
    return {
        "embeddings": [[0.1 * i for i in range(128)] for _ in texts]
    }

文档上传与处理端点

from fastapi import UploadFile, File

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {
        "filename": file.filename,
        "type": file.content_type,
        "size": len(content)
    }

📁 可以结合 OCR/NLP 模块对上传文档进行进一步处理。

多模型路由策略（按需切换模型）

model_mapping = {
    "gpt": GPTModel(),
    "llama": LlamaModel(),
    "chatglm": ChatGLMModel()
}

@app.post("/multi-model")
async def multi_model(model_name: str, request: ChatRequest):
    if model_name not in model_mapping:
        raise HTTPException(status_code=404, detail="Model not found")
    model = model_mapping[model_name]
    response = model.generate(request.prompt, ...)
    return {"response": response}

3. 服务层实现【实战部分】

集成 OpenAI / 本地模型

import openai

class GPTModel:
    def generate(self, prompt, max_tokens=50, temperature=0.7):
        response = openai.Completion.create(
            engine="text-davinci-003",
            prompt=prompt,
            max_tokens=max_tokens,
            temperature=temperature
        )
        return response.choices[0].text.strip()

对于本地模型，如 HuggingFace Transformers：

from transformers import pipeline

class LocalModel:
    def __init__(self):
        self.pipe = pipeline("text-generation", model="distilgpt2")
    
    def generate(self, prompt, max_length=50):
        result = self.pipe(prompt, max_length=max_length, num_return_sequences=1)
        return result[0]['generated_text']

实现缓存与速率限制

使用 Redis 缓存：

import redis.asyncio as redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

async def get_cached_response(prompt):
    cached = await redis_client.get(prompt)
    return cached.decode() if cached else None

async def set_cached_response(prompt, response):
    await redis_client.setex(prompt, 3600, response)

错误处理与优雅降级

from fastapi.exceptions import HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail},
    )

并发控制与负载均衡

可以通过中间件或使用 concurrent.futures.ThreadPoolExecutor 来限制并发请求数量。

4. 安全与认证

API Key 认证机制

from fastapi.security import APIKeyHeader

API_KEY_HEADER = APIKeyHeader(name="X-API-Key")

@app.get("/secure")
async def secure_endpoint(api_key: str = Depends(API_KEY_HEADER)):
    if api_key != "SECRET123":
        raise HTTPException(status_code=403, detail="Invalid API key")
    return {"message": "Access granted"}

JWT 身份验证

使用 python-jose 和 passlib 实现 JWT 登录流程：

pip install python-jose[cryptography] passlib

from jose import jwt
from datetime import datetime, timedelta

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

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

请求签名与加密通信

建议使用 HTTPS + 签名机制防止重放攻击。例如使用 HMAC 对请求头进行签名：

import hmac
import hashlib

def verify_signature(payload: str, signature: str, secret: str):
    expected_sig = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected_sig, signature)

5. 部署与扩展【实战部分】

Docker 容器化部署

编写 Dockerfile：

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

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

构建镜像并运行：

docker build -t llm-api .
docker run -p 8000:8000 llm-api

Kubernetes 编排方案

使用 Helm Chart 或 YAML 文件部署到 K8s：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: llm-api
  template:
    metadata:
      labels:
        app: llm-api
    spec:
      containers:
      - name: llm-api
        image: your-dockerhub/llm-api:latest
        ports:
        - containerPort: 8000

使用 Nginx 做反向代理与负载均衡

upstream backend {
    least_conn;
    server llm-api-0:8000;
    server llm-api-1:8000;
    server llm-api-2:8000;
}

server {
    listen 80;

    location / {
        proxy_pass http://backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

监控与日志收集（Prometheus + Grafana）

使用 starlette_exporter 集成 Prometheus 指标：

pip install starlette-exporter

from starlette_exporter import PrometheusMiddleware, handle_metrics

app.add_middleware(PrometheusMiddleware)
app.add_route("/metrics", handle_metrics)

访问 /metrics 即可获取指标，供 Prometheus 抓取。

性能优化与最佳实践

实战案例研究

案例一：多租户大模型服务

通过 URL 路径或子域名识别租户 ID，动态加载模型：

@app.post("/{tenant_id}/chat")
async def tenant_chat(tenant_id: str, request: ChatRequest):
    model = load_tenant_model(tenant_id)
    response = model.generate(request.prompt)
    return {"response": response}

案例二：支持多种模型（GPT、Llama、ChatGLM）

model_map = {
    "gpt": GPTModel(),
    "llama": LlamaModel(),
    "chatglm": ChatGLMModel()
}

@app.post("/chat")
async def chat(model: str, request: ChatRequest):
    if model not in model_map:
        raise HTTPException(404, "Model not supported")
    return {"response": model_map[model].generate(request.prompt)}

案例三：企业级聊天机器人平台（Chatbot Platform）

一、项目概述

本案例旨在构建一个企业级聊天机器人平台，集成多个 AI 模块，包括：

• NLU（自然语言理解）模块：用于意图识别和槽位提取；
• 对话管理模块（Dialogue Manager）：管理多轮对话状态；
• 知识库检索模块：基于用户问题从数据库或向量库中查找答案；
• 外部接口调用模块：如天气、订单查询等第三方服务；
• 统一 API 接口层：对外暴露 /chatbot 接口，供前端或其他系统调用。

我们使用 FastAPI + Python 构建后端服务，并提供完整的部署指南。

二、项目结构设计

enterprise-chatbot/
├── app/
│   ├── main.py                       # FastAPI 入口
│   ├── api/
│   │   └── chatbot.py              # 对外 API 接口
│   ├── services/
│   │   ├── nlu_service.py          # NLU 模块封装
│   │   ├── dialogue_manager.py     # 对话状态管理
│   │   ├── knowledge_base.py       # 知识库检索
│   │   ├── external_api.py         # 外部服务调用
│   │   └── chatbot_service.py      # 整合所有模块的主流程
│   ├── models/
│   │   └── request_models.py       # Pydantic 数据模型定义
│   ├── utils/
│   │   └── logger.py               # 日志工具
│   └── config.py                   # 配置文件
├── requirements.txt                # 依赖包
├── Dockerfile                      # 容器化配置
└── README.md

三、核心代码实现

1. 请求数据模型定义（models/request_models.py）

from pydantic import BaseModel
from typing import Optional, Dict

class ChatbotRequest(BaseModel):
    user_id: str
    message: str
    session_id: Optional[str] = None

class ChatbotResponse(BaseModel):
    reply: str
    intent: str
    confidence: float
    session_id: str
    metadata: Dict = {}

2. NLU 服务模拟（services/nlu_service.py）

class NLUService:
    def recognize_intent(self, text: str) -> dict:
        # 实际应调用 BERT/NLU 模型进行意图识别
        if "天气" in text:
            return {"intent": "weather", "confidence": 0.95}
        elif "订单" in text:
            return {"intent": "order_status", "confidence": 0.85}
        else:
            return {"intent": "greeting", "confidence": 0.7}

nlu_service = NLUService()

3. 对话管理器（services/dialogue_manager.py）

class DialogueManager:
    def __init__(self):
        self.sessions = {}

    def update_state(self, session_id: str, state: dict):
        self.sessions[session_id] = state

    def get_state(self, session_id: str) -> dict:
        return self.sessions.get(session_id, {})

dialogue_manager = DialogueManager()

4. 知识库检索（services/knowledge_base.py）

import faiss
import numpy as np

class KnowledgeBaseService:
    def __init__(self):
        # 模拟 FAISS 向量库
        self.index = faiss.IndexFlatL2(768)
        self.documents = ["你好，我是客服助手。", "我们的营业时间是每天上午9点到晚上8点。"]

    def search(self, query_embedding: np.ndarray, top_k=1):
        D, I = self.index.search(query_embedding.reshape(1, -1), top_k)
        return [self.documents[i] for i in I[0]]

kb_service = KnowledgeBaseService()

5. 外部接口调用（services/external_api.py）

import requests

class ExternalAPIService:
    def get_weather(self, city: str):
        # 示例调用天气 API
        try:
            res = requests.get(f"https://api.weather.com/city/{city}")
            return res.json().get("temperature")
        except Exception as e:
            return f"获取天气失败: {str(e)}"

external_api = ExternalAPIService()

6. 聊天机器人服务整合（services/chatbot_service.py）

from .nlu_service import nlu_service
from .dialogue_manager import dialogue_manager
from .knowledge_base import kb_service
from .external_api import external_api

class ChatbotService:
    def process(self, request):
        session_id = request.session_id or f"sess_{request.user_id}"
        intent_info = nlu_service.recognize_intent(request.message)
        intent = intent_info["intent"]
        confidence = intent_info["confidence"]

        # 根据意图执行不同逻辑
        if intent == "weather":
            response = f"北京今天的温度是 {external_api.get_weather('Beijing')}°C"
        elif intent == "order_status":
            response = f"您的订单正在处理中，请稍候..."
        else:
            response = kb_service.search(np.random.rand(768))  # 模拟 embedding 查询

        # 更新对话状态
        dialogue_manager.update_state(session_id, {
            "last_intent": intent,
            "last_message": request.message
        })

        return {
            "reply": response,
            "intent": intent,
            "confidence": confidence,
            "session_id": session_id,
            "metadata": {}
        }

chatbot_service = ChatbotService()

7. API 接口（api/chatbot.py）

from fastapi import APIRouter
from ..models.request_models import ChatbotRequest, ChatbotResponse
from ..services.chatbot_service import chatbot_service

router = APIRouter(prefix="/chatbot")

@router.post("/", response_model=ChatbotResponse)
async def chatbot_endpoint(req: ChatbotRequest):
    result = chatbot_service.process(req)
    return result

8. 主入口（main.py）

from fastapi import FastAPI
from app.api.chatbot import router as chatbot_router

app = FastAPI(title="Enterprise Chatbot Platform")

app.include_router(chatbot_router)

@app.on_event("startup")
def on_startup():
    print("Chatbot service started...")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

四、部署指南

1. 安装依赖

pip install fastapi uvicorn pydantic faiss-cpu python-jose[cryptography] requests

如需 GPU 支持可安装 faiss-gpu

2. 本地运行

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

访问 http://localhost:8000/docs 查看自动生成的 Swagger 文档。

3. 使用 Docker 容器化部署

Dockerfile

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

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

构建镜像并运行：

docker build -t enterprise-chatbot .
docker run -p 8000:8000 enterprise-chatbot

4. Kubernetes 部署（可选）

提供一个简化的 Deployment 和 Service YAML 文件：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: chatbot
spec:
  replicas: 3
  selector:
    matchLabels:
      app: chatbot
  template:
    metadata:
      labels:
        app: chatbot
    spec:
      containers:
      - name: chatbot
        image: your-dockerhub/enterprise-chatbot:latest
        ports:
        - containerPort: 8000
---
apiVersion: v1
kind: Service
metadata:
  name: chatbot-service
spec:
  selector:
    app: chatbot
  ports:
    - protocol: TCP
      port: 80
      targetPort: 8000

应用配置：

kubectl apply -f deployment.yaml

五、性能优化建议

六、测试示例（使用 curl）

curl -X POST http://localhost:8000/chatbot/ \
-H "Content-Type: application/json" \
-d '{
    "user_id": "U12345",
    "message": "今天北京天气怎么样？"
}'

输出示例：

{
  "reply": "北京今天的温度是 22°C",
  "intent": "weather",
  "confidence": 0.95,
  "session_id": "sess_U12345",
  "metadata": {}
}

七、结语

通过本案例，你已经掌握了一个完整的企业级聊天机器人平台搭建方法，涵盖从模块划分、服务整合、API 设计到容器化部署的全流程。

如果你希望将其扩展为支持多租户、多模型、插件化架构的 AI 中台系统，可以在此基础上继续演进。

未来展望

Serverless 架构下的大模型服务

• AWS Lambda、Azure Functions 支持 Python，可用于部署轻量级推理接口。
• 结合模型压缩（如 ONNX、量化）降低冷启动延迟。

自动扩缩容与弹性调度

• Kubernetes HPA + Prometheus 指标实现自动伸缩。
• 使用 Ray、Celery 分布式任务队列处理批量推理。

AI 中间件平台的发展趋势

• 将大模型抽象为“能力即服务”（Capability-as-a-Service）
• 统一接入 OpenAI、Anthropic、百川、通义千问等多平台模型
• 提供可视化编排界面，支持 Prompt 编写、插件调用、链式推理等高级功能

FastAPI 作为现代 Web 框架的代表，凭借其简洁易用、高性能、强类型等特性，在构建大模型 API 服务方面展现出巨大潜力。本文不仅介绍了其核心原理，还提供了丰富的实战代码与部署方案，帮助你快速搭建一个企业级的大模型服务。

如果你希望更深入地了解如何构建 AI 应用平台，欢迎关注《AI大模型应知应会100篇》专栏系列文章！

📘 参考文档

• FastAPI 官方文档
• Uvicorn 文档
• OpenAI API 文档
• HuggingFace Transformers 文档