個人開発を 2014 年から続けてきて、累計 5,000 万 DL のアプリ事業を一人で回しているうちに、社内向け（といっても私一人のチームですが）の BI ダッシュボードを Streamlit と Gemini で組む機会が増えました。Lacrima と Mystery という二つのブログを毎日自動投稿で回し、6 つの壁紙アプリの AdMob 収益と App Store / Google Play のレビューを毎週眺める——これを「素の Streamlit」で組んだ最初の版は、3 週間で限界が来ました。
何が壊れたかというと、ログインがないので URL を知っている人は誰でも見えてしまうこと、Gemini の請求が読みもしないクエリで月 ¥4,000 を超えてきたこと、同じ CSV を二度投げると二度フルで API が走ること、エラーが出ても気づかないこと、の 4 点です。個人実験用に書いた Streamlit と、毎日見るための BI として運用する Streamlit は、コードの 6〜7 割を書き直すくらい設計が違います。この記事はその「書き直した側」の設計メモを、私が実際に組んだ構成ベースで残しておくものです。
なぜ「個人実験用」と「社内 BI」では設計が完全に変わるのか
最初に違いを地図にしておきます。Streamlit + Gemini の入門記事はネット上に大量にありますが、その先で必要になる 5 つのレイヤがほとんど書かれていないので、ここをまず明示しておきます。
- 認証層 — 誰がアクセスできるか。素の Streamlit はゼロ。
- コスト層 — Gemini への課金がユーザー操作に対して線形に増えるのを断ち切る仕組み。
- キャッシュ層 — 同じ問いに二度フル課金されない構造。
- レート制限層 — Gemini のクォータと、ユーザー単位のスロットルを別レイヤで持つ。
- 可観測性層 — 誰が何を聞いて何トークン消費したかを後で追える状態。
個人実験は 1〜5 が全部「なし」でも動きます。BI として毎日同僚や自分の別端末から触る瞬間に、1〜5 の全部が「ない方が異常」になります。順番に組み上げていきます。
認証層: streamlit-authenticator と Auth0 と Cloudflare Access の使い分け
私が個人開発で組む BI は、せいぜい 1〜5 名のチーム規模です。この帯では、選択肢は実質 3 つに絞れます。
streamlit-authenticator(社内 5 名以下・即日運用)
streamlit-authenticator は Python パッケージで、YAML にユーザーとハッシュ済みパスワードを書いて読み込ませるだけで認証画面が出ます。私が個人開発 BI で最初に採用したのはこれです。理由はシンプルで、追加インフラが不要だから。Streamlit Community Cloud にデプロイするなら、Secrets に YAML を貼って終わりです。
# auth.py
import streamlit_authenticator as stauth
import yaml
from yaml.loader import SafeLoader
import streamlit as st
def get_authenticator():
config = yaml.load(st.secrets["AUTH_CONFIG"], Loader=SafeLoader)
return stauth.Authenticate(
config["credentials"],
config["cookie"]["name"],
config["cookie"]["key"],
config["cookie"]["expiry_days"],
)
def login_gate():
auth = get_authenticator()
auth.login(location="main")
if st.session_state.get("authentication_status") is not True:
st.warning("ログインが必要です")
st.stop()
return st.session_state["username"]
落とし穴は 3 つあります。1 つ目は、expiry_days を 30 にすると Cookie ベースで「実質ログイン状態」が長く保持されるので、デバイス紛失時の取り消し手段がないこと。私は 7 日に縮め、加えて週次でパスワードハッシュを更新するスクリプトを cron で走らせています。2 つ目は、YAML が GitHub に漏れた瞬間に終わるので、必ず Streamlit Secrets か外部 KV から読み込む構成にすること。3 つ目は、streamlit-authenticator は MFA を提供しないため、機微データを扱う BI には不向きという点です。
Auth0(社内 10〜30 名・MFA 必要)
人数が 10 を超え、退職などで権限剥奪を確実にやりたい段階になったら Auth0 に上げます。Auth0 は無料枠が 25,000 MAU まであるので、社内 BI の規模なら課金ゼロで運用できます。Streamlit との接続は OIDC で、Authlib を入れて 30 行ほどで実装できます。
# auth_oidc.py
from authlib.integrations.requests_client import OAuth2Session
import streamlit as st
AUTH0_DOMAIN = st.secrets["AUTH0_DOMAIN"]
CLIENT_ID = st.secrets["AUTH0_CLIENT_ID"]
CLIENT_SECRET = st.secrets["AUTH0_CLIENT_SECRET"]
REDIRECT_URI = st.secrets["AUTH0_REDIRECT_URI"]
def login_gate():
if "user" in st.session_state:
return st.session_state["user"]
code = st.query_params.get("code")
if code is None:
sess = OAuth2Session(CLIENT_ID, CLIENT_SECRET, redirect_uri=REDIRECT_URI)
url, _ = sess.create_authorization_url(
f"https://{AUTH0_DOMAIN}/authorize", scope="openid profile email"
)
st.markdown(f"[ログイン]({url})")
st.stop()
sess = OAuth2Session(CLIENT_ID, CLIENT_SECRET, redirect_uri=REDIRECT_URI)
token = sess.fetch_token(
f"https://{AUTH0_DOMAIN}/oauth/token", code=code
)
user = sess.get(f"https://{AUTH0_DOMAIN}/userinfo").json()
st.session_state["user"] = user
st.query_params.clear()
return user
私が Auth0 を採用する場合、Rules でメールドメイン制限（@dolice.design のみ許可など）を必ず仕掛けます。これを忘れると Auth0 はオープンにサインアップを許してしまうので、本番運用前の必須チェック項目です。
Cloudflare Access(社内 BI で最も推奨する)
私の個人開発 BI で現在の本命はこれです。Cloudflare Access は、アプリ側に認証コードをほぼ書かない代わりに、Cloudflare のエッジで認証を完結させます。Streamlit を Cloud Run か自前の Cloudflare Tunnel 経由で公開し、Cloudflare Zero Trust の Application に登録すると、Cloudflare が Google ワンタイムパスや GitHub などの IdP で認証してから初めてオリジンにリクエストが届きます。
採用理由は 3 つです。第一に、認証ロジックを Streamlit 側に持たないので、Streamlit のアップデートで認証が壊れる事故が消えます。第二に、無料枠が 50 ユーザーまであり、個人開発の規模では完全無料です。第三に、Cf-Access-Authenticated-User-Email ヘッダがリクエストごとに付くので、ユーザー識別が確実かつ簡潔になります。
# auth_cf.py
import streamlit as st
def login_gate():
# Cloudflare Access が認証済みユーザーをヘッダで渡す
# Streamlit は HTTP ヘッダを直接読めないので、リバプロ経由で session に書く
headers = st.context.headers if hasattr(st, "context") else {}
email = headers.get("Cf-Access-Authenticated-User-Email")
if not email:
st.error("認証ヘッダが届いていません。Cloudflare Access 経由でアクセスしてください")
st.stop()
return email
私の場合、5 人以下の社内 BI なら streamlit-authenticator、それ以上または機微データなら Cloudflare Access を推奨します。Auth0 は「社外への共有が必須」かつ「Google/GitHub SSO 以外の IdP も使いたい」など、要件が増えた時の選択肢です。
Gemini API のコスト構造を実数値で把握する
ここからがプレミアム記事の本番です。本番運用で最初にぶつかるのは、ほぼ確実に「Gemini の請求がじわじわ増えていくのに、誰のどのクエリが原因か追えない」という現象です。
2026 年 5 月時点の Gemini の主要モデルを、データ分析 BI で使う観点から並べると以下のとおりです。私が個人開発 BI で計測した実値ベースで書きます。
- Gemini 2.5 Flash — 入力 1M トークンあたり約 ¥45、出力約 ¥190。CSV のスキーマ要約と単発分析クエリではこれで 95% 足ります。1 クエリあたり平均 1,200 トークンとして、入出力合わせて ¥0.4 程度。
- Gemini 2.5 Pro — Flash の概ね 8 倍のコスト。長文要約や多段推論で結果に明確な差が出るクエリにだけ使う。1 クエリあたり平均で ¥3〜¥4。
- Gemini 3.x Pro 系 — さらに推論深度が必要な「データから戦略提案」クラスに限定。1 クエリで ¥10〜¥15 のオーダーになるので、トリガーは慎重に。
私の運用基準では、デフォルトは必ず Flash、Pro 以上は ユーザーが明示的に「深く分析」ボタンを押した時のみ起動します。これだけで、月の Gemini 請求が ¥4,000 → ¥600 まで下がったのが実測値です。
# model_router.py
from google import genai
client = genai.Client(api_key=st.secrets["GEMINI_API_KEY"])
MODEL_TIERS = {
"fast": "gemini-2.5-flash",
"deep": "gemini-2.5-pro",
"research": "gemini-3.0-pro-latest",
}
def route(prompt: str, tier: str = "fast", schema=None):
model = MODEL_TIERS[tier]
cfg = {"response_mime_type": "application/json"} if schema else {}
if schema:
cfg["response_schema"] = schema
return client.models.generate_content(
model=model, contents=prompt, config=cfg
)
ティアを文字列キーで持つのは、後で UI 側から st.radio で選ばせるためです。「深く分析」をデフォルト OFF にして、選んだ時にだけ tier を deep に切り替える設計が、私の経験上 BI の請求を一番安定させます。
データ分析クエリの結果キャッシュ設計
「同じ CSV を二度投げたら、二度フル課金される」を断ち切るのがこの層です。私は 3 層キャッシュを組んでいます。
第 1 層: @st.cache_data によるプロセス内メモリキャッシュ。 Streamlit 標準機能で、関数引数のハッシュで結果を保持します。同一ユーザーの同一セッション内で「同じ問い」を繰り返したときに最も効きます。TTL は 1 時間が私の経験上のスイートスポット。
@st.cache_data(ttl=3600, show_spinner="Gemini に問い合わせ中…")
def analyze_schema(schema_summary: str, model_tier: str) -> dict:
resp = route(
prompt=f"次のスキーマから重要傾向を 3 点抽出: {schema_summary}",
tier=model_tier,
schema=ANALYSIS_SCHEMA,
)
return resp.parsed
第 2 層: SHA-256 ベースの永続キャッシュ。 プロセスをまたいでも残るキャッシュです。CSV のスキーマ要約と質問文を結合した SHA-256 をキーに、Cloudflare KV か SQLite に結果を保存します。私は個人開発の小規模 BI では SQLite で十分でした。重要なのは、生 CSV ではなくスキーマ要約をハッシュすること。生 CSV にはユーザー固有データが入るので、ハッシュキャッシュが PII を漏らすリスクがあります。
import hashlib, sqlite3, json
def cache_key(schema_summary: str, question: str, tier: str) -> str:
payload = f"{tier}|{schema_summary}|{question}".encode()
return hashlib.sha256(payload).hexdigest()
def get_or_run(schema_summary: str, question: str, tier: str):
key = cache_key(schema_summary, question, tier)
conn = sqlite3.connect("cache.db")
row = conn.execute(
"SELECT result FROM cache WHERE key=? AND ts > strftime('%s','now')-86400",
(key,),
).fetchone()
if row:
return json.loads(row[0]), True # hit
result = route(prompt=question + "\n\n" + schema_summary, tier=tier).text
conn.execute(
"INSERT OR REPLACE INTO cache(key, ts, result) VALUES(?, strftime('%s','now'), ?)",
(key, json.dumps(result)),
)
conn.commit()
return result, False # miss
第 3 層: Gemini API の Context Caching。 スキーマ要約が大きく（5,000 トークン超）、同一スキーマに対する複数質問が続く場合のみ有効。Gemini の cachedContents を作って、入力トークン課金を 1/4 程度に圧縮できます。BI 用途では「同じダッシュボードで毎朝同じ CSV を投げる」シナリオが該当します。
私の個人開発 BI ではキャッシュヒット率は 42% です。これは Lacrima と Mystery の毎日同じ KPI を眺めるという用途の偏りが大きく、汎用 BI ではもう少し低めに出るはずですが、それでも 20〜30% は出ると見ています。
同時実行制御とユーザー単位のレート制限
Gemini の無料枠は分間 15 リクエスト、有料でも分間 60 や 360 程度が多くのプロジェクトで設定されています。BI を複数ユーザーが同時に触ると、平均的なクリック頻度でも 429 が降ってきます。
対策は 2 段構えで組みます。
第 1 段: Gemini API クォータに対するグローバル・バックオフ。 tenacity で指数バックオフを必ず噛ませます。これがないと、Streamlit の UI が固まって体験が破綻します。
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=1.5, min=2, max=20),
retry=retry_if_exception_type(Exception),
reraise=True,
)
def safe_generate(**kwargs):
return client.models.generate_content(**kwargs)
第 2 段: ユーザー単位のトークンバケット。 「特定ユーザーが連打して他ユーザーを巻き込んで全員 429」を防ぐため、ユーザーごとに分間 10 リクエストの上限を入れます。SQLite か KV にユーザーごとのタイムスタンプリストを持つだけで実装できます。
def user_rate_check(email: str, limit_per_min: int = 10) -> bool:
conn = sqlite3.connect("rate.db")
cutoff = int(time.time()) - 60
conn.execute("DELETE FROM hits WHERE ts < ?", (cutoff,))
count = conn.execute("SELECT COUNT(*) FROM hits WHERE email=?", (email,)).fetchone()[0]
if count >= limit_per_min:
return False
conn.execute("INSERT INTO hits(email, ts) VALUES(?, ?)", (email, int(time.time())))
conn.commit()
return True
この 2 段構えを入れてから、私の BI で 429 由来のエラー画面が出た記憶がほぼありません。
プロンプトインジェクション対策(BI 特有の落とし穴)
BI で Gemini に SQL を生成させたり、Plotly のグラフコードを生成させたりすると、ユーザーが CSV のセルに "; DROP TABLE ..." のような文字列を仕込んだ瞬間にインジェクションが現実問題になります。私の場合、対策は 3 層です。
第一に、Gemini に生成させるのは 必ず構造化出力にし、自由テキストで SQL を返させありません。response_schema で {"chart_type": "bar"|"line", "x_column": str, "y_column": str} のように制約します。
第二に、CSV のセル値を Gemini に渡す前に、長さ 200 文字超や制御文字を含むものをサニタイズ（あるいはサンプルから除外）します。
第三に、生成された SQL / コードは 必ずホワイトリストの関数とテーブルしか参照できないサブセットでパースしてから実行します。私は sqlglot で AST に落とし、SELECT 以外を弾く実装にしています。
import sqlglot
from sqlglot import exp
ALLOWED_TABLES = {"orders", "users", "events"}
def validate_sql(sql: str) -> bool:
try:
tree = sqlglot.parse_one(sql)
except Exception:
return False
if not isinstance(tree, exp.Select):
return False
tables = {t.name for t in tree.find_all(exp.Table)}
return tables.issubset(ALLOWED_TABLES)
可観測性: 誰が何を聞いて何トークン使ったかを追えるようにする
これがないと、月末の請求書を見て初めて「先週の火曜に誰かが Pro で 200 クエリ叩いた」みたいな事実に気づくことになります。本番運用に上げるなら、最低でも以下の 3 つを残します。
- リクエストログ: ユーザー email、tier、入力トークン、出力トークン、キャッシュヒット/ミス、応答時間。
- コスト計算: トークン数と単価から JPY を即時計算し、ユーザー別・日別・月別に集計可能にします。
- アラート: 日次でユーザー別コストが ¥500 を超えたら Slack か Discord に通知。
実装は SQLite + Streamlit のサイドバーで十分です。私は Streamlit のページとして /admin を作り、自分の email だけ表示できるガード付きで集計を出しています。
def log_call(email, tier, in_tok, out_tok, cached, latency_ms):
cost_jpy = (in_tok / 1_000_000 * PRICE_IN[tier] +
out_tok / 1_000_000 * PRICE_OUT[tier])
conn = sqlite3.connect("obs.db")
conn.execute(
"""INSERT INTO calls(email, tier, in_tok, out_tok, cached, latency_ms, cost_jpy, ts)
VALUES(?,?,?,?,?,?,?,?)""",
(email, tier, in_tok, out_tok, int(cached), latency_ms, cost_jpy, int(time.time())),
)
conn.commit()
if cost_jpy and daily_cost(email) > 500:
notify_slack(f"⚠️ {email} の本日のコストが ¥{daily_cost(email):.0f} を超過")
可観測性を入れて 1 週間で、私は Pro モデルを誤って常用に設定していたバグを発見しました。これがなかったら、その月だけで余計に ¥3,000 ほど払うところでした。
デプロイ先の比較: Cloud Run vs Streamlit Community Cloud vs Cloudflare Tunnel
最後にデプロイ先について。私が試した範囲での結論は以下です。
- Streamlit Community Cloud — 完全無料、デプロイ最速、ただしカスタムドメインなし・常時起動なしでコールドスタートあり。個人実験と社内 5 名以下なら最良。
- Cloud Run — 月 ¥300〜¥1,500 程度、カスタムドメインと Cloudflare Access が綺麗に組める。スケールも自動。社内 BI で本命採用。
- Cloudflare Tunnel + 自前サーバー — 既存 VPS や自宅 Mac mini を晒すパターン。学習コストは一番低いが、可用性は自分次第。個人開発の検証用。
私の場合、本番運用に上げる BI は Cloud Run + Cloudflare Access、検証や私一人で使う BI は Cloudflare Tunnel + Mac mini という二段構えに落ち着いています。
個人開発で実際に組んだ構成と、ハマったポイント
最後に、累計 5,000 万 DL のアプリ事業を一人で回す傍ら、Lacrima と Mystery の運営を含めて使っている私の現行 BI 構成を共有します。
- 認証: Cloudflare Access(Google ワンタイムパス、私の Google アカウントだけ許可)
- ホスティング: Cloud Run(min instances=0, max=2)
- モデル: デフォルト Gemini 2.5 Flash、「深く」ボタンで 2.5 Pro
- キャッシュ:
@st.cache_data + SQLite で 2 層、TTL 1 時間と 24 時間
- レート制限: tenacity 指数バックオフ + ユーザー単位 10 req/min
- 可観測性: SQLite ログ + 日次 ¥500 超過で Discord 通知
ハマったポイントを 3 つだけ。第一に、Cloud Run の min instances を 0 にしたら、月数百円浮く代わりに毎朝最初のクエリだけ 8 秒待つことになり、結局 1 に上げました。第二に、@st.cache_data の引数に DataFrame を直接渡すとハッシュが遅すぎてキャッシュより無キャッシュの方が速いという罠があり、必ずスキーマ要約の文字列を渡すように設計し直しました。第三に、Gemini の response_schema は深いネストになると 400 が返ることがあり、最大 2 階層に抑える運用にしています。
次のアクション
この記事を読み終えた段階で、最初に試すべきことは「自分の現行 Streamlit + Gemini アプリの認証層が今どうなっているか」を 1 行で書き出してみることだと思います。素の Streamlit のまま社内 URL を共有しているなら、まず streamlit-authenticator を 30 分で入れ、その上で本記事の第 2 〜 5 層を順番に積み上げていくのが、最短かつ最も事故の少ない進め方です。