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） - データベースや API 呼び出しなどの I/O 待ちタスクには async def を使用し、スループットを最大化してください。
• ページ設定を行う（Streamlit） - プロット用のスペースが必要な科学ダッシュボードでは、st.set_page_config(layout="wide") を使用してください。
• 例外を処理する - FastAPI の HTTPException を使用して、明確なエラーコード（400、404、500）をユーザーに返してください。
• モジュール化する - 科学計算のロジックは別のファイルまたはパッケージにまとめ、API と UI の両方からインポートしてください。

❌ 禁止事項

• UI スレッドで重い処理を実行しない - Streamlit では、1 秒を超える関数はキャッシュしないと、UI の応答性が損なわれます。
• 非同期ループをブロックしない（FastAPI） - 関数が CPU 負荷の高い処理（重い NumPy 計算など）を行う場合は、async def ではなく通常の 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 ワーカー

本番環境では、より多くのリクエストを処理できるよう、複数のワーカーで実行してください。

uvicorn main:app --workers 4

2. Streamlit：st.cache_resource

データベース接続や ML モデルなど、ユーザーやセッションをまたいでメモリ上に保持するオブジェクトには 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 では、100ms を超える計算をすべてキャッシュします
4. 非同期処理を適切に使う - FastAPI の非同期処理は I/O に適していますが、CPU 負荷の高いタスクは同期処理にします
5. それぞれを個別にテストする - FastAPI のエンドポイントは httpx または requests でテストし、Streamlit の UI は手動でテストします
6. API をドキュメント化する - FastAPI はドキュメントを自動生成しますが、Pydantic モデルには docstring を追加します
7. エラーを適切に処理する - どちらのフレームワークにも優れたエラー処理機能があるため、活用します
8. パフォーマンスを監視する - FastAPI の組み込みメトリクスと Streamlit の実行時間表示を使用します

---

FastAPI + Streamlit のスタックは、科学計算における「最後の一歩」です。生のコードを利用しやすいツールへと変換し、モデルを世界中の人々に役立つものにします。