Gemini API × Streamlit で CSV 分析アプリをつくる — 自動洞察と可視化を備えた実用ダッシュボードの実装ガイド

# requirements.txt
streamlit>=1.42
google-genai>=0.8
pandas>=2.2
plotly>=5.24
python-dotenv>=1.0
chardet>=5.2

# secrets_loader.py
import os
from dotenv import load_dotenv
import streamlit as st
 
def load_api_key() -> str:
    """ローカルでは .env、Streamlit Cloud では st.secrets から API キーを読む。
    両方に存在しない場合は明示的に例外を投げる。"""
    load_dotenv()
    key = os.getenv("GOOGLE_API_KEY")
    if not key:
        try:
            key = st.secrets["GOOGLE_API_KEY"]
        except (KeyError, FileNotFoundError):
            key = None
    if not key:
        raise RuntimeError(
            "GOOGLE_API_KEY が設定されていません。"
            ".env または .streamlit/secrets.toml を確認してください。"
        )
    return key

# loader.py
import io
from typing import Optional
import chardet
import pandas as pd
import streamlit as st
 
MAX_BYTES = 50 * 1024 * 1024  # 50 MB
 
def read_uploaded_csv(file) -> pd.DataFrame:
    """Streamlit の UploadedFile を受け取り、エンコーディング自動判定で DataFrame を返す。
    50 MB を超えたら早期に拒否し、ユーザーに分かりやすく説明する。"""
    raw: bytes = file.getvalue()
    if len(raw) > MAX_BYTES:
        raise ValueError(
            f"ファイルが大きすぎます({len(raw) / 1024 / 1024:.1f} MB)。"
            f"50 MB 以下に分割してください。"
        )
    # 先頭 32 KB だけで推定すれば 0.1 秒以内に終わる
    detected = chardet.detect(raw[:32_000])
    encoding = detected.get("encoding") or "utf-8"
    try:
        return pd.read_csv(io.BytesIO(raw), encoding=encoding)
    except UnicodeDecodeError:
        # 推定が外れることもあるので保険として cp932 を試す
        return pd.read_csv(io.BytesIO(raw), encoding="cp932")
    except pd.errors.ParserError as e:
        raise ValueError(f"CSV のパースに失敗しました: {e}") from e

# schema.py
from typing import Any
import pandas as pd
 
def build_schema_summary(df: pd.DataFrame, sample_rows: int = 5) -> dict[str, Any]:
    """Gemini に渡すための軽量なスキーマ要約を作る。
    全カラムを送るのではなく、型・null率・統計・サンプルだけを抽出する。"""
    summary: dict[str, Any] = {
        "shape": {"rows": len(df), "columns": len(df.columns)},
        "columns": [],
        "sample": df.head(sample_rows).to_dict(orient="records"),
    }
    for col in df.columns:
        s = df[col]
        info: dict[str, Any] = {
            "name": col,
            "dtype": str(s.dtype),
            "null_ratio": round(s.isna().mean(), 3),
            "unique": int(s.nunique(dropna=True)),
        }
        if pd.api.types.is_numeric_dtype(s):
            info["stats"] = {
                "min": _safe_num(s.min()),
                "max": _safe_num(s.max()),
                "mean": _safe_num(s.mean()),
                "std": _safe_num(s.std()),
            }
        elif pd.api.types.is_datetime64_any_dtype(s):
            info["range"] = {"min": str(s.min()), "max": str(s.max())}
        else:
            # 文字列カラムは頻出値だけを示す（個人情報リスクを下げるため上位 3 件まで）
            top = s.dropna().astype(str).value_counts().head(3)
            info["top_values"] = top.to_dict()
        summary["columns"].append(info)
    return summary
 
def _safe_num(x):
    """NaN / inf を JSON シリアライズ可能な None に変換する。"""
    try:
        v = float(x)
        if v != v or v in (float("inf"), float("-inf")):
            return None
        return round(v, 4)
    except (TypeError, ValueError):
        return None

# insights.py
import json
import time
from typing import Any
from google import genai
from google.genai import types
 
INSIGHT_SCHEMA = {
    "type": "OBJECT",
    "properties": {
        "headline": {
            "type": "STRING",
            "description": "このデータを 1 文で表すヘッドライン",
        },
        "trends": {
            "type": "ARRAY",
            "items": {"type": "STRING"},
            "description": "重要な傾向を 3〜5 件、各 60 字以内",
        },
        "anomalies": {
            "type": "ARRAY",
            "items": {"type": "STRING"},
            "description": "異常値や不自然な分布。発見できなければ空配列",
        },
        "next_questions": {
            "type": "ARRAY",
            "items": {"type": "STRING"},
            "description": "次に検証すべき仮説や追加データの提案を 2〜4 件",
        },
        "recommended_charts": {
            "type": "ARRAY",
            "items": {
                "type": "OBJECT",
                "properties": {
                    "kind": {"type": "STRING", "description": "scatter / bar / line / hist / box のいずれか"},
                    "x": {"type": "STRING"},
                    "y": {"type": "STRING"},
                    "color": {"type": "STRING"},
                    "rationale": {"type": "STRING"},
                },
                "required": ["kind", "x"],
            },
        },
    },
    "required": ["headline", "trends", "anomalies", "next_questions", "recommended_charts"],
}
 
INSTRUCTION = """あなたは経験豊富なデータアナリストです。
与えられたスキーマ要約とサンプル行から、ビジネス担当者が即座に行動できる洞察を返してください。
- trends は数値と単位を含めて具体的に書く(例「東京の客単価が他地域より 18% 高い」)
- 推奨グラフは実在する列名のみを使用する
- 個人を特定する記述は絶対にしない
"""
 
def generate_insights(client: genai.Client, schema_summary: dict[str, Any]) -> dict[str, Any]:
    """スキーマ要約を Gemini に渡し、構造化された洞察を返す。
    429 / 503 は最大 3 回までエクスポネンシャルバックオフで再試行する。"""
    prompt = f"以下が CSV のスキーマと先頭サンプルです。\n```json\n{json.dumps(schema_summary, ensure_ascii=False, indent=2)}\n```"
    config = types.GenerateContentConfig(
        system_instruction=INSTRUCTION,
        response_mime_type="application/json",
        response_schema=INSIGHT_SCHEMA,
        temperature=0.3,
    )
    last_err: Exception | None = None
    for attempt in range(3):
        try:
            res = client.models.generate_content(
                model="gemini-2.5-flash",
                contents=prompt,
                config=config,
            )
            return json.loads(res.text)
        except Exception as e:
            last_err = e
            msg = str(e).lower()
            if "429" in msg or "503" in msg or "overloaded" in msg:
                time.sleep(2 ** attempt + 0.5)
                continue
            raise
    raise RuntimeError(f"Gemini からの応答取得に失敗しました: {last_err}")

# charts.py
import plotly.express as px
import pandas as pd
 
def render_chart(df: pd.DataFrame, spec: dict) -> "plotly.graph_objs._figure.Figure | None":
    """Gemini の出力した chart spec を Plotly Figure に変換する。
    指定列が存在しない場合は None を返し、UI 側でユーザーに伝える。"""
    kind = spec.get("kind", "").lower()
    x = spec.get("x")
    y = spec.get("y")
    color = spec.get("color")
    if x and x not in df.columns:
        return None
    if y and y not in df.columns:
        return None
    if color and color not in df.columns:
        color = None
    try:
        if kind == "scatter" and y:
            return px.scatter(df, x=x, y=y, color=color, opacity=0.6)
        if kind == "bar" and y:
            return px.bar(df.groupby(x, dropna=False)[y].mean().reset_index(), x=x, y=y, color=color)
        if kind == "line" and y:
            return px.line(df.sort_values(x), x=x, y=y, color=color)
        if kind == "hist":
            return px.histogram(df, x=x, color=color, nbins=40)
        if kind == "box" and y:
            return px.box(df, x=x, y=y, color=color)
    except Exception:
        return None
    return None

# app.py
import streamlit as st
from google import genai
from secrets_loader import load_api_key
from loader import read_uploaded_csv
from schema import build_schema_summary
from insights import generate_insights
from charts import render_chart
 
st.set_page_config(page_title="CSV Insight", page_icon="📊", layout="wide")
st.title("📊 CSV Insight — Gemini で自動洞察")
 
if "client" not in st.session_state:
    st.session_state.client = genai.Client(api_key=load_api_key())
 
uploaded = st.file_uploader("分析したい CSV をアップロード", type=["csv"])
if not uploaded:
    st.info("CSV をアップロードすると、自動で洞察と推奨グラフを生成します。")
    st.stop()
 
with st.spinner("CSV を読み込み中..."):
    df = read_uploaded_csv(uploaded)
 
st.success(f"読み込み完了: {len(df):,} 行 × {len(df.columns)} 列")
with st.expander("先頭 20 行を確認"):
    st.dataframe(df.head(20))
 
if st.button("🚀 Gemini で洞察を生成", type="primary"):
    with st.spinner("Gemini が分析中..."):
        summary = build_schema_summary(df)
        insights = generate_insights(st.session_state.client, summary)
    st.session_state.insights = insights
    st.session_state.df = df
 
if "insights" in st.session_state:
    ins = st.session_state.insights
    st.subheader(ins["headline"])
 
    col1, col2 = st.columns(2)
    with col1:
        st.markdown("### 🔍 重要な傾向")
        for t in ins["trends"]:
            st.write(f"- {t}")
        st.markdown("### ⚠️ 異常値・気になる分布")
        if ins["anomalies"]:
            for a in ins["anomalies"]:
                st.write(f"- {a}")
        else:
            st.write("特筆すべき異常は見つかりませんでした。")
    with col2:
        st.markdown("### 💡 次に検証すべき仮説")
        for q in ins["next_questions"]:
            st.write(f"- {q}")
 
    st.markdown("### 📈 推奨グラフ")
    for spec in ins["recommended_charts"]:
        fig = render_chart(st.session_state.df, spec)
        if fig is None:
            st.caption(f"⚠ {spec.get('kind')} を描画できませんでした(列が不正)")
            continue
        st.plotly_chart(fig, use_container_width=True)
        if spec.get("rationale"):
            st.caption(f"なぜこのグラフか: {spec['rationale']}")

# chat.py
from google import genai
from google.genai import types
 
CHAT_INSTRUCTION = """あなたはデータアナリストです。直前にユーザーが上げた CSV のスキーマと洞察を踏まえ、
ユーザーの質問に短く・具体的に答えてください。確証がない場合は「データから断定できない」と明示してください。"""
 
def chat_reply(client: genai.Client, history: list[dict], schema_summary: dict, question: str) -> str:
    # 履歴が長くなったら古い半分を要約して 1 つの message に畳む
    if len(history) > 10:
        old = history[:-6]
        recent = history[-6:]
        summary_prompt = "次の対話を 200 字以内で要約してください:\n" + \
            "\n".join(f"{m['role']}: {m['content']}" for m in old)
        s = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=summary_prompt,
            config=types.GenerateContentConfig(temperature=0.0),
        )
        history = [{"role": "user", "content": "[これまでの要約] " + s.text}, *recent]
 
    convo = "\n".join(f"{m['role']}: {m['content']}" for m in history)
    prompt = f"スキーマ要約:\n{schema_summary}\n\n対話履歴:\n{convo}\n\nuser: {question}\nassistant:"
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(system_instruction=CHAT_INSTRUCTION, temperature=0.4),
    )
    return res.text

# token_guard.py
def estimate_tokens(text: str) -> int:
    """日本語混在テキストのトークン数をざっくり見積もる。
    厳密には count_tokens を呼ぶべきだが、UI 側の判断には十分な精度。"""
    return len(text) // 2 + text.count(" ")
 
def shrink_summary_if_needed(summary: dict, limit: int = 4000) -> dict:
    """JSON 文字列化したときのトークン見積もりが limit を超えるなら、
    サンプル行と top_values を半分に削る。"""
    import json
    s = json.dumps(summary, ensure_ascii=False)
    while estimate_tokens(s) > limit and summary["sample"]:
        summary["sample"] = summary["sample"][: max(1, len(summary["sample"]) // 2)]
        for col in summary["columns"]:
            if "top_values" in col:
                col["top_values"] = dict(list(col["top_values"].items())[:1])
        s = json.dumps(summary, ensure_ascii=False)
    return summary