fastapi-streamlit

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

FastAPI & Streamlit - Deployment & Interaction

FastAPI & Streamlit - 部署与交互

This combination allows scientists to move from a Jupyter Notebook to a production-ready system. FastAPI handles the backend (model serving, data processing), while Streamlit provides the frontend (interactive widgets, real-time plotting).

该组合能帮助科研人员从 Jupyter Notebook 过渡到生产级系统。FastAPI 负责后端部分（模型服务、数据处理），而 Streamlit 则提供前端交互（交互式组件、实时绘图）。

FIRST: Verify Prerequisites

第一步：检查前置依赖

bash

pip install fastapi uvicorn streamlit pydantic

bash

pip install fastapi uvicorn streamlit pydantic

When to Use

适用场景

FastAPI:

Serving Machine Learning models as REST APIs.
Creating microservices for heavy scientific computations.
Building backends that require high concurrency (async/await).
Automatically generating API documentation (Swagger/Redoc).

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

Streamlit:

Building interactive dashboards for data exploration.
Creating "Apps" to demonstrate scientific results to non-technical stakeholders.
Rapid prototyping of UIs for internal tools.
Visualizing complex datasets with interactive sliders, maps, and charts.

构建用于数据探索的交互式仪表板。
创建向非技术利益相关者展示科研成果的「应用」。
为内部工具快速原型化 UI。
通过交互式滑块、地图和图表可视化复杂数据集。

Reference Documentation

参考文档

FastAPI docs: https://fastapi.tiangolo.com/
Streamlit docs: https://docs.streamlit.io/

Search patterns:

fastapi.app

pydantic.BaseModel

st.slider

st.cache_data

st.sidebar

FastAPI 文档：https://fastapi.tiangolo.com/
Streamlit 文档：https://docs.streamlit.io/

常用搜索关键词：

fastapi.app

pydantic.BaseModel

st.slider

st.cache_data

st.sidebar

Core Principles

核心原理

FastAPI: Type Safety and Async

FastAPI：类型安全与异步特性

FastAPI is built on Pydantic for data validation and Starlette for web capabilities. Every input is validated against Python type hints. It is one of the fastest Python frameworks thanks to async/await.

FastAPI 基于 Pydantic 实现数据验证，基于 Starlette 提供 Web 能力。所有输入都会通过 Python 类型提示进行验证。得益于 async/await 特性，它是速度最快的 Python 框架之一。

Streamlit: Execution Model

Streamlit：执行模型

Streamlit scripts run from top to bottom every time a user interacts with a widget. It uses a "magic" caching system to prevent expensive scientific functions from re-running unnecessarily.

每当用户与组件交互时，Streamlit 脚本都会从头至尾重新运行。它采用「魔法」缓存系统，避免耗时的科学计算函数重复执行。

Quick Reference

快速参考

Installation

安装

bash

pip install fastapi uvicorn streamlit pydantic

bash

pip install fastapi uvicorn streamlit pydantic

Standard Imports

标准导入

python

undefined

python

undefined

FastAPI

from fastapi import FastAPI, HTTPException from pydantic import BaseModel

Streamlit

import streamlit as st import requests # To communicate with FastAPI

undefined

import streamlit as st import requests # To communicate with FastAPI

undefined

Basic Pattern - FastAPI Model Server

基础模式 - FastAPI 模型服务

python

undefined

python

undefined

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}

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

undefined

undefined

Basic Pattern - Streamlit Dashboard

基础模式 - Streamlit 仪表板

python

undefined

python

undefined

main_ui.py

import streamlit as st import pandas as pd

st.title("Scientific Data Explorer")

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

undefined

undefined

Critical Rules

关键规则

✅ DO

✅ 建议做法

Use Pydantic Schemas (FastAPI) - Always define your API inputs and outputs using classes inheriting from
```
BaseModel
```
.
Use st.cache_data (Streamlit) - Wrap heavy data loading or heavy math functions with
```
@st.cache_data
```
to keep the UI responsive.
Use Async/Await (FastAPI) - For I/O bound tasks (database, API calls), use
```
async def
```
to maximize throughput.
Set Page Config (Streamlit) - Use
```
st.set_page_config(layout="wide")
```
for scientific dashboards that need space for plots.
Handle Exceptions - Use FastAPI's
```
HTTPException
```
to return clear error codes (400, 404, 500) to the user.
Modularize - Keep your scientific logic in a separate file/package, imported by both API and UI.

使用 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 模块导入使用。

❌ DON'T

❌ 不建议做法

Don't Run Heavy Logic in UI Thread - In Streamlit, if a function takes >1s, it must be cached or the UI will feel broken.
Don't Block the Async Loop (FastAPI) - If a function is CPU-intensive (e.g., heavy NumPy math), use standard
```
def
```
instead of
```
async def
```
; FastAPI will run it in a thread pool.
Don't Store Sensitive Data in UI Code - Use environment variables or
```
.streamlit/secrets.toml
```
.
Don't Over-nest Widgets - Streamlit's "top-down" execution gets confusing if the UI logic is too complex.

不要在 UI 线程中执行耗时逻辑 - 在 Streamlit 中，如果函数执行时间超过 1 秒，必须进行缓存，否则 UI 会出现卡顿。
不要阻塞异步循环（FastAPI） - 如果函数是 CPU 密集型（如复杂 NumPy 计算），使用标准
```
def
```
而非
```
async def
```
；FastAPI 会将其放入线程池执行。
不要在 UI 代码中存储敏感数据 - 使用环境变量或
```
.streamlit/secrets.toml
```
存储敏感信息。
不要过度嵌套组件 - 如果 UI 逻辑过于复杂，Streamlit 的「自上而下」执行模式会变得难以理解。

Anti-Patterns (NEVER)

反模式（绝对避免）

python

undefined

python

undefined

❌ 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

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()

undefined

@st.cache_data def load_massive_data(): return pd.read_csv("massive_data.csv")

data = load_massive_data()

undefined

FastAPI: Advanced Features

FastAPI: 高级特性

Dependency Injection (e.g., Database/Model loading)

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

python

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}

python

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}

Background Tasks (Long-running computations)

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

python

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"}

python

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: Layout and Interaction

Streamlit: 布局与交互

Multi-column and Sidebars

多列布局与侧边栏

python

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)

python

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)

Session State (Keeping track of user data)

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

python

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)}")

python

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)}")

Practical Workflows

实用工作流

1. Scientific Model Serving (FastAPI + PyTorch)

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

python

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()}

python

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. Interactive Data Cleaning Tool (Streamlit + Polars)

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

python

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")

python

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. Real-time Monitoring App

3. 实时监控应用

python

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)

python

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)

Performance Optimization

性能优化

1. FastAPI: Uvicorn Workers

1. FastAPI: Uvicorn 多进程

For production, run with multiple workers to handle more requests.

bash

uvicorn main:app --workers 4

生产环境中，使用多进程运行以处理更多请求。

bash

uvicorn main:app --workers 4

2. Streamlit: st.cache_resource

Use

cache_resource

for objects that should stay in memory across users/sessions, like Database connections or ML models.

python

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

使用

cache_resource

缓存跨用户/会话的内存对象，如数据库连接或机器学习模型。

python

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

3. Streamlit: PyArrow

Streamlit uses Apache Arrow for data exchange. Ensuring your data is in Arrow-compatible formats (like Polars or Pandas) makes UI rendering instant.

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

Common Pitfalls and Solutions

常见问题与解决方案

FastAPI: Input Data Validation Errors

FastAPI: 输入数据验证错误

If a user sends a string where a float is expected, FastAPI returns 422 Unprocessable Entity.

python

undefined

如果用户在需要浮点数的位置传入字符串，FastAPI 会返回 422 不可处理实体错误。

python

undefined

✅ Solution: Wrap Pydantic models in try-except if needed,

but usually, let FastAPI handle it and customize exception_handlers.

undefined

undefined

Streamlit: The "Double Rerun"

Streamlit: 「重复重运行」问题

Sometimes widgets trigger multiple reruns.

python

undefined

有时组件会触发多次重运行。

python

undefined

✅ 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")

undefined

with st.form("my_form"): # ... inputs ... submitted = st.form_submit_button("Submit")

undefined

Deployment Port Conflict

部署端口冲突

By default, Streamlit uses 8501 and FastAPI (Uvicorn) uses 8000.

python

undefined

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

python

undefined

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

undefined

undefined

Best Practices

最佳实践

Separate Concerns - Keep scientific logic separate from API/UI code for reusability
Type Everything - Use Pydantic models for all FastAPI endpoints to catch errors early
Cache Aggressively - In Streamlit, cache any computation that takes >100ms
Use Async Wisely - FastAPI async is great for I/O, but CPU-bound tasks should be sync
Test Both Separately - Test your FastAPI endpoints with
```
httpx
```
or
```
requests
```
, test Streamlit UI manually
Document APIs - FastAPI auto-generates docs, but add docstrings to your Pydantic models
Handle Errors Gracefully - Both frameworks have good error handling; use it
Monitor Performance - Use FastAPI's built-in metrics and Streamlit's execution time display

The FastAPI + Streamlit stack is the "Last Mile" of scientific computing. It transforms raw code into accessible tools, making your models useful to the rest of the world.

关注点分离 - 将科学计算逻辑与 API/UI 代码分离，提升可复用性
类型标注 - 为所有 FastAPI 端点使用 Pydantic 模型，提前捕获错误
积极缓存 - 在 Streamlit 中，对所有执行时间超过 100ms 的计算进行缓存
合理使用异步 - FastAPI 异步特性适用于 I/O 密集型任务，CPU 密集型任务应使用同步函数
独立测试 - 使用
```
httpx
```
或
```
requests
```
测试 FastAPI 端点，手动测试 Streamlit UI
API 文档 - FastAPI 会自动生成文档，同时为 Pydantic 模型添加文档字符串
优雅处理错误 - 两个框架都具备完善的错误处理机制，充分利用它们
性能监控 - 使用 FastAPI 的内置指标和 Streamlit 的执行时间显示功能监控性能

FastAPI + Streamlit 组合是科学计算的「最后一公里」工具。它能将原始代码转化为可访问的工具，让你的模型真正为他人所用。