Build a CSV Insight App with Gemini API and Streamlit — A Production-Ready Dashboard with Auto-Insights and Visualization

Two Decisions I'd Defend Behind This Stack

Before getting into code, let me name two choices I made that aren't obvious and that I'd push back on if a teammate suggested swapping them.

The first is choosing Gemini Flash as the default model rather than 2.5 Pro. The intuition is that "Pro must be better at analysis," and for one-off complex reports it is. But for a Streamlit app where the user is iterating — uploading, looking at charts, asking a follow-up, uploading another file — Flash's sub-second latency completely changes the feel of the product. The accuracy gap on schema summaries this small (under 4,000 tokens) is barely measurable. I keep Pro behind a sidebar toggle for cases where Flash returns something unconvincing, but Flash carries 95% of real usage.

The second is structured output via response_schema instead of asking for JSON in the prompt. You can technically get Gemini to return JSON with prompt engineering alone, and many tutorials do exactly that. The reality is that without schema enforcement, format drift is a constant tax on production reliability — one in fifty calls returns malformed JSON, your app shows an error, and the user assumes it's broken. The schema is the difference between an app that "works in the demo" and an app that survives a thousand uploads. If I had to pick one technique from this article to stress, it would be this one.

Setup and Dependencies

Start with a minimal dependency list. This is enough to deploy on Streamlit Community Cloud:

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

Read the API key from .env locally and from st.secrets in the cloud. Writing the loader once with both fallbacks saves you a round of post-deploy debugging:

# secrets_loader.py
import os
from dotenv import load_dotenv
import streamlit as st
 
def load_api_key() -> str:
    """Load the API key from .env locally or st.secrets in the cloud.
    Raise an explicit error if neither has it."""
    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 is not set. "
            "Check your .env file or .streamlit/secrets.toml."
        )
    return key

Sample run:

$ python -c "from secrets_loader import load_api_key; print(load_api_key()[:6] + '...')"
YOUR_A...

With this in place, every downstream module can call load_api_key() without caring whether it's running locally or in the cloud.

Reading the CSV Without Crashing

This is the unglamorous part that breaks first in production. CSVs out in the wild come in Shift_JIS, BOM-prefixed UTF-8, latin-1, you name it. A UnicodeDecodeError on the very first action a user takes is the worst possible welcome.

# loader.py
import io
import chardet
import pandas as pd
 
MAX_BYTES = 50 * 1024 * 1024  # 50 MB
 
def read_uploaded_csv(file) -> pd.DataFrame:
    """Take a Streamlit UploadedFile and return a DataFrame, detecting encoding.
    Reject anything over 50 MB up front with a friendly message."""
    raw: bytes = file.getvalue()
    if len(raw) > MAX_BYTES:
        raise ValueError(
            f"File is too large ({len(raw) / 1024 / 1024:.1f} MB). "
            f"Please split it into chunks under 50 MB."
        )
    # Detecting on the first 32 KB takes well under 100 ms
    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:
        # Detection isn't perfect — fall back to cp932 as a safety net
        return pd.read_csv(io.BytesIO(raw), encoding="cp932")
    except pd.errors.ParserError as e:
        raise ValueError(f"Failed to parse the CSV: {e}") from e

Why cap at 50 MB? Because Streamlit's default upload limit (200 MB) sits well above what you can realistically push through a single Gemini request, and beyond that size you really want users to pre-aggregate in SQL or DuckDB before uploading. The result quality also improves once aggregation pushes upstream.

Building a Schema Summary (the Key to Token Economics)

What we send to Gemini is not "all rows of the CSV." It's a description: what shape does this table have, and what kinds of values live in each column. The combination I've found best balances accuracy against cost is:

# schema.py
from typing import Any
import pandas as pd
 
def build_schema_summary(df: pd.DataFrame, sample_rows: int = 5) -> dict[str, Any]:
    """Build a lightweight schema summary intended for Gemini.
    Send types, null ratios, statistics, and a few sample rows — never every column verbatim."""
    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:
            # For string columns, ship only top-3 frequent values to limit PII exposure
            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):
    """Convert NaN / inf to JSON-serializable 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

The _safe_num helper carries more weight than it looks. Pandas happily returns NaN and inf for empty or pathological columns, and json.dumps blows up on both. I missed this on my first build and silently produced an app that crashed on every CSV with a fully-empty column. Catch it at summary time and the rest of the pipeline never has to care.

Capping string-column frequencies at the top three values is also deliberate. Sending value_counts() for a name or address column would leak PII straight into the prompt. Top three is statistically meaningful and minimizes incidental exposure of sensitive fields.

Generating Structured Output with Gemini

Here's the heart of the article. If you ask Gemini for "freeform insights," the output shape drifts every call and your UI breaks. Use response_schema to force a JSON object split into four fixed fields:

# 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": "One-sentence headline summarizing this dataset",
        },
        "trends": {
            "type": "ARRAY",
            "items": {"type": "STRING"},
            "description": "3-5 important trends, each under 80 chars",
        },
        "anomalies": {
            "type": "ARRAY",
            "items": {"type": "STRING"},
            "description": "Outliers or unusual distributions. Empty array if nothing notable",
        },
        "next_questions": {
            "type": "ARRAY",
            "items": {"type": "STRING"},
            "description": "2-4 hypotheses or follow-up data requests worth pursuing",
        },
        "recommended_charts": {
            "type": "ARRAY",
            "items": {
                "type": "OBJECT",
                "properties": {
                    "kind": {"type": "STRING", "description": "One of: 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 = """You are a seasoned data analyst.
Given the schema summary and sample rows, return insights a business stakeholder can act on immediately.
- Trends should include numbers and units (e.g. "Tokyo's average order value is 18% higher than other regions")
- Recommended charts must reference real column names only
- Never produce text that could identify an individual person
"""
 
def generate_insights(client: genai.Client, schema_summary: dict[str, Any]) -> dict[str, Any]:
    """Send the schema summary to Gemini and return structured insights.
    Retry up to 3 times with exponential backoff on 429 / 503."""
    prompt = f"Below is the schema and head sample of a 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"Failed to get a response from Gemini: {last_err}")

I keep temperature=0.3 because data analysis output should be roughly deterministic. At 0.7 you get a different headline and different rounded numbers every time, which makes users wonder if the app is broken. At 0.0 the output becomes uncomfortably uniform. My personal sweet spot for analytical workloads is 0.2 to 0.4. There's a deeper write-up at Gemini API Temperature Best Practices by Task.

Rendering Recommended Charts with Plotly

Gemini will hand back specs like {"kind": "scatter", "x": "age", "y": "income", "color": "region"}. Translate them to Plotly, but always validate that the columns actually exist — the model occasionally hallucinates column names.

# charts.py
import plotly.express as px
import pandas as pd
 
def render_chart(df: pd.DataFrame, spec: dict):
    """Convert a Gemini chart spec into a Plotly Figure.
    Return None if any referenced column is missing so the UI can explain the skip."""
    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

In the Streamlit layer this becomes a single for loop. When kind is something unexpected the function returns None — surface that to the user with a small caption so they're not left wondering why a card is empty.

Wiring It All Up

Combine everything into app.py:

# 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 — Auto-Analysis with Gemini")
 
if "client" not in st.session_state:
    st.session_state.client = genai.Client(api_key=load_api_key())
 
uploaded = st.file_uploader("Upload a CSV to analyze", type=["csv"])
if not uploaded:
    st.info("Drop a CSV above and the app will produce insights and charts automatically.")
    st.stop()
 
with st.spinner("Reading CSV..."):
    df = read_uploaded_csv(uploaded)
 
st.success(f"Loaded {len(df):,} rows × {len(df.columns)} columns")
with st.expander("Preview first 20 rows"):
    st.dataframe(df.head(20))
 
if st.button("🚀 Generate insights with Gemini", type="primary"):
    with st.spinner("Gemini is analyzing..."):
        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("### 🔍 Key Trends")
        for t in ins["trends"]:
            st.write(f"- {t}")
        st.markdown("### ⚠️ Anomalies")
        if ins["anomalies"]:
            for a in ins["anomalies"]:
                st.write(f"- {a}")
        else:
            st.write("No notable anomalies detected.")
    with col2:
        st.markdown("### 💡 Hypotheses to Test Next")
        for q in ins["next_questions"]:
            st.write(f"- {q}")
 
    st.markdown("### 📈 Recommended Charts")
    for spec in ins["recommended_charts"]:
        fig = render_chart(st.session_state.df, spec)
        if fig is None:
            st.caption(f"⚠ Could not render {spec.get('kind')} (invalid column reference)")
            continue
        st.plotly_chart(fig, use_container_width=True)
        if spec.get("rationale"):
            st.caption(f"Why this chart: {spec['rationale']}")

Run it with streamlit run app.py, drop in any CSV, and within seconds you have a row of insight cards and a stack of charts. That's a complete MVP.

Adding Chat Follow-Ups (with History Compression)

Once people see the insights, the next question is always something like "and which store does this anomaly come from?" Streamlit's st.chat_input adds a chat UI in about ten lines, but if you don't compress history you'll burn tokens fast. My go-to pattern is to fold the older half of the conversation into a single summary message once the history grows beyond ten turns:

# chat.py
from google import genai
from google.genai import types
 
CHAT_INSTRUCTION = """You are a data analyst. Given the schema summary and prior insights for the CSV the user just uploaded,
answer their question concisely and concretely. If the data is insufficient, say so explicitly."""
 
def chat_reply(client: genai.Client, history: list[dict], schema_summary: dict, question: str) -> str:
    # When history grows long, summarize the older half into one compact message
    if len(history) > 10:
        old = history[:-6]
        recent = history[-6:]
        summary_prompt = "Summarize the following conversation in under 200 chars:\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": "[Prior summary] " + s.text}, *recent]
 
    convo = "\n".join(f"{m['role']}: {m['content']}" for m in history)
    prompt = f"Schema summary:\n{schema_summary}\n\nConversation:\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

I cover this technique standalone in Gemini API Rolling Summary Chat History Compression, and in a data analysis app it pays off doubly. The schema summary alone already costs around 1,000 tokens per call, so unbounded chat history pushes you past 10K tokens within ten exchanges.

Handling Rate Limits and Token Bloat

The single most common production failure with this kind of app is "worked locally, started returning 429 in the cloud." Gemini Flash's free tier caps requests per minute, and a couple of concurrent users will hit it. I use a three-layer defense.

First, estimate token usage before sending and shrink the summary if it's about to exceed 4,000 tokens. Second, retry with exponential backoff on 429/503 (already in generate_insights). Third, expose a model fallback in the Streamlit sidebar — if Flash is throttled, let the user click over to 2.5 Pro for the next call.

# token_guard.py
def estimate_tokens(text: str) -> int:
    """Rough token estimate for mixed Japanese/English text.
    Strict measurement would call count_tokens, but this is precise enough for UI gating."""
    return len(text) // 2 + text.count(" ")
 
def shrink_summary_if_needed(summary: dict, limit: int = 4000) -> dict:
    """If the JSON-serialized summary is estimated to exceed `limit` tokens,
    halve the sample rows and trim top_values until it fits."""
    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

For strict measurement you'd call count_tokens, but for "should I send this or shrink it first" decisions, character-count divided by two is good enough. When you need finer control, pair this with Gemini API Rate Limiting & Quota Management Production Guide and Gemini API Cost Optimization Complete Guide.

Common Mistakes and Pitfalls

After shipping versions of this app for four different sites and seeing colleagues use it, here are the five mistakes I keep watching people repeat — along with how to avoid each.

First, the urge to send all rows to Gemini. "Maybe accuracy is higher if I just give it everything?" Empirically, accuracy barely moves while cost and latency explode. Past about 10,000 rows, the wasted tokens cause more harm than the extra signal helps.

Second, passing numeric columns as strings. CSVs with comma-thousands like 1,234 get parsed as object dtype, which Gemini reads as strings — and strings have no mean. Run pd.to_numeric(s, errors="coerce") before building the summary and answer quality jumps noticeably.

Third, leaking PII into the prompt. Names, emails, addresses sent verbatim through top_values end up in logs and caches. Either flag PII-shaped column names and skip their top_values, or mask sample rows for those columns. The patterns in Gemini API PII Redaction & Audit Logging translate directly here.

Fourth, writing response_schema too loosely. If you make recommended_charts an array of strings, the format will drift call-to-call and Plotly can't consume them. Force OBJECT items with required kind, x, y — render-time failures drop by an order of magnitude. The Pydantic-typed approach is in Gemini API + Pydantic Type-Safe Structured Output.

Fifth, trusting Streamlit session state too much. st.session_state evaporates when the user closes the tab. If users want to keep insights, give them an explicit "save" path. I always provide a st.download_button for both insights.json and an HTML report.html carrying the recommended charts.

Deploy and Operate in Production

The fastest path to a live URL is Streamlit Community Cloud — push to GitHub, paste GOOGLE_API_KEY into secrets.toml, and you're live. The catch is the 1 GB memory cap on the free tier; if you anticipate users uploading datasets where pandas alone consumes 200 MB+, start on Cloud Run or Fly.io to avoid migrating later.

For Cloud Run, a Dockerfile based on Python 3.12 with Streamlit listening on PORT=8080 works immediately. I've verified that even the smallest spec (256 MB / 0.5 CPU) handles 10 MB CSVs without trouble. Setting Concurrency to 80 and min instances to 0 keeps personal-project bills under a few dollars a month. I cover this in detail at Building Usage-Based SaaS with Gemini API and Cloud Run.

The non-negotiable security rule is don't persist the uploaded CSV anywhere on the server. Streamlit handles UploadedFile in memory, but a stray logger.info(df.to_string()) will write the entire dataset to your platform's log retention. st.write(df) is safe — it renders to the user only — but bespoke logging needs to be audited.

Where to Take This Next

The app stays in one file, but the architecture cleanly absorbs several extensions, and each one only needs a hundred lines or so on top of what we already have.

A multi-CSV join mode is the easiest natural progression. Add a second st.file_uploader, run a left join on a column the user picks, and feed the joined frame into the existing build_schema_summary. Nothing else changes. This is also where Gemini's value compounds — the model is genuinely good at hypothesizing how two tables relate, so prompting it with both schemas produces useful suggestions for join keys and likely cardinality issues.

A Slack integration turns the same code into team analytics. With Slack Bolt for Python, a slash command that accepts a file URL and returns the headline plus top-three trends is around 30 lines. The Streamlit UI stays for ad-hoc exploration; the Slack bot covers the daily digest use case.

For monetization, the path to a B2B SaaS is shorter than it looks. Drop Stripe in front of the upload widget, gate analyses behind a subscription, and charge by analysis count or row count. The patterns from Gemini API × Stripe SaaS Billing Webhook Production Guide drop in with almost no modification — checkout session, webhook handler, KV-backed entitlement check.

The unit economics work out kindly for solo developers. Gemini Flash is cheap enough that a single CSV analysis costs well under a cent of API spend. A thousand monthly analyses keeps your inference bill under a couple of dollars even before any caching, and Streamlit Cloud or Cloud Run hosting fits in another handful of dollars per month. That's the whole reason this stack appeals to me: the fixed cost floor is low enough that you can launch a niche tool, get one paying customer, and already be profitable on a unit basis from day one.

A Note on Quality Calibration

One question that keeps coming up when developers try this pattern: how do I tell if the insights are actually good?

I've adopted a simple two-step ritual. First, before publishing the app, I run it on three datasets where I already know the answer — sales data from a business I run, a public dataset I've manually analyzed, and a synthetic dataset where I planted specific anomalies. If the model misses the planted anomaly or fabricates a trend that doesn't exist, the schema summary needs more signal (more sample rows, finer-grained statistics, or stronger column descriptions in the prompt).

Second, in production, I log every insight along with the input schema hash, then sample 5% for manual review weekly. After a month you have hundreds of insight-pairs to learn from, and you start spotting systematic failure modes — say, the model always overstates the importance of high-null columns, or always recommends bar charts when a histogram would be better. From there you tune the system instruction and watch the failure rate fall.

This is the same approach you'd use to QA any structured-output pipeline, but for data analysis it pays off especially fast because the failure modes are usually subtle (wrong-but-plausible) rather than obvious (broken JSON).

Wrap-Up

The whole craft of this app comes down to designing what you send to Gemini. Not the raw CSV — a compressed schema-and-statistics summary, returned as structured output. That single discipline gives you cost, latency, and answer quality all at once. Everything else in this article is scaffolding around that core idea.

Your first move should be to run streamlit run app.py against a CSV you actually use. A sales export from your business, a dataset from Kaggle, anything you've stared at recently. The experience of getting the analysis you usually do by hand back in five seconds will reframe how you think about data tooling. From there, the gap between "personal utility" and "internal product your team relies on" is much smaller than you'd expect — usually a weekend of polish and a payment integration.