用Streamlit+FastAPI打造机器学习应用：从本地调试到Docker部署全流程

在数据科学和机器学习领域，快速构建和部署应用一直是个挑战。传统方法需要前端开发、后端API和部署运维的多方面技能，而Streamlit和FastAPI的组合彻底改变了这一局面。本文将带你从零开始，通过一个汽车评估案例，完整演示如何构建前后端分离的机器学习应用，并最终通过Docker容器化部署。

1. 技术栈选择与项目架构设计

Streamlit和FastAPI的组合为数据科学家提供了全栈开发能力。Streamlit以其极简的Python API著称，能让开发者用几行代码构建交互式Web界面。FastAPI则是高性能的Python Web框架，特别适合构建机器学习API服务。

我们的汽车评估应用将采用前后端分离架构：

• 前端：Streamlit构建用户界面，收集用户输入并展示预测结果
• 后端：FastAPI提供RESTful API，处理模型推理请求
• 部署：Docker容器化，使用Docker Compose编排服务

这种架构的优势在于：

• 前后端开发完全解耦，可独立迭代
• 容器化部署确保环境一致性
• 易于扩展为微服务架构

2. 环境准备与依赖安装

开始编码前，我们需要设置开发环境。推荐使用Python 3.8+和虚拟环境：

# 创建并激活虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows

# 安装核心依赖
pip install streamlit fastapi uvicorn pandas scikit-learn

对于汽车评估案例，我们将使用UCI的Car Evaluation数据集。可以通过以下代码快速加载：

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier

# 加载数据集
url = "https://archive.ics.uci.edu/ml/machine-learning-databases/car/car.data"
columns = ["buying", "maint", "doors", "persons", "lug_boot", "safety", "class"]
data = pd.read_csv(url, names=columns)

# 数据预处理
X = pd.get_dummies(data.drop("class", axis=1))
y = data["class"].astype("category").cat.codes

# 训练简单模型
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
model = RandomForestClassifier().fit(X_train, y_train)

3. 构建FastAPI后端服务

FastAPI后端负责模型推理，我们将创建一个标准的RESTful API：

# api/main.py
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import pandas as pd
import joblib

app = FastAPI()

# 定义请求体模型
class CarFeatures(BaseModel):
    buying: str
    maint: str
    doors: str
    persons: str
    lug_boot: str
    safety: str

# 加载预训练模型
model = joblib.load("model.pkl")

@app.post("/predict")
async def predict(features: List[CarFeatures]):
    # 转换输入为DataFrame
    df = pd.DataFrame([f.dict() for f in features])
    df = pd.get_dummies(df)
    
    # 确保所有训练时的列都存在
    for col in model.feature_names_in_:
        if col not in df.columns:
            df[col] = 0
    
    # 预测并返回结果
    predictions = model.predict(df[model.feature_names_in_])
    return {"predictions": predictions.tolist()}

启动API服务：

uvicorn api.main:app --reload --port 8001

提示：在生产环境中，应该添加适当的错误处理、日志记录和输入验证

4. 开发Streamlit前端界面

Streamlit前端提供用户友好的交互界面：

# app.py
import streamlit as st
import requests

# 配置页面
st.set_page_config(page_title="汽车评估系统", layout="wide")
st.title("🚗 汽车评估系统")

# 定义评估等级映射
class_values = {
    0: "不可接受",
    1: "可接受", 
    2: "良好",
    3: "优秀"
}

# 创建表单收集用户输入
with st.form("car_evaluation"):
    st.subheader("请输入汽车特征")
    
    col1, col2 = st.columns(2)
    with col1:
        buying = st.radio("购买价格", ["vhigh", "high", "med", "low"])
        maint = st.radio("维护成本", ["vhigh", "high", "med", "low"])
        doors = st.select_slider("车门数量", options=["2", "3", "4", "5more"])
    
    with col2:
        persons = st.select_slider("载客量", options=["2", "4", "more"])
        lug_boot = st.select_slider("行李厢大小", options=["small", "med", "big"])
        safety = st.select_slider("安全等级", options=["low", "med", "high"])
    
    submitted = st.form_submit_button("评估")

# 处理表单提交
if submitted:
    inputs = {
        "inputs": [{
            "buying": buying,
            "maint": maint,
            "doors": doors,
            "persons": persons,
            "lug_boot": lug_boot,
            "safety": safety
        }]
    }
    
    try:
        response = requests.post(
            "http://localhost:8001/predict",
            json=inputs
        )
        prediction = class_values[response.json()["predictions"][0]]
        st.success(f"评估结果: {prediction}")
    except Exception as e:
        st.error(f"评估失败: {str(e)}")

启动前端应用：

streamlit run app.py

5. Docker容器化部署

为了确保应用在不同环境中的一致性，我们使用Docker进行容器化。首先为每个服务创建Dockerfile。

前端Dockerfile:

# frontend/Dockerfile
FROM python:3.9-slim

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

COPY . .

EXPOSE 8501
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

后端Dockerfile:

# backend/Dockerfile
FROM python:3.9-slim

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

COPY . .

EXPOSE 8001
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8001"]

docker-compose.yml:

version: '3.8'

services:
  frontend:
    build: ./frontend
    ports:
      - "8501:8501"
    depends_on:
      - backend
    environment:
      - BACKEND_URL=http://backend:8001

  backend:
    build: ./backend
    ports:
      - "8001:8001"

启动所有服务：

docker-compose up --build

6. 性能优化与扩展

随着应用规模增长，我们需要考虑以下优化：

1. API性能：

   • 添加缓存层（如Redis）
   • 实现异步端点
   • 启用Gzip压缩

2. 前端体验：

   • 使用Streamlit缓存减少重复计算
   • 添加加载状态指示器
   • 实现表单验证

3. 部署架构：

   • 添加Nginx反向代理
   • 配置负载均衡
   • 设置健康检查

# 示例：带缓存的FastAPI端点
from fastapi import FastAPI, Depends
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from redis import asyncio as aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://redis")
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

@app.get("/expensive-operation")
@cache(expire=60)  # 缓存60秒
async def expensive_operation():
    # 耗时计算
    return {"result": "data"}

7. 监控与日志

生产环境需要完善的监控系统：

• Prometheus：收集指标数据
• Grafana：可视化监控数据
• ELK Stack：集中日志管理

FastAPI内置了Prometheus指标端点：

from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app)

对于Streamlit，可以添加自定义日志：

import logging
from streamlit.logger import get_logger

logger = get_logger(__name__)
logger.setLevel(logging.INFO)

def some_function():
    try:
        # 业务逻辑
        logger.info("操作成功")
    except Exception as e:
        logger.error(f"操作失败: {str(e)}")
        raise

8. 安全最佳实践

确保应用安全的关键措施：

1. API安全：

   • 启用CORS限制
   • 实现认证/授权
   • 使用HTTPS

2. 输入验证：

   • 严格校验所有用户输入
   • 防范SQL注入
   • 限制请求大小

3. 依赖安全：

   • 定期更新依赖
   • 扫描漏洞
   • 锁定依赖版本

FastAPI安全配置示例：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

Streamlit安全提示：

# 禁用危险功能
st.set_option("server.enableCORS", False)
st.set_option("server.enableXsrfProtection", True)

9. 持续集成与部署

自动化部署流程能显著提高效率：

1. GitHub Actions示例：

name: Deploy to Production

on:
  push:
    branches: [ main ]

jobs:
  build-and-deploy:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    
    - name: Build and push Docker images
      run: |
        docker-compose -f docker-compose.prod.yml build
        echo "$DOCKER_PASSWORD" | docker login -u "$DOCKER_USERNAME" --password-stdin
        docker-compose -f docker-compose.prod.yml push
    
    - name: Deploy to server
      uses: appleboy/ssh-action@master
      with:
        host: ${{ secrets.SSH_HOST }}
        username: ${{ secrets.SSH_USERNAME }}
        key: ${{ secrets.SSH_KEY }}
        script: |
          docker-compose -f docker-compose.prod.yml pull
          docker-compose -f docker-compose.prod.yml up -d

2. 部署检查清单：
   • 测试覆盖率 >80%
   • 通过所有CI流水线
   • 完成负载测试
   • 备份现有部署

10. 实际案例：汽车评估系统增强版

让我们扩展基础功能，添加更多实用特性：

# 增强版app.py
import streamlit as st
import requests
import pandas as pd
import plotly.express as px

# 添加多页面支持
st.sidebar.title("导航")
page = st.sidebar.radio("", ["评估", "数据分析", "关于"])

if page == "评估":
    # 原有评估逻辑...
elif page == "数据分析":
    st.title("汽车数据洞察")
    
    # 加载数据集
    url = "https://archive.ics.uci.edu/ml/machine-learning-databases/car/car.data"
    columns = ["buying", "maint", "doors", "persons", "lug_boot", "safety", "class"]
    data = pd.read_csv(url, names=columns)
    
    # 交互式可视化
    st.subheader("数据分布")
    col = st.selectbox("选择分析维度", columns[:-1])
    fig = px.histogram(data, x=col, color="class", barmode="group")
    st.plotly_chart(fig)
    
    # 相关性分析
    st.subheader("特征相关性")
    numeric_data = pd.get_dummies(data)
    fig = px.imshow(numeric_data.corr())
    st.plotly_chart(fig)
    
elif page == "关于":
    st.title("关于汽车评估系统")
    st.markdown("""
    ### 技术栈
    - 前端: Streamlit
    - 后端: FastAPI
    - 部署: Docker + Compose
    
    ### 数据集
    使用UCI机器学习仓库的[Car Evaluation Dataset](https://archive.ics.uci.edu/ml/datasets/Car+Evaluation)
    """)

这个增强版展示了如何利用Streamlit快速构建数据分析功能，为用户提供更深入的洞察。