fastapi-streamlit

FastAPI 与 Streamlit — 部署与交互

这种组合能帮助科学家从 Jupyter Notebook 迈向可用于生产环境的系统。FastAPI 负责后端（模型服务、数据处理），而 Streamlit 提供前端（交互式控件、实时绘图）。

首先：验证先决条件

pip install fastapi uvicorn streamlit pydantic

适用场景

FastAPI：

• 将机器学习模型作为 REST API 提供服务。
• 为繁重的科学计算创建微服务。
• 构建需要高并发能力（async/await）的后端。
• 自动生成 API 文档（Swagger/Redoc）。

Streamlit：

• 构建用于数据探索的交互式仪表板。
• 创建用于向非技术相关方展示科学成果的“应用”。
• 为内部工具快速构建 UI 原型。
• 使用交互式滑块、地图和图表可视化复杂数据集。

参考文档

• FastAPI 文档：https://fastapi.tiangolo.com/
• Streamlit 文档：https://docs.streamlit.io/
• 搜索模式：fastapi.app、pydantic.BaseModel、st.slider、st.cache_data、st.sidebar

核心原则

FastAPI：类型安全与异步

FastAPI 基于用于数据验证的 Pydantic 和用于 Web 功能的 Starlette 构建。每个输入都会根据 Python 类型提示进行验证。得益于 async/await，它是速度最快的 Python 框架之一。

Streamlit：执行模型

每当用户与控件交互时，Streamlit 脚本都会从上到下运行。它使用一种“魔法”缓存系统，避免不必要地重复运行耗时的科学计算函数。

快速参考

安装

pip install fastapi uvicorn streamlit pydantic

标准导入

# FastAPI
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# Streamlit
import streamlit as st
import requests # To communicate with FastAPI

基本模式 — FastAPI 模型服务

# main_api.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ModelInput(BaseModel):
    temperature: float
    pressure: float

@app.post("/predict")
def predict(data: ModelInput):
    # Imagine a complex physical model here
    result = data.temperature * 0.5 + data.pressure * 0.2
    return {"prediction": result}

# Run with: uvicorn main_api:app --reload

基本模式 — Streamlit 仪表板

# main_ui.py
import streamlit as st
import pandas as pd

st.title("Scientific Data Explorer")

# 1. Widgets for input
val = st.slider("Select a threshold", 0.0, 100.0, 50.0)

# 2. Logic/Processing
df = pd.DataFrame({"x": range(100), "y": [x**2 for x in range(100)]})
filtered_df = df[df["y"] > val]

# 3. Visualization
st.line_chart(filtered_df)
st.write(f"Points above threshold: {len(filtered_df)}")

# Run with: streamlit run main_ui.py

关键规则

✅ 应该做

• 使用 Pydantic 模式（FastAPI） — 始终使用继承自 BaseModel 的类定义 API 输入和输出。
• 使用 st.cache_data（Streamlit） — 使用 @st.cache_data 包装耗时的数据加载或复杂数学计算函数，保持 UI 响应流畅。
• 使用 Async/Await（FastAPI） — 对于 I/O 密集型任务（数据库、API 调用），使用 async def 以最大化吞吐量。
• 设置页面配置（Streamlit） — 对于需要充足绘图空间的科学仪表板，使用 st.set_page_config(layout="wide")。
• 处理异常 — 使用 FastAPI 的 HTTPException 向用户返回明确的错误代码（400、404、500）。
• 模块化 — 将科学计算逻辑放在单独的文件或包中，并由 API 和 UI 共同导入。

❌ 不应该做

• 不要在 UI 线程中运行繁重逻辑 — 在 Streamlit 中，如果函数耗时超过 1 秒，就必须缓存，否则 UI 会显得失灵。
• 不要阻塞异步循环（FastAPI） — 如果函数属于 CPU 密集型（例如，繁重的 NumPy 运算），请使用标准的 def，而不是 async def；FastAPI 会在线程池中运行它。
• 不要在 UI 代码中存储敏感数据 — 使用环境变量或 .streamlit/secrets.toml。
• 不要过度嵌套控件 — 如果 UI 逻辑过于复杂，Streamlit 的“从上到下”执行方式会让人难以理清。

反模式（绝不采用）

# ❌ BAD: Manual JSON parsing in FastAPI
# @app.post("/data")
# def handle_data(raw_json: dict):
#     val = raw_json.get("value") # No validation!

# ✅ GOOD: Pydantic validation
class DataPoint(BaseModel):
    value: float

@app.post("/data")
def handle_data(data: DataPoint):
    return data.value # Guaranteed to be a float

# ❌ BAD: Loading data in every Streamlit rerun
# data = pd.read_csv("massive_data.csv") # Re-reads every time you move a slider!

# ✅ GOOD: Caching
@st.cache_data
def load_massive_data():
    return pd.read_csv("massive_data.csv")

data = load_massive_data()

FastAPI：高级功能

依赖注入（例如，数据库/模型加载）

from functools import lru_cache

@lru_cache()
def load_model():
    # Load your PyTorch or Scikit-learn model here
    return MyHeavyModel().load("weights.pt")

@app.get("/status")
def get_status(model = Depends(load_model)):
    return {"model_version": model.version}

后台任务（长时间运行的计算）

from fastapi import BackgroundTasks

def solve_pde_task(params):
    # Long FEniCS simulation
    pass

@app.post("/run-sim")
def run_simulation(params: Params, background_tasks: BackgroundTasks):
    background_tasks.add_task(solve_pde_task, params)
    return {"message": "Simulation started in background"}

Streamlit：布局与交互

多列与侧边栏

st.sidebar.header("Settings")
mode = st.sidebar.selectbox("Model Mode", ["Fast", "Accurate"])

col1, col2 = st.columns(2)

with col1:
    st.header("Input Parameters")
    temp = st.number_input("Temperature (K)")

with col2:
    st.header("Results Visualization")
    # Plotly/Matplotlib chart
    st.plotly_chart(fig)

会话状态（跟踪用户数据）

if 'results_history' not in st.session_state:
    st.session_state.results_history = []

if st.button("Run Experiment"):
    res = run_model()
    st.session_state.results_history.append(res)

st.write(f"History length: {len(st.session_state.results_history)}")

实用工作流

1. 科学模型服务（FastAPI + PyTorch）

import torch
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
model = torch.load("model.pth")
model.eval()

class PredictionRequest(BaseModel):
    features: list[float]

@app.post("/v1/predict")
def get_prediction(req: PredictionRequest):
    input_tensor = torch.tensor([req.features])
    with torch.no_grad():
        output = model(input_tensor)
    return {"class": output.argmax().item(), "confidence": output.max().item()}

2. 交互式数据清理工具（Streamlit + Polars）

import streamlit as st
import polars as pl

st.title("Data Cleaner")
uploaded_file = st.file_uploader("Choose a CSV file")

if uploaded_file:
    df = pl.read_csv(uploaded_file)
    
    st.write("Original Data Summary", df.describe())
    
    col_to_drop = st.multiselect("Drop columns", df.columns)
    if st.button("Clean Data"):
        df_clean = df.drop(col_to_drop).drop_nulls()
        st.dataframe(df_clean)
        st.download_button("Download Clean CSV", df_clean.write_csv(), "clean.csv")

3. 实时监控应用

import streamlit as st
import time

placeholder = st.empty()

for i in range(100):
    with placeholder.container():
        st.metric("Current Sensor Reading", f"{get_val()} units")
        st.progress(i + 1)
    time.sleep(1)

性能优化

1. FastAPI：Uvicorn Workers

在生产环境中，使用多个 worker 运行，以处理更多请求。

uvicorn main:app --workers 4

2. Streamlit：st.cache_resource

对于应在不同用户/会话之间常驻内存的对象（例如数据库连接或机器学习模型），请使用 cache_resource。

@st.cache_resource
def get_database_connection():
    return create_engine("postgresql://...")

3. Streamlit：PyArrow

Streamlit 使用 Apache Arrow 进行数据交换。确保数据采用 Arrow 兼容的格式（如 Polars 或 Pandas），即可让 UI 即时渲染。

常见陷阱及解决方案

FastAPI：输入数据验证错误

如果用户发送字符串，而预期值为浮点数，FastAPI 会返回 422 Unprocessable Entity。

# ✅ Solution: Wrap Pydantic models in try-except if needed, 
# but usually, let FastAPI handle it and customize exception_handlers.

Streamlit：“双重重新运行”

有时，控件会触发多次重新运行。

# ✅ Solution: Use st.form to group widgets so the script 
# only reruns once when the "Submit" button is clicked.
with st.form("my_form"):
    # ... inputs ...
    submitted = st.form_submit_button("Submit")

部署端口冲突

默认情况下，Streamlit 使用 8501，FastAPI（Uvicorn）使用 8000。

# ✅ Solution: Be explicit in Docker/Compose files about ports.

最佳实践

1. 分离关注点 - 将科学逻辑与 API/UI 代码分开，以便重复使用
2. 为所有内容添加类型 - 所有 FastAPI 端点都使用 Pydantic 模型，以便及早发现错误
3. 积极使用缓存 - 在 Streamlit 中，缓存所有耗时超过 100 ms 的计算
4. 明智地使用异步 - FastAPI 的异步处理非常适合 I/O，但 CPU 密集型任务应使用同步处理
5. 分别测试两者 - 使用 httpx 或 requests 测试 FastAPI 端点，手动测试 Streamlit UI
6. 记录 API 文档 - FastAPI 会自动生成文档，但仍应为 Pydantic 模型添加文档字符串
7. 妥善处理错误 - 两个框架都提供了良好的错误处理机制；请加以利用
8. 监控性能 - 使用 FastAPI 内置的指标和 Streamlit 的执行时间显示

---

FastAPI + Streamlit 技术栈是科学计算的“最后一公里”。它将原始代码转化为易于使用的工具，让其他人也能从中受益。