Upload 5 files

README.md

@@ -1,19 +1,166 @@
+# Data Analysis Agent (Streamlit + LangGraph ReAct Agent)
+
+A lightweight Streamlit application that ingests **Excel (.xlsx)** files into a local **DuckDB** database and lets you **chat with your data**. It uses a **LangGraph ReAct-style agent** to generate DuckDB SQL from natural language, refine on errors, and optionally render plots. Answers stream live, are aware of chat history, and show the generated SQL in an expander for transparency.
+
 ---
-title: Data Analysis Agent
-emoji: 🚀
-colorFrom: red
-colorTo: red
-sdk: docker
-app_port: 8501
-tags:
-- streamlit
-pinned: false
-short_description: Streamlit template space
+
+## Highlights
+
+- **One‑click ingestion**: Excel → DuckDB using `excel_to_duckdb.py` (keeps ingestion logs).
+- **Overview & preview**: Human‑friendly overview and quick table previews before chatting.
+- **Chat with your dataset**: Natural‑language questions → SQL → answer (+ optional chart).
+- **Streaming**: Answers appear token‑by‑token (“Answer pending…” shows immediately).
+- **History‑aware**: Follow‑ups consider prior Q&A context for better intent & filters.
+- **Conservative plotting**: Only plots when useful (trends/many rows or explicitly asked).
+- **Clean UI**: Chart appears **below** the answer and **before** the generated SQL.
+- **Session isolation**: Uploading a new file **clears** overview, previews, and chat.
+- **Local by default**: DuckDB DB & plots are saved locally (`./dbs`, `./plots`).
+
+---
+
+## Repo Structure
+
+```
+streamlit_app.py          # Streamlit app (UI, flow, chat)
+duckdb_react_agent.py     # LangGraph ReAct agent (SQL generation, refine, answer, plots)
+excel_to_duckdb.py     # Excel → DuckDB ingestion script (unchanged behavior)
+uploads/                  # Uploaded Excel files (created at runtime)
+dbs/                      # DuckDB files per upload (created at runtime)
+plots/                    # Charts generated by the agent (created at runtime)
+```
+
+> **Note:** The app intentionally skips internal tables (e.g., `__excel_*`) when discovering “user tables.” Those tables hold ingestion metadata and previews.
+
 ---
 
-# Welcome to Streamlit!
+## Requirements
+
+- Python 3.10+
+- Recommended packages:
+  ```bash
+  pip install streamlit duckdb pandas matplotlib python-dotenv langgraph langchain langchain-openai
+  ```
+- **OpenAI** API key in environment or `.env` file:
+  ```dotenv
+  OPENAI_API_KEY=sk-...
+  # optional: model override (defaults to gpt-4o-mini in app; gpt-4o in CLI agent)
+  OPENAI_MODEL=gpt-4o-mini
+  ```
+
+If your ingestion or preview relies on Excel parsing features, make sure you have:
+```bash
+pip install openpyxl
+```
+
+---
+
+## Quick Start
+
+1. Place the three files side‑by‑side:
+   ```
+   streamlit_app.py
+   duckdb_react_agent.py
+   excel_to_duckdb.py
+   ```
+
+2. Set your API key (env var or `.env`):
+   ```bash
+   export OPENAI_API_KEY=sk-...        # macOS/Linux
+   setx OPENAI_API_KEY "sk-..."        # Windows PowerShell (new sessions)
+   ```
+
+3. Run the app:
+   ```bash
+   streamlit run streamlit_app.py
+   ```
+
+4. In the UI:
+   - **Upload** an `.xlsx` file → watch the **Processing logs**.
+   - Once ingestion finishes, you’ll see the **overview** and **quick previews**.
+   - Scroll down to **Chat with your dataset** and start asking questions.
+   - Each bot reply shows: **answer text**, optional **chart**, and an expander for **SQL**.
+
+> **New upload behavior:** As soon as you upload a different file, the app **clears** the previous overview, table previews, and chat, then ingests the new file.
+
+---
+
+## How It Works
+
+### 1) Ingestion (Excel → DuckDB)
+- The app shells out to `excel_to_duckdb.py`:
+  - Detects multiple tables per sheet (using blank rows as separators), handles merged headers, ignores pivot/chart‑like blocks, and writes one DuckDB table per block.
+  - Creates helpful metadata tables (e.g., `__excel_tables`, `__excel_schema`).
+- The app then opens the DuckDB file and builds a **schema snapshot** for the agent.
+- **User tables**: The app queries `information_schema` to list the “real” user tables (skips internals like `__excel_*`).
+
+### 2) LangGraph ReAct Agent (DuckDB Q&A)
+- **ReAct loop**: Draft a SELECT query → run → if error, **refine up to 3×** with the error and prior SQL.
+- **Safety**: Only **SELECT** queries are permitted.
+- **Streaming answers**: Final natural‑language answer streams to the UI.
+- **History awareness**: Recent Q&A is summarized and fed to the LLM for better follow‑ups.
+- **Plotting (optional)**: If the result truly benefits from visuals (or you ask for one), a chart is generated and saved to `./plots`; the app then displays it **below** the answer.
+
+> The agent avoids revealing plot file names or paths. Answers refer generically to “the chart” and the UI renders the image inline.
+
+---
+
+## Configuration & Tweaks
+
+- **Models**: Set `OPENAI_MODEL` in your env to change the LLM (`gpt-4o-mini`, `gpt-4o`, etc.).
+- **Plot policy**: The agent is conservative by default. To make it more/less eager to chart, tune the **viz prompt** or the **row‑count threshold** in `duckdb_react_agent.py` (`many_rows = df.shape[0] > 20`).
+- **Chart size**: In `streamlit_app.py`, the plot width is ~**560px** for answers and ~**320‑360px** in history; adjust to taste.
+- **Schemas**: The schema summary focuses on `main`. To include more schemas, extend the `allowed_schemas` list where `get_schema_summary()` is called.
+
+---
+
+## Troubleshooting
+
+**“No user tables were discovered. Showing ingestion metadata…”**  
+- Your Excel may not have recognizable data blocks under the current heuristics. Check the **Processing logs** and the `__excel_tables` preview to verify what got detected.
+- If it’s a header/merged‑cell issue, revisit your spreadsheet (ensure at least one empty row between blocks).
+
+**Agent errors / SQL fails repeatedly**  
+- The agent refines up to 3×. If it still fails, it returns the error. Try asking the question more explicitly (table names, time ranges, filters).
+
+**Streamlit cache / stale code**  
+- If updates don’t show up, restart Streamlit or clear cache:
+  ```bash
+  streamlit cache clear
+  ```
+
+**Windows console encoding warnings**  
+- The app sets `PYTHONIOENCODING=utf-8` for ingestion logs. If you still see encoding issues, ensure system locale supports UTF‑8.
+
+---
+
+## Security & Privacy Notes
+
+- **Local files**: DuckDB files and plots are stored locally by default (`./dbs`, `./plots`).
+- **What goes to the LLM**: The agent sends the **schema snapshot** and a **small result preview** (first rows) along with your **question** and a compact **history summary**—not the full dataset. For stricter controls, reduce or strip the preview passed to the LLM in `duckdb_react_agent.py`.
+
+---
+
+## Extending the App
+
+- **CSV support**: Add a CSV branch next to the Excel ingest that writes to the same DuckDB.  
+- **Persisted chat**: Save `chat_history` to a file keyed by DB name to restore on reload.  
+- **Role‑specific prompts**: Swap in system prompts for financial analysis, marketing analytics, etc.  
+- **Auth & multi‑user**: Gate uploads and route per‑user databases under a workspace prefix.
+
+---
+
+## CLI (Agent Only)
+
+You can also run the agent from the command line for ad‑hoc Q&A:
+
+```bash
+python duckdb_react_agent.py --duckdb ./dbs/mydata.duckdb --stream
+```
+
+Type your question; the agent prints streaming answers and shows where charts were saved (the Streamlit app displays them inline).
+
+---
 
-Edit `/src/streamlit_app.py` to customize this app to your heart's desire. :heart:
+## License
 
-If you have any questions, checkout our [documentation](https://docs.streamlit.io) and [community
-forums](https://discuss.streamlit.io).
+Proprietary / internal. Adapt as needed for your environment.

duckdb_react_agent.py

@@ -0,0 +1,628 @@
+"""
+DuckDB Q&A Agent (LangGraph + ReAct) — Streaming Edition
+========================================================
+
+Adds:
+- Streaming responses for the final natural-language answer (CLI prints tokens live).
+- Stricter schema discovery focused on *user* tables from the provided DB (defaults to schema 'main').
+- Keeps SELECT-only safety, up to 3 SQL refinements on error, optional plotting saved to ./plots.
+
+Quick start
+-----------
+1) Install deps (example):
+   pip install --upgrade duckdb pandas matplotlib python-dotenv langgraph langchain langchain-openai
+
+2) Ensure an OpenAI API key is available:
+   - Put it in a .env file as: OPENAI_API_KEY=sk-...
+   - Or set an env var: set OPENAI_API_KEY=... (Windows) / export OPENAI_API_KEY=... (macOS/Linux)
+
+3) Run the agent in an interactive loop:
+   python duckdb_react_agent.py --duckdb path/to/your.db --stream
+
+Notes
+-----
+- Targets DuckDB SQL. The LLM prompt instructs to write *only* a SELECT statement.
+- ReAct loop: on error, the LLM sees the error message + previous SQL and attempts to fix it (max 3 tries).
+- The agent avoids DDL/DML; SELECT-only for safety.
+- Plots are saved to ./plots/ as PNG files (no GUI required). The script uses a non-interactive backend.
+- Internal/helper tables beginning with "__" (e.g., ingestion metadata) and non-user schemas are ignored.
+"""
+
+import os
+import re
+import json
+import uuid
+import argparse
+from typing import Any, Dict, List, Optional, TypedDict, Callable
+
+# Non-GUI backend for saving plots on servers/Windows terminals
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+
+import duckdb
+import pandas as pd
+
+from dotenv import load_dotenv, find_dotenv
+
+# LangChain / LangGraph
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import SystemMessage, HumanMessage
+from langgraph.graph import StateGraph, END
+
+
+# -----------------------------
+# Utility: safe number formatting
+# -----------------------------
+def format_number(x: Any) -> str:
+    try:
+        if isinstance(x, (int, float)) and not isinstance(x, bool):
+            if abs(x) >= 1000 and abs(x) < 1e12:
+                return f"{x:,.2f}".rstrip('0').rstrip('.')
+            else:
+                return f"{x:.4g}" if isinstance(x, float) else str(x)
+        return str(x)
+    except Exception:
+        return str(x)
+
+
+# -----------------------------
+# Schema introspection from DuckDB (user tables only)
+# -----------------------------
+def get_schema_summary(con: duckdb.DuckDBPyConnection, sample_rows: int = 3, include_views: bool = True,
+                       allowed_schemas: Optional[List[str]] = None) -> str:
+    """
+    Build a compact schema snapshot for user tables in the *provided DB*.
+    By default we only list tables/views in schema 'main' (user content) and skip internals.
+    """
+    if allowed_schemas is None:
+        allowed_schemas = ["main"]
+
+    type_filter = "('BASE TABLE')" if not include_views else "('BASE TABLE','VIEW')"
+
+    tables_df = con.execute(
+        f"""
+        SELECT table_schema, table_name, table_type
+        FROM information_schema.tables
+        WHERE table_type IN {type_filter}
+          AND table_schema IN ({','.join(['?']*len(allowed_schemas))})
+          AND table_name NOT LIKE 'duckdb_%'
+          AND table_name NOT LIKE 'sqlite_%'
+        ORDER BY table_schema, table_name
+        """,
+        allowed_schemas
+    ).fetchdf()
+
+    lines: List[str] = []
+    for _, row in tables_df.iterrows():
+        schema = row["table_schema"]
+        name = row["table_name"]
+        if name.startswith("__"):
+            # Common ingestion metadata; not user-facing
+            continue
+
+        cols = con.execute(
+            """
+            SELECT column_name, data_type
+            FROM information_schema.columns
+            WHERE table_schema = ? AND table_name = ?
+            ORDER BY ordinal_position
+            """,
+            [schema, name],
+        ).fetchdf()
+
+        lines.append(f"TABLE {schema}.{name}")
+        if len(cols) == 0:
+            lines.append("  (no columns discovered)")
+            continue
+
+        # Sample small set of values per column to guide the LLM
+        try:
+            sample = con.execute(f"SELECT * FROM {schema}.{name} LIMIT {sample_rows}").fetchdf()
+        except Exception:
+            sample = pd.DataFrame(columns=cols["column_name"].tolist())
+
+        for _, c in cols.iterrows():
+            col = c["column_name"]
+            dtype = c["data_type"]
+            examples: List[str] = []
+            if col in sample.columns:
+                for v in sample[col].tolist():
+                    examples.append(format_number(v)[:80])
+            example_str = ", ".join(examples) if examples else ""
+            lines.append(f"  - {col} :: {dtype}  e.g. [{example_str}]")
+        lines.append("")
+
+    if not lines:
+        lines.append("(No user tables discovered in schema(s): " + ", ".join(allowed_schemas) + ")")
+
+    return "\n".join(lines)
+
+
+# -----------------------------
+# LLM helpers
+# -----------------------------
+def make_llm(model: str = "gpt-4o-mini", temperature: float = 0.0) -> ChatOpenAI:
+    return ChatOpenAI(model=model, temperature=temperature)
+
+
+SQL_SYSTEM_PROMPT = """You are an expert DuckDB SQL writer. You will receive:
+- A database schema (tables and columns) with a few sample values
+- A user's natural-language question
+
+Write exactly ONE valid DuckDB SQL SELECT statement to answer the question.
+Rules:
+- Output ONLY the SQL (no backticks, no explanation, no comments)
+- Use DuckDB SQL
+- Never write DDL/DML (CREATE/INSERT/UPDATE/DELETE), only SELECT statements
+- Prefer explicit column names, avoid SELECT *
+- If time grouping is needed, use date_trunc('month', col), date_trunc('day', col), etc.
+- If casting is needed, use CAST(... AS TYPE)
+- Use ONLY the listed user tables from the provided database (e.g., schema 'main'). Do not rely on external/attached sources.
+- Avoid tables starting with "__" unless explicitly referenced.
+- If uncertain, make a reasonable assumption and produce the best possible query
+
+Be careful with joins and filters. Ensure SQL parses successfully.
+"""
+
+REFINE_SYSTEM_PROMPT = """You are fixing a DuckDB SQL query that errored. You'll see:
+- the user's question
+- the previous SQL
+- the exact error message
+
+Return a corrected DuckDB SELECT statement that addresses the error.
+Rules:
+- Output ONLY the SQL (no backticks, no explanation, no comments)
+- SELECT-only (no DDL/DML)
+- Keep the intent of the question intact
+- Fix joins, field names, casts, aggregations or date functions as needed
+- Use ONLY the listed user tables from the provided database (e.g., schema 'main')
+"""
+
+ANSWER_SYSTEM_PROMPT = """You are a helpful data analyst. Given a user's question and the query result data,
+write a clear, concise, conversational answer in plain English. Use bullet points or short paragraphs.
+- Format large numbers with thousands separators where appropriate.
+- If percentages are present, include % signs.
+- If a chart/image accompanies the answer, DO NOT mention file paths or filenames. Refer generically, e.g., 'as depicted in the chart' or 'see the chart alongside this answer.'
+- If the result is empty, say so and suggest a next step.
+- Do not invent columns or values that are not in the result.
+- (Optional) You may refer to prior Q&A context if it informs this answer.
+"""
+
+VIZ_SYSTEM_PROMPT = """You will decide whether a chart would help answer the question using the SQL result.
+Return STRICT JSON with keys: {"make_plot": bool, "chart": "line|bar|scatter|hist|box|pie", "x": "<col or null>", "y": "<col or null>", "series": "<col or null>", "agg": "<sum|avg|count|max|min|null>", "reason": "<short reason>"}.
+Conservative rules (prefer NO plot):
+- Only set make_plot=true if the user explicitly asks for a chart/visualization OR the result involves trends over time, distributions, or comparisons with many rows (e.g., > 20) where a chart adds clarity.
+- For simple aggregates or few rows (<= 10), set make_plot=false unless explicitly requested.
+Guidelines when make_plot=true:
+- Prefer line for trends over time
+- Prefer bar for ranking/comparison of categories
+- Prefer scatter for correlation between two numeric columns
+- Prefer hist for distribution of a single numeric column
+- Prefer box for distribution with quartiles per category
+- Prefer pie sparingly for simple part-to-whole
+JSON only. No prose.
+"""
+
+
+def extract_sql(text: str) -> str:
+    """Extract SQL from potential LLM output. We expect raw SQL with no fences, but be defensive."""
+    text = text.strip()
+    fence = re.compile(r"```(?:sql)?(.*?)```", re.DOTALL | re.IGNORECASE)
+    m = fence.search(text)
+    if m:
+        return m.group(1).strip()
+    return text
+
+
+# -----------------------------
+# LangGraph State
+# -----------------------------
+class AgentState(TypedDict):
+    question: str
+    schema: str
+    attempts: int
+    sql: Optional[str]
+    error: Optional[str]
+    result_json: Optional[str]        # JSON records preview
+    result_columns: Optional[List[str]]
+    plot_path: Optional[str]
+    final_answer: Optional[str]
+    _result_df: Optional[pd.DataFrame]
+    _viz_spec: Optional[Dict[str, Any]]
+    _stream: bool
+    _token_cb: Optional[Callable[[str], None]]
+
+
+# -----------------------------
+# Nodes: SQL drafting, execution, refinement, viz, answer
+# -----------------------------
+def node_draft_sql(state: AgentState, llm: ChatOpenAI) -> AgentState:
+    msgs = [
+        SystemMessage(content=SQL_SYSTEM_PROMPT),
+        HumanMessage(content=f"SCHEMA:\n{state['schema']}\n\nQUESTION:\n{state['question']}"),
+    ]
+    resp = llm.invoke(msgs)  # type: ignore
+    sql = extract_sql(resp.content or "")
+    return {**state, "sql": sql, "error": None}
+
+
+def run_duckdb_query(con: duckdb.DuckDBPyConnection, sql: str) -> pd.DataFrame:
+    first_token = re.split(r"\s+", sql.strip(), maxsplit=1)[0].upper()
+    if first_token != "SELECT" and not sql.strip().upper().startswith("WITH "):
+        raise ValueError("Only SELECT queries are allowed.")
+    df = con.execute(sql).fetchdf()
+    return df
+
+
+def node_run_sql(state: AgentState, con: duckdb.DuckDBPyConnection) -> AgentState:
+    try:
+        df = run_duckdb_query(con, state["sql"] or "")
+        preview = df.head(50).to_dict(orient="records")
+        return {
+            **state,
+            "error": None,
+            "result_json": json.dumps(preview, default=str),
+            "result_columns": list(df.columns),
+            "_result_df": df,
+        }
+    except Exception as e:
+        return {**state, "error": str(e), "result_json": None, "result_columns": None, "_result_df": None}
+
+
+def node_refine_sql(state: AgentState, llm: ChatOpenAI) -> AgentState:
+    if state.get("attempts", 0) >= 3:
+        return state
+    msgs = [
+        SystemMessage(content=REFINE_SYSTEM_PROMPT),
+        HumanMessage(
+            content=(
+                f"QUESTION:\n{state['question']}\n\n"
+                f"PREVIOUS SQL:\n{state.get('sql','')}\n\n"
+                f"ERROR:\n{state.get('error','')}"
+            )
+        ),
+    ]
+    resp = llm.invoke(msgs)  # type: ignore
+    sql = extract_sql(resp.content or "")
+    return {**state, "sql": sql, "attempts": state.get("attempts", 0) + 1, "error": None}
+
+
+def node_decide_viz(state: AgentState, llm: ChatOpenAI) -> AgentState:
+    if not state.get("result_json"):
+        return state
+
+    result_preview = state["result_json"]
+    msgs = [
+        SystemMessage(content=VIZ_SYSTEM_PROMPT),
+        HumanMessage(
+            content=(
+                f"QUESTION:\n{state['question']}\n\n"
+                f"COLUMNS: {state.get('result_columns',[])}\n"
+                f"RESULT PREVIEW (first rows):\n{result_preview}"
+            )
+        ),
+    ]
+    resp = llm.invoke(msgs)  # type: ignore
+
+    spec = {"make_plot": False}
+    try:
+        spec = json.loads(resp.content)  # type: ignore
+        if not isinstance(spec, dict) or "make_plot" not in spec:
+            spec = {"make_plot": False}
+    except Exception:
+        spec = {"make_plot": False}
+
+    return {**state, "_viz_spec": spec}
+
+
+def node_make_plot(state: AgentState) -> AgentState:
+    spec = state.get("_viz_spec") or {}
+    if not spec or not spec.get("make_plot"):
+        return state
+
+    df: Optional[pd.DataFrame] = state.get("_result_df")
+    if df is None or df.empty:
+        return state
+
+    # Additional guard: avoid trivial plots unless explicitly requested
+    q_lower = (state.get('question') or '').lower()
+    explicit_viz = any(k in q_lower for k in ['chart', 'plot', 'graph', 'visual', 'visualize', 'trend'])
+    many_rows = df.shape[0] > 20
+    if not explicit_viz and not many_rows:
+        return state
+
+    x = spec.get("x")
+    y = spec.get("y")
+    series = spec.get("series")
+    chart = (spec.get("chart") or "").lower()
+
+    def col_ok(c: Optional[str]) -> bool:
+        return isinstance(c, str) and c in df.columns
+
+    if not col_ok(x) and df.shape[1] >= 1:
+        x = df.columns[0]
+    if not col_ok(y) and df.shape[1] >= 2:
+        y = df.columns[1]
+
+    try:
+        os.makedirs("plots", exist_ok=True)
+        fig = plt.figure()
+        ax = fig.gca()
+
+        if chart == "line":
+            if series and col_ok(series):
+                for k, g in df.groupby(series):
+                    ax.plot(g[x], g[y], label=str(k))
+                ax.legend(loc="best")
+            else:
+                ax.plot(df[x], df[y])
+            ax.set_xlabel(str(x)); ax.set_ylabel(str(y))
+
+        elif chart == "bar":
+            if series and col_ok(series):
+                pivot = df.pivot_table(index=x, columns=series, values=y, aggfunc="sum", fill_value=0)
+                pivot.plot(kind="bar", ax=ax)
+                ax.set_xlabel(str(x)); ax.set_ylabel(str(y))
+            else:
+                ax.bar(df[x], df[y])
+                ax.set_xlabel(str(x)); ax.set_ylabel(str(y))
+            fig.autofmt_xdate(rotation=45)
+
+        elif chart == "scatter":
+            ax.scatter(df[x], df[y])
+            ax.set_xlabel(str(x)); ax.set_ylabel(str(y))
+
+        elif chart == "hist":
+            ax.hist(df[y] if col_ok(y) else df[x], bins=30)
+            ax.set_xlabel(str(y if col_ok(y) else x)); ax.set_ylabel("Frequency")
+
+        elif chart == "box":
+            ax.boxplot([df[y]] if col_ok(y) else [df[x]])
+            ax.set_ylabel(str(y if col_ok(y) else x))
+
+        elif chart == "pie":
+            labels = df[x].astype(str).tolist()
+            values = df[y].tolist()
+            ax.pie(values, labels=labels, autopct="%1.1f%%")
+            ax.axis("equal")
+
+        else:
+            plt.close(fig)
+            return state
+
+        out_path = os.path.join("plots", f"duckdb_answer_{uuid.uuid4().hex[:8]}.png")
+        fig.tight_layout()
+        fig.savefig(out_path, dpi=150)
+        plt.close(fig)
+        return {**state, "plot_path": out_path}
+
+    except Exception:
+        return state
+
+
+def _stream_llm_answer(llm: ChatOpenAI, msgs: List[Any], token_cb: Optional[Callable[[str], None]]) -> str:
+    """Stream tokens for the final answer; returns full text as well."""
+    out = ""
+    try:
+        for chunk in llm.stream(msgs):  # type: ignore
+            piece = getattr(chunk, "content", None)
+            if not piece and hasattr(chunk, "message") and getattr(chunk, "message", None):
+                piece = getattr(chunk.message, "content", "")
+            if not piece:
+                continue
+            out += piece
+            if token_cb:
+                token_cb(piece)
+    except Exception:
+        # Fallback to non-streaming if stream not supported
+        resp = llm.invoke(msgs)  # type: ignore
+        out = resp.content if isinstance(resp.content, str) else str(resp.content)
+        if token_cb:
+            token_cb(out)
+    return out
+
+
+def node_answer(state: AgentState, llm: ChatOpenAI) -> AgentState:
+    preview = state.get("result_json") or "[]"
+    columns = state.get("result_columns") or []
+    plot_path = state.get("plot_path")
+    if len(preview) > 4000:
+        preview = preview[:4000] + " ..."
+
+    msgs = [
+        SystemMessage(content=ANSWER_SYSTEM_PROMPT),
+        HumanMessage(
+            content=(
+                f"QUESTION:\n{state['question']}\n\n"
+                f"COLUMNS: {columns}\n"
+                f"RESULT PREVIEW (rows):\n{preview}\n\n"
+                f"PLOT_PATH: {plot_path if plot_path else 'None'}"
+            )
+        ),
+    ]
+
+    if state.get("error") and not state.get("result_json"):
+        err_text = (
+            "I couldn't produce a working SQL query after 3 attempts.\n\n"
+            "Details:\n" + (state.get("error") or "Unknown error")
+        )
+        # Stream the error too for consistency
+        if state.get("_stream") and state.get("_token_cb"):
+            state["_token_cb"](err_text)
+        return {**state, "final_answer": err_text}
+
+    # Stream the final answer if requested
+    if state.get("_stream"):
+        answer = _stream_llm_answer(llm, msgs, state.get("_token_cb"))
+    else:
+        resp = llm.invoke(msgs)  # type: ignore
+        answer = resp.content if isinstance(resp.content, str) else str(resp.content)
+
+    return {**state, "final_answer": answer}
+
+
+# -----------------------------
+# Graph assembly
+# -----------------------------
+def build_graph(con: duckdb.DuckDBPyConnection, llm: ChatOpenAI):
+    g = StateGraph(AgentState)
+
+    g.add_node("draft_sql", lambda s: node_draft_sql(s, llm))
+    g.add_node("run_sql", lambda s: node_run_sql(s, con))
+    g.add_node("refine_sql", lambda s: node_refine_sql(s, llm))
+    g.add_node("decide_viz", lambda s: node_decide_viz(s, llm))
+    g.add_node("make_plot", node_make_plot)
+    g.add_node("answer", lambda s: node_answer(s, llm))
+
+    g.set_entry_point("draft_sql")
+    g.add_edge("draft_sql", "run_sql")
+
+    def on_run_sql(state: AgentState):
+        if state.get("error"):
+            if state.get("attempts", 0) < 3:
+                return "refine_sql"
+            else:
+                return "answer"
+        return "decide_viz"
+
+    g.add_conditional_edges("run_sql", on_run_sql, {
+        "refine_sql": "refine_sql",
+        "decide_viz": "decide_viz",
+        "answer": "answer",
+    })
+
+    g.add_edge("refine_sql", "run_sql")
+
+    def on_decide_viz(state: AgentState):
+        spec = state.get("_viz_spec") or {}
+        if spec.get("make_plot"):
+            return "make_plot"
+        return "answer"
+
+    g.add_conditional_edges("decide_viz", on_decide_viz, {
+        "make_plot": "make_plot",
+        "answer": "answer",
+    })
+
+    g.add_edge("make_plot", "answer")
+    g.add_edge("answer", END)
+
+    return g.compile()
+
+
+# -----------------------------
+# Public function to ask a question
+# -----------------------------
+def answer_question(con: duckdb.DuckDBPyConnection, llm: ChatOpenAI, schema_text: str, question: str,
+                    stream: bool = False, token_callback: Optional[Callable[[str], None]] = None) -> Dict[str, Any]:
+    app = build_graph(con, llm)
+    initial: AgentState = {
+        "question": question,
+        "schema": schema_text,
+        "attempts": 0,
+        "sql": None,
+        "error": None,
+        "result_json": None,
+        "result_columns": None,
+        "plot_path": None,
+        "final_answer": None,
+        "_result_df": None,
+        "_viz_spec": None,
+        "_stream": stream,
+        "_token_cb": token_callback,
+    }
+    final_state: AgentState = app.invoke(initial)  # type: ignore
+
+    return {
+        "question": question,
+        "sql": final_state.get("sql"),
+        "answer": final_state.get("final_answer"),
+        "plot_path": final_state.get("plot_path"),
+        "error": final_state.get("error"),
+    }
+
+
+# -----------------------------
+# CLI
+# -----------------------------
+def main():
+    load_dotenv(find_dotenv())
+
+    parser = argparse.ArgumentParser(description="DuckDB Q&A ReAct Agent (Streaming)")
+    parser.add_argument("--duckdb", required=True, help="Path to DuckDB database file")
+    parser.add_argument("--model", default="gpt-4o", help="OpenAI chat model (default: gpt-4o)")
+    parser.add_argument("--stream", action="store_true", help="Stream the final answer to stdout")
+    parser.add_argument("--schemas", nargs="*", default=["main"], help="Schemas to include (default: main)")
+    args = parser.parse_args()
+
+    api_key = os.environ.get("OPENAI_API_KEY")
+    if not api_key:
+        print("ERROR: OPENAI_API_KEY not set. Put it in a .env file or export the env var.")
+        return
+
+    # Connect DuckDB strictly to the provided file (user DB)
+    try:
+        con = duckdb.connect(database=args.duckdb, read_only=True)
+    except Exception as e:
+        print(f"Failed to open DuckDB at {args.duckdb}: {e}")
+        return
+
+    # Introspect schema (user tables only)
+    schema_text = get_schema_summary(con, allowed_schemas=args.schemas)
+
+    # Init LLM; enable streaming capability (used only in final answer node)
+    llm = make_llm(model=args.model, temperature=0.0)
+
+    print("DuckDB Q&A Agent (ReAct, Streaming)\n")
+    print(f"Connected to: {args.duckdb}")
+    print(f"Schemas included: {', '.join(args.schemas)}")
+    print("\nSchema snapshot:\n-----------------\n")
+    print(schema_text)
+    print("\nType your question and press ENTER. Type 'exit' to quit.\n")
+
+    def print_token(t: str):
+        print(t, end="", flush=True)
+
+    while True:
+        try:
+            q = input("Q> ").strip()
+        except (EOFError, KeyboardInterrupt):
+            print("\nExiting.")
+            break
+        if q.lower() in ("exit", "quit"):
+            print("Goodbye.")
+            break
+        if not q:
+            continue
+
+        if args.stream:
+            print("\n--- ANSWER (streaming) ---")
+            result = answer_question(con, llm, schema_text, q, stream=True, token_callback=print_token)
+            print("")  # newline after stream
+            if result.get("plot_path"):
+                print(f"\nChart saved to: {result['plot_path']}")
+            print("\n--- SQL ---")
+            print((result.get("sql") or "").strip())
+            if result.get("error"):
+                print("\nERROR: " + str(result.get("error")))
+        else:
+            result = answer_question(con, llm, schema_text, q, stream=False, token_callback=None)
+            print("\n--- SQL ---")
+            print((result.get("sql") or "").strip())
+            if result.get("error"):
+                print("\n--- RESULT ---")
+                print("Sorry, I couldn't resolve a working query after 3 attempts.")
+                print("Error: " + str(result.get("error")))
+            else:
+                print("\n--- ANSWER ---")
+                print(result.get("answer") or "")
+                if result.get("plot_path"):
+                    print(f"\nChart saved to: {result['plot_path']}")
+        print("\n")
+
+    con.close()
+
+
+if __name__ == "__main__":
+    main()

excel_to_duckdb.py

@@ -0,0 +1,595 @@
+"""
+Excel → DuckDB ingestion (generic, robust, multi-table)
+
+- Hierarchical headers with merged-cell parent context (titles removed)
+- Merged rows/cols resolved to master (top-left) value for consistent replication
+- Multiple tables detected ONLY when separated by at least one completely empty row
+- Footer detection (ignore trailing notes/summaries)
+- Pivot detection (skip pivot-looking rows; optional sheet-level pivot/charthood skip)
+- Optional LLM inference for unnamed columns and table titles (EXCEL_LLM_INFER=1)
+- One DuckDB table per detected table block
+- Traceability:
+    __excel_schema  (sheet_name, table_name, column_ordinal, original_name, sql_column)
+    __excel_tables  (sheet_name, table_name, block_index, start_row, end_row,
+                    header_rows_json, inferred_title, original_title_text)
+
+Usage:
+  python excel_to_duckdb.py --excel /path/file.xlsx --duckdb /path/out.duckdb
+"""
+
+import os
+import re
+import sys
+import json
+import argparse
+from pathlib import Path
+from typing import List, Tuple, Dict
+import hashlib
+
+from openpyxl import load_workbook
+from openpyxl.worksheet.worksheet import Worksheet
+
+
+def _nonempty(vals):
+    return [v for v in vals if v not in (None, "")]
+
+def _is_numlike(x):
+    if isinstance(x, (int, float)):
+        return True
+    s = str(x).strip().replace(",", "")
+    if s.endswith("%"):
+        s = s[:-1]
+    if not s:
+        return False
+    if any(c.isalpha() for c in s):
+        return False
+    try:
+        float(s); return True
+    except: return False
+
+def _is_year_token(x):
+    if isinstance(x, int) and 1800 <= x <= 2100: return True
+    s = str(x).strip()
+    return s.isdigit() and 1800 <= int(s) <= 2100
+
+def is_probably_footer(cells):
+    nonempty = [(i, v) for i, v in enumerate(cells) if v not in (None, "")]
+    if not nonempty: return False
+    if len(nonempty) <= 2:
+        text = " ".join(str(v) for _, v in nonempty).strip().lower()
+        if any(text.startswith(k) for k in ["note","notes","source","summary","disclaimer"]): return True
+        if len(text) > 50: return True
+    return False
+
+def is_probably_data(cells, num_cols):
+    vals = [v for v in cells if v not in (None, "")]
+    if not vals: return False
+    nums_list = [v for v in vals if _is_numlike(v)]
+    num_num = len(nums_list); num_text = len(vals) - num_num
+    density = len(vals) / max(1, num_cols)
+    if num_num >= 2 and all(_is_year_token(v) for v in nums_list) and num_text >= 2:
+        return False
+    if num_num >= max(2, num_text): return True
+    if density >= 0.6 and num_num >= 2: return True
+    first = str(vals[0]).strip().lower() if vals else ""
+    if first in ("total","totals","grand total"): return True
+    return False
+
+PIVOT_MARKERS = {"row labels","column labels","values","grand total","report filter","filters","∑ values","σ values","Σ values"}
+def is_pivot_marker_string(s: str) -> bool:
+    if not s: return False
+    t = str(s).strip().lower()
+    if t in PIVOT_MARKERS: return True
+    if t.startswith(("sum of ","count of ","avg of ","average of ")): return True
+    if t.endswith(" total") or t.startswith("total "): return True
+    return False
+
+def is_pivot_row(cells) -> bool:
+    text_cells = [str(v).strip() for v in cells if v not in (None, "")]
+    if not text_cells: return False
+    if any(is_pivot_marker_string(x) for x in text_cells): return True
+    agg_hits = sum(1 for x in text_cells if x.lower().startswith(("sum of","count of","avg of","average of","min of","max of")))
+    return agg_hits >= 2
+
+def is_pivot_or_chart_sheet(ws: Worksheet) -> bool:
+    try:
+        if getattr(ws, "_charts", None): return True
+    except Exception: pass
+    if hasattr(ws, "_pivots") and getattr(ws, "_pivots"): return True
+    scan_rows = min(ws.max_row, 40); scan_cols = min(ws.max_column, 20)
+    pivotish = 0
+    for r in range(1, scan_rows+1):
+        row = [ws.cell(r,c).value for c in range(1, scan_cols+1)]
+        if is_pivot_row(row):
+            pivotish += 1
+            if pivotish >= 2: return True
+    name = (ws.title or "").lower()
+    if any(k in name for k in ("pivot","dashboard","chart","charts")): return True
+    return False
+
+def sanitize_table_name(name: str) -> str:
+    t = re.sub(r"[^\w]", "_", str(name))
+    t = re.sub(r"_+", "_", t).strip("_")
+    if t and not t[0].isalpha(): t = "table_" + t
+    return t or "sheet_data"
+
+def _samples_for_column(rows, col_idx, max_items=20):
+    vals = []
+    for row in rows:
+        if col_idx < len(row):
+            v = row[col_idx]
+            if v not in (None, ""): vals.append(v)
+        if len(vals) >= max_items: break
+    return vals
+
+def _heuristic_infer_col_name(samples):
+    if not samples: return None
+    if sum(1 for v in samples if _is_year_token(v)) >= max(2, int(0.8*len(samples))): return "year"
+    pct_hits = 0
+    for v in samples:
+        s = str(v).strip()
+        if s.endswith("%"): pct_hits += 1
+        else:
+            try:
+                f = float(s.replace(",",""))
+                if 0 <= f <= 1.0 or 0 <= f <= 100: pct_hits += 0.5
+            except: pass
+    if pct_hits >= max(2, int(0.7*len(samples))): return "percentage"
+    if sum(1 for v in samples if _is_numlike(v)) >= max(3, int(0.7*len(samples))):
+        intish = 0
+        for v in samples:
+            try:
+                if float(str(v).replace(",","")) == int(float(str(v).replace(",",""))): intish += 1
+            except: pass
+        if intish >= max(2, int(0.6*len(samples))): return "count"
+        return "value"
+    uniq = {str(v).strip().lower() for v in samples}
+    if len(uniq) <= 3 and max(len(str(v)) for v in samples) >= 30: return "question"
+    if sum(1 for v in samples if re.search(r"\d", str(v)) and ("-" in str(v) or "–" in str(v))) >= max(2, int(0.6*len(samples))): return "range"
+    if len(uniq) < max(5, int(0.5*len(samples))): return "category"
+    return None
+
+def clean_col_name(s: str) -> str:
+    s = re.sub(r"[^\w\s%#‰]", "", str(s).strip())
+    s = s.replace("%"," pct").replace("‰"," permille").replace("#"," count ")
+    s = re.sub(r"\s+"," ", s)
+    s = re.sub(r"\s+","_", s)
+    s = re.sub(r"_+","_", s).strip("_")
+    if s and s[0].isdigit(): s = "col_" + s
+    return s or "unnamed_column"
+
+def ensure_unique(names):
+    seen = {}; out = []
+    for n in names:
+        base = (n or "unnamed_column").lower()
+        if base not in seen:
+            seen[base] = 0; out.append(n)
+        else:
+            i = seen[base] + 1
+            while f"{n}_{i}".lower() in seen: i += 1
+            seen[base] = i; out.append(f"{n}_{i}")
+            seen[(f"{n}_{i}").lower()] = 0
+    return out
+
+def compose_col(parts):
+    cleaned = []; prev = None
+    for p in parts:
+        if not p: continue
+        p_norm = str(p).strip()
+        if prev is not None and p_norm.lower() == prev.lower(): continue
+        cleaned.append(p_norm); prev = p_norm
+    if not cleaned: return ""
+    return clean_col_name("_".join(cleaned))
+
+def used_bounds(ws: Worksheet) -> Tuple[int,int,int,int]:
+    min_row, max_row, min_col, max_col = None, 0, None, 0
+    for r in ws.iter_rows():
+        for c in r:
+            v = c.value
+            if v is not None and str(v).strip() != "":
+                if min_row is None or c.row < min_row: min_row = c.row
+                if c.row > max_row: max_row = c.row
+                if min_col is None or c.column < min_col: min_col = c.column
+                if c.column > max_col: max_col = c.column
+    if min_row is None: return 1,0,1,0
+    return min_row, max_row, min_col, max_col
+
+def build_merged_master_map(ws: Worksheet):
+    mapping = {}
+    for mr in ws.merged_cells.ranges:
+        min_col, min_row, max_col, max_row = mr.min_col, mr.min_row, mr.max_col, mr.max_row
+        master = (min_row, min_col)
+        for r in range(min_row, max_row+1):
+            for c in range(min_col, max_col+1):
+                mapping[(r,c)] = master
+    return mapping
+
+def build_value_grid(ws: Worksheet, min_row: int, max_row: int, min_col: int, max_col: int):
+    merged_map = build_merged_master_map(ws)
+    nrows = max_row - min_row + 1; ncols = max_col - min_col + 1
+    grid = [[None]*ncols for _ in range(nrows)]
+    for r in range(min_row, max_row+1):
+        rr = r - min_row
+        for c in range(min_col, max_col+1):
+            cc = c - min_col
+            master = merged_map.get((r,c))
+            if master:
+                mr, mc = master; grid[rr][cc] = ws.cell(mr, mc).value
+            else:
+                grid[rr][cc] = ws.cell(r, c).value
+    return grid
+
+def row_vals_from_grid(grid, r, min_row):
+    return grid[r - min_row]
+
+def is_empty_row_vals(vals):
+    return not any(v not in (None, "") for v in vals)
+
+def is_title_like_row_vals(vals, total_cols=20):
+    vals_ne = _nonempty(vals)
+    if not vals_ne: return False
+    if len(vals_ne) == 1: return True
+    coverage = len(vals_ne) / max(1, total_cols)
+    if coverage <= 0.2 and all(isinstance(v,str) and len(str(v))>20 for v in vals_ne): return True
+    uniq = {str(v).strip().lower() for v in vals_ne}
+    if len(uniq) == 1: return True
+    block = {"local currency unit per us dollar","exchange rate","average annual exchange rate"}
+    if any(str(v).strip().lower() in block for v in vals_ne): return True
+    return False
+
+def is_header_candidate_row_vals(vals, total_cols=20):
+    vals_ne = _nonempty(vals)
+    if not vals_ne: return False
+    if is_title_like_row_vals(vals, total_cols): return False
+    nums = sum(1 for v in vals_ne if _is_numlike(v))
+    years = sum(1 for v in vals_ne if _is_year_token(v))
+    has_text = any(not _is_numlike(v) for v in vals_ne)
+    if years >= 2 and has_text: return True
+    if nums >= max(2, len(vals_ne)-nums): return years >= max(2, int(0.6*len(vals_ne)))
+    uniq_labels = {str(v).strip().lower() for v in vals_ne if not _is_numlike(v)}
+    return (len(vals_ne) >= 2) or (len(uniq_labels) >= 2)
+
+def detect_tables_fast(ws: Worksheet, grid, min_row, max_row, min_col, max_col):
+    blocks = []
+    if is_pivot_or_chart_sheet(ws): return blocks
+    total_cols = max_col - min_col + 1
+    r = min_row
+    while r <= max_row:
+        vals = row_vals_from_grid(grid, r, min_row)
+        if is_empty_row_vals(vals) or is_title_like_row_vals(vals, total_cols) or is_pivot_row(vals):
+            r += 1; continue
+        if not is_probably_data(vals, total_cols):
+            r += 1; continue
+        data_start = r
+        header_rows = []
+        up = data_start - 1
+        while up >= min_row:
+            vup = row_vals_from_grid(grid, up, min_row)
+            if is_empty_row_vals(vup): break
+            if is_title_like_row_vals(vup, total_cols) or is_pivot_row(vup):
+                up -= 1; continue
+            if is_header_candidate_row_vals(vup, total_cols):
+                header_rows = []
+                hdr_row = up
+                while hdr_row >= min_row:
+                    hdr_vals = row_vals_from_grid(grid, hdr_row, min_row)
+                    if is_empty_row_vals(hdr_vals): break
+                    if is_header_candidate_row_vals(hdr_vals, total_cols):
+                        header_rows.insert(0, hdr_row); hdr_row -= 1
+                    else: break
+            break
+        data_end = data_start
+        rr = data_start + 1
+        while rr <= max_row:
+            v = row_vals_from_grid(grid, rr, min_row)
+            if is_probably_footer(v) or is_pivot_row(v): break
+            if is_empty_row_vals(v): break
+            if is_probably_data(v, total_cols) or is_header_candidate_row_vals(v, total_cols):
+                data_end = rr
+            rr += 1
+        title_text = None
+        if header_rows:
+            top = header_rows[0]
+            for tr in range(max(min_row, top-3), top):
+                tv = row_vals_from_grid(grid, tr, min_row)
+                if is_title_like_row_vals(tv, total_cols):
+                    first = next((str(x).strip() for x in tv if x not in (None,"")), None)
+                    if first: title_text = first
+                    break
+        if (header_rows or data_end - data_start >= 1) and data_start <= data_end:
+            blocks.append({"header_rows": header_rows, "data_start": data_start, "data_end": data_end, "title_text": title_text})
+        r = data_end + 1
+        while r <= max_row and is_empty_row_vals(row_vals_from_grid(grid, r, min_row)):
+            r += 1
+    return blocks
+
+def expand_headers_from_grid(grid, header_rows, min_row, min_col, eff_max_col):
+    if not header_rows: return []
+    mat = []
+    for r in header_rows:
+        row_vals = row_vals_from_grid(grid, r, min_row)
+        row = [("" if (row_vals[c] is None) else str(row_vals[c]).strip()) for c in range(0, eff_max_col)]
+        last = ""
+        for i in range(len(row)):
+            if row[i] == "" and i > 0: row[i] = last
+            else: last = row[i]
+        mat.append(row)
+    return mat
+
+def sheet_block_to_df_fast(ws, grid, min_row, max_row, min_col, max_col, header_rows, data_start, data_end):
+    import pandas as pd
+    total_cols = max_col - min_col + 1
+    if (not header_rows) and data_start and data_start > min_row:
+        prev = row_vals_from_grid(grid, data_start - 1, min_row)
+        if is_header_candidate_row_vals(prev, total_cols):
+            header_rows = [data_start - 1]
+    if (not header_rows) and data_start:
+        cur = row_vals_from_grid(grid, data_start, min_row)
+        nxt = row_vals_from_grid(grid, data_start + 1, min_row) if data_start + 1 <= max_row else []
+        if is_header_candidate_row_vals(cur, total_cols) and is_probably_data(nxt, total_cols):
+            header_rows = [data_start]; data_start += 1
+    if not header_rows or data_start is None or data_end is None or data_end < data_start:
+        return pd.DataFrame(), [], []
+    def used_upto_col():
+        maxc = 0
+        for r in list(header_rows) + list(range(data_start, data_end+1)):
+            vals = row_vals_from_grid(grid, r, min_row)
+            for c_off in range(total_cols):
+                v = vals[c_off]
+                if v not in (None, ""): maxc = max(maxc, c_off+1)
+        return maxc or total_cols
+    eff_max_col = used_upto_col()
+    header_mat = expand_headers_from_grid(grid, header_rows, min_row, min_col, eff_max_col)
+    def is_title_level(values):
+        total = len(values)
+        filled = [str(v).strip() for v in values if v not in (None, "")]
+        if total == 0: return False
+        coverage = len(filled) / total
+        if coverage <= 0.2 and len(filled) <= 2: return True
+        if filled:
+            uniq = {v.lower() for v in filled}
+            if len(uniq) == 1:
+                label = next(iter(uniq))
+                dom = sum(1 for v in values if isinstance(v,str) and v.strip().lower() == label)
+                if dom / total >= 0.6: return True
+        return False
+    usable_levels = [i for i in range(len(header_mat)) if not is_title_level(header_mat[i])]
+    if not usable_levels and header_mat: usable_levels = [len(header_mat) - 1]
+    cols = []
+    for c_off in range(eff_max_col):
+        parts = [header_mat[l][c_off] for l in range(usable_levels[0], usable_levels[-1]+1)] if usable_levels else []
+        cols.append(compose_col(parts))
+    cols = ensure_unique([clean_col_name(x) for x in cols])
+    data_rows = []
+    for r in range(data_start, data_end+1):
+        vals = row_vals_from_grid(grid, r, min_row)
+        row = [vals[c_off] for c_off in range(eff_max_col)]
+        if is_probably_footer(row): break
+        data_rows.append(row[:len(cols)])
+    if not data_rows: return pd.DataFrame(columns=cols), header_mat, cols
+    keep_mask = [any(row[i] not in (None, "") for row in data_rows) for i in range(len(cols))]
+    kept_cols = [c for c,k in zip(cols, keep_mask) if k]
+    trimmed_rows = [[v for v,k in zip(row, keep_mask) if k] for row in data_rows]
+    df = pd.DataFrame(trimmed_rows, columns=kept_cols)
+    if any(str(c).startswith("unnamed_column") for c in df.columns):
+        new_names = list(df.columns)
+        for idx, name in enumerate(list(df.columns)):
+            if not str(name).startswith("unnamed_column"): continue
+            samples = _samples_for_column(trimmed_rows, idx, max_items=20)
+            guess = _heuristic_infer_col_name(samples)
+            if guess: new_names[idx] = clean_col_name(guess)
+        df.columns = ensure_unique([clean_col_name(x) for x in new_names])
+    return df, header_mat, kept_cols
+
+def _llm_infer_table_title(header_mat, sample_rows, sheet_name):
+    if os.environ.get("EXCEL_LLM_INFER","0") != "1": return None
+    api_key = os.environ.get("OPENAI_API_KEY")
+    if not api_key: return None
+    headers = []
+    if header_mat:
+        for c in range(len(header_mat[0])):
+            parts = [header_mat[l][c] for l in range(len(header_mat))]
+            parts = [p for p in parts if p]
+            if parts: headers.append("_".join(parts))
+        headers = headers[:10]
+    samples = [[str(x) for x in r[:6]] for r in sample_rows[:5]]
+    prompt = (
+        "Propose a short, human-readable title for a data table.\n"
+        "Keep it 3-6 words, Title Case, no punctuation at the end.\n"
+        f"Sheet: {sheet_name}\nHeaders: {headers}\nRow samples: {samples}\n"
+        "Answer with JSON: {\"title\": \"...\"}"
+    )
+    try:
+        from openai import OpenAI
+        client = OpenAI(api_key=api_key)
+        resp = client.chat.completions.create(
+            model=os.environ.get("OPENAI_MODEL","gpt-4o-mini"),
+            messages=[{"role":"user","content":prompt}], temperature=0.2,
+        )
+        text = resp.choices[0].message.content.strip()
+    except Exception:
+        return None
+    import re, json as _json
+    m = re.search(r"\{.*\}", text, re.S)
+    if not m: return None
+    try:
+        obj = _json.loads(m.group(0)); title = obj.get("title","").strip()
+        return title or None
+    except Exception: return None
+
+def _heuristic_table_title(header_mat, sheet_name, idx):
+    if header_mat:
+        parts = []
+        levels = len(header_mat)
+        cols = len(header_mat[0]) if header_mat else 0
+        for c in range(min(6, cols)):
+            colparts = [header_mat[l][c] for l in range(min(levels, 2)) if header_mat[l][c]]
+            if colparts: parts.extend(colparts)
+        if parts:
+            base = " ".join(dict.fromkeys(parts))
+            return base[:60]
+    return f"{sheet_name} Table {idx}"
+
+def infer_table_title(header_mat, sample_rows, sheet_name, idx):
+    title = _heuristic_table_title(header_mat, sheet_name, idx)
+    llm = _llm_infer_table_title(header_mat, sample_rows, sheet_name)
+    return llm or title
+
+def ensure_unique_table_name(existing: set, name: str) -> str:
+    base = sanitize_table_name(name) or "table"
+    if base not in existing:
+        existing.add(base); return base
+    i = 2
+    while f"{base}_{i}" in existing: i += 1
+    out = f"{base}_{i}"; existing.add(out); return out
+
+# --- NEW: block coalescing to avoid nested/overlapping duplicates ---
+def coalesce_blocks(blocks: List[Dict]) -> List[Dict]:
+    """Keep only maximal non-overlapping blocks by data row range."""
+    if not blocks: return blocks
+    blocks_sorted = sorted(blocks, key=lambda b: (b["data_start"], b["data_end"]))
+    result = []
+    for b in blocks_sorted:
+        if any(b["data_start"] >= x["data_start"] and b["data_end"] <= x["data_end"] for x in result):
+            # fully contained -> drop
+            continue
+        result.append(b)
+    return result
+
+# ------------------------- Persistence -------------------------
+def persist(excel_path, duckdb_path):
+    try:
+        from duckdb import connect
+    except ImportError:
+        print("Error: DuckDB library not installed. Install with: pip install duckdb"); sys.exit(1)
+    try:
+        wb = load_workbook(excel_path, data_only=True)
+    except FileNotFoundError:
+        print(f"Error: Excel file not found: {excel_path}"); sys.exit(1)
+    except Exception as e:
+        print(f"Error loading Excel file: {e}"); sys.exit(1)
+
+    db_path = Path(duckdb_path)
+    db_path.parent.mkdir(parents=True, exist_ok=True)
+    new_db = not db_path.exists()
+    con = connect(str(db_path))
+    if new_db: print(f"Created new DuckDB at: {db_path}")
+
+    con.execute("""
+        CREATE TABLE IF NOT EXISTS __excel_schema (
+            sheet_name TEXT,
+            table_name TEXT,
+            column_ordinal INTEGER,
+            original_name TEXT,
+            sql_column TEXT
+        )
+    """)
+    con.execute("""
+        CREATE TABLE IF NOT EXISTS __excel_tables (
+            sheet_name TEXT,
+            table_name TEXT,
+            block_index INTEGER,
+            start_row INTEGER,
+            end_row INTEGER,
+            header_rows_json TEXT,
+            inferred_title TEXT,
+            original_title_text TEXT
+        )
+    """)
+    con.execute("DELETE FROM __excel_schema")
+    con.execute("DELETE FROM __excel_tables")
+
+    used_names = set(); total_tables = 0; total_rows = 0
+
+    for sheet in wb.sheetnames:
+        ws = wb[sheet]
+        try:
+            if not isinstance(ws, Worksheet):
+                print(f"Skipping chartsheet: {sheet}"); continue
+        except Exception: pass
+        if is_pivot_or_chart_sheet(ws):
+            print(f"Skipping pivot/chart-like sheet: {sheet}"); continue
+
+        min_row, max_row, min_col, max_col = used_bounds(ws)
+        if max_row < min_row: continue
+        grid = build_value_grid(ws, min_row, max_row, min_col, max_col)
+
+        blocks = detect_tables_fast(ws, grid, min_row, max_row, min_col, max_col)
+        blocks = coalesce_blocks(blocks)  # NEW: drop contained duplicates
+        if not blocks: continue
+
+        # NEW: per-sheet content hash set to avoid identical duplicate content
+        seen_content = set()
+
+        for idx, blk in enumerate(blocks, start=1):
+            df, header_mat, kept_cols = sheet_block_to_df_fast(
+                ws, grid, min_row, max_row, min_col, max_col,
+                blk["header_rows"], blk["data_start"], blk["data_end"]
+            )
+            if df.empty: continue
+
+            # Content hash (stable CSV representation)
+            csv_bytes = df.to_csv(index=False).encode("utf-8")
+            h = hashlib.sha256(csv_bytes).hexdigest()
+            if h in seen_content:
+                # Skip duplicate block with identical content
+                print(f"Skipping duplicate content on sheet {sheet} (block {idx})")
+                continue
+            seen_content.add(h)
+
+            # Build original composite header names for lineage mapping
+            original_cols = []
+            if header_mat:
+                levels = len(header_mat)
+                cols = len(header_mat[0]) if header_mat else 0
+                for c in range(cols):
+                    parts = [header_mat[l][c] for l in range(levels)]
+                    original_cols.append("_".join([p for p in parts if p]))
+            else:
+                original_cols = list(df.columns)
+            while len(original_cols) < len(df.columns): original_cols.append("unnamed")
+
+            title_orig = blk.get("title_text")
+            title = title_orig or infer_table_title(header_mat, df.values.tolist(), sheet, idx)
+            candidate = title if title else f"{sheet} Table {idx}"
+            table = ensure_unique_table_name(used_names, candidate)
+
+            con.execute(f'DROP TABLE IF EXISTS "{table}"')
+            con.register(f"{table}_temp", df)
+            con.execute(f'CREATE TABLE "{table}" AS SELECT * FROM {table}_temp')
+            con.unregister(f"{table}_temp")
+
+            for cidx, (orig, sqlc) in enumerate(zip(original_cols[:len(df.columns)], df.columns), start=1):
+                con.execute("INSERT INTO __excel_schema VALUES (?, ?, ?, ?, ?)", [sheet, table, cidx, str(orig), str(sqlc)])
+
+            con.execute(
+                """
+                INSERT INTO __excel_tables
+                (sheet_name, table_name, block_index, start_row, end_row, header_rows_json, inferred_title, original_title_text)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                [sheet, table, int(idx), int(blk["data_start"]), int(blk["data_end"]), json.dumps(blk["header_rows"]), title if title else None, title_orig if title_orig else None],
+            )
+
+            print(f"Created table {table} from sheet {sheet} with {len(df)} rows and {len(df.columns)} columns.")
+            total_tables += 1; total_rows += len(df)
+
+    con.close()
+    print(f"""\n✅ Completed.
+   - Created {total_tables} tables with {total_rows} total rows
+   - Column lineage: __excel_schema
+   - Block metadata: __excel_tables""")
+
+
+def main():
+    ap = argparse.ArgumentParser(description="Excel → DuckDB (v2-compatible perf + no-dup safe guards).")
+    ap.add_argument("--excel", required=True, help="Path to .xlsx")
+    ap.add_argument("--duckdb", required=True, help="Path to DuckDB file")
+    args = ap.parse_args()
+    if not os.path.exists(args.excel):
+        print(f"Error: Excel file not found: {args.excel}"); sys.exit(1)
+    persist(args.excel, args.duckdb)
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -1,3 +1,13 @@
-altair
-pandas
-streamlit
+duckdb 
+pandas 
+openpyxl 
+xlsxwriter
+langgraph 
+langchain-openai
+langchain-core
+faker
+python-dotenv==1.0.1
+streamlit
+matplotlib 
+openai
+tabulate

streamlit_app.py

@@ -0,0 +1,574 @@
+import os
+import sys
+import duckdb
+import json
+import subprocess
+from pathlib import Path
+from typing import Dict, List, Tuple
+
+import streamlit as st
+
+
+try:
+    from dotenv import load_dotenv, find_dotenv
+    load_dotenv(find_dotenv())
+except Exception:
+    pass
+
+# ---------- Basic page setup ----------
+st.set_page_config(page_title="Excel → Dataset", page_icon="📊", layout="wide")
+
+PRIMARY_DIR = Path(__file__).parent.resolve()
+UPLOAD_DIR = PRIMARY_DIR / "uploads"
+DB_DIR = PRIMARY_DIR / "dbs"
+SCRIPT_PATH = PRIMARY_DIR / "excel_to_duckdb.py"  # must be colocated
+
+UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
+DB_DIR.mkdir(parents=True, exist_ok=True)
+
+st.markdown(
+    """
+    <style>
+      .logbox { border: 1px solid #e5e7eb; background:#fafafa; padding:10px; border-radius:12px; }
+      .logbox code { white-space: pre-wrap; font-size: 0.85rem; }
+    </style>
+    """,
+    unsafe_allow_html=True,
+)
+
+st.title("Data Analysis Agent")
+# --------- Session state helpers ---------
+if "processing" not in st.session_state:
+    st.session_state.processing = False
+if "processed_key" not in st.session_state:
+    st.session_state.processed_key = None
+if "last_overview_md" not in st.session_state:
+    st.session_state.last_overview_md = None
+if "last_preview_items" not in st.session_state:
+    st.session_state.last_preview_items = []  # list of dicts: {'table_ref': str, 'label': str}
+
+def _file_key(uploaded) -> str:
+    # unique-ish key per upload (name + size)
+    try:
+        size = len(uploaded.getbuffer())
+    except Exception:
+        size = 0
+    return f"{uploaded.name}:{size}"
+
+# ---------- DuckDB helpers ----------
+def list_user_tables(con: duckdb.DuckDBPyConnection) -> List[str]:
+    """
+    Robust discovery:
+      1) information_schema.tables for BASE TABLE/VIEW in any schema, excluding __*
+      2) duckdb_tables() as a secondary path
+      3) __excel_tables mapping + existence check as last resort
+    """
+    # 1) information_schema (most portable)
+    try:
+        q = (
+            "SELECT table_schema, table_name "
+            "FROM information_schema.tables "
+            "WHERE table_type IN ('BASE TABLE','VIEW') "
+            "AND table_name NOT LIKE '__%%' "
+            "ORDER BY table_schema, table_name"
+        )
+        rows = con.execute(q).fetchall()
+        names = []
+        for schema, name in rows:
+            if (schema or '').lower() == 'main':
+                names.append(name)
+            else:
+                names.append(f'{schema}."{name}"')
+        if names:
+            return names
+    except Exception:
+        pass
+
+    # 2) duckdb_tables()
+    try:
+        q2 = (
+            "SELECT schema_name, table_name "
+            "FROM duckdb_tables() "
+            "WHERE table_type = 'BASE TABLE' "
+            "AND table_name NOT LIKE '__%%' "
+            "ORDER BY schema_name, table_name"
+        )
+        rows = con.execute(q2).fetchall()
+        names = []
+        for schema, name in rows:
+            if (schema or '').lower() == 'main':
+                names.append(name)
+            else:
+                names.append(f'{schema}."{name}"')
+        if names:
+            return names
+    except Exception:
+        pass
+
+    # 3) Fallback to metadata table
+    try:
+        meta = con.execute("SELECT DISTINCT table_name FROM __excel_tables").fetchall()
+        names = []
+        for (t,) in meta:
+            try:
+                con.execute(f'SELECT 1 FROM "{t}" LIMIT 1').fetchone()
+                names.append(t)
+            except Exception:
+                continue
+        return names
+    except Exception:
+        return []
+
+def get_columns(con: duckdb.DuckDBPyConnection, table: str) -> List[Tuple[str,str]]:
+    # Normalize table name for information_schema lookup
+    if table.lower().startswith("main."):
+        tname = table.split('.', 1)[1].strip('"')
+        schema_filter = 'main'
+    elif '.' in table:
+        schema_filter, tname_raw = table.split('.', 1)
+        tname = tname_raw.strip('"')
+    else:
+        schema_filter = 'main'
+        tname = table.strip('"')
+    q = (
+        "SELECT column_name, data_type "
+        "FROM information_schema.columns "
+        "WHERE table_schema=? AND table_name=? "
+        "ORDER BY ordinal_position"
+    )
+    return con.execute(q, [schema_filter, tname]).fetchall()
+
+def detect_year_column(con, table: str, col: str) -> bool:
+    try:
+        sql = (
+            f'SELECT AVG(CASE WHEN TRY_CAST("{col}" AS INTEGER) BETWEEN 1900 AND 2100 '
+            f'THEN 1.0 ELSE 0.0 END) FROM {table}'
+        )
+        v = con.execute(sql).fetchone()[0]
+        return (v or 0) > 0.7
+    except Exception:
+        return False
+
+def role_of_column(con, table: str, col: str, dtype: str) -> str:
+    d = (dtype or '').upper()
+    if any(tok in d for tok in ["DATE", "TIMESTAMP"]):
+        return "date"
+    if any(tok in d for tok in ["INT", "BIGINT", "DOUBLE", "DECIMAL", "FLOAT", "HUGEINT", "REAL"]):
+        if detect_year_column(con, table, col):
+            return "year"
+        return "numeric"
+    if any(tok in d for tok in ["CHAR", "STRING", "TEXT", "VARCHAR"]):
+        try:
+            sql = f'SELECT COUNT(*), COUNT(DISTINCT "{col}") FROM {table}'
+            n, nd = con.execute(sql).fetchone()
+            if n and nd is not None:
+                ratio = (nd / n) if n else 0
+                if ratio > 0.95:
+                    return "id_like"
+                if 0.01 <= ratio <= 0.35:
+                    return "category"
+                if ratio < 0.01:
+                    return "binary_flag"
+        except Exception:
+            pass
+        return "text"
+    return "other"
+
+def quick_table_profile(con: duckdb.DuckDBPyConnection, table: str) -> Dict:
+    rows = con.execute(f'SELECT COUNT(*) FROM {table}').fetchone()[0]
+    cols = get_columns(con, table)
+    roles = {"category": [], "numeric": [], "date": [], "year": [], "id_like": [], "text": [], "binary": [], "other": []}
+    for c, d in cols:
+        r = role_of_column(con, table, c, d)
+        if r == "binary_flag":
+            roles["binary"].append(c)
+        else:
+            roles.setdefault(r, []).append(c)
+    return {
+        "rows": int(rows or 0),
+        "n_cols": len(cols),
+        "n_cat": len(roles["category"]),
+        "n_num": len(roles["numeric"]),
+        "n_time": len(roles["year"]) + len(roles["date"]),
+    }
+
+def table_mapping(con: duckdb.DuckDBPyConnection, user_tables: List[str]) -> Dict[str, Dict]:
+    """
+    Map db_table (normalized) -> {sheet_name, title} using __excel_tables if present.
+    """
+    normalize = lambda t: t.split('.', 1)[1].strip('"') if '.' in t else t.strip('"')
+    want_names = {normalize(t) for t in user_tables}
+    mapping: Dict[str, Dict] = {}
+    try:
+        rows = con.execute(
+            "SELECT sheet_name, table_name, inferred_title, original_title_text, block_index, start_row "
+            "FROM __excel_tables ORDER BY block_index, start_row"
+        ).fetchall()
+        for sheet_name, table_name, inferred_title, original_title_text, block_index, start_row in rows:
+            if table_name not in want_names:
+                continue
+            title = inferred_title or original_title_text or 'untitled'
+            mapping[table_name] = {'sheet_name': sheet_name, 'title': title}
+    except Exception:
+        pass
+    return mapping
+
+def excel_schema_samples(con: duckdb.DuckDBPyConnection, mapping: Dict[str, Dict], max_cols: int = 8) -> Dict[str, List[str]]:
+    """ Return up to max_cols original column names per table_name (normalized) for LLM hints. """
+    samples: Dict[str, List[str]] = {}
+    try:
+        rows = con.execute("SELECT sheet_name, table_name, column_ordinal, original_name FROM __excel_schema ORDER BY sheet_name, table_name, column_ordinal").fetchall()
+        for sheet_name, table_name, ordn, orig in rows:
+            if table_name not in mapping:
+                continue
+            lst = samples.setdefault(table_name, [])
+            if orig and len(lst) < max_cols:
+                lst.append(str(orig))
+    except Exception:
+        pass
+    return samples
+
+# ---------- OpenAI ----------
+def ai_overview_from_context(context: Dict) -> str:
+    api_key = os.environ.get("OPENAI_API_KEY") or st.secrets.get("OPENAI_API_KEY", None)
+    if not api_key:
+        raise RuntimeError("OPENAI_API_KEY is not set. Please add it to .env or Streamlit secrets.")
+
+    try:
+        from openai import OpenAI
+        client = OpenAI(api_key=api_key)
+        model = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
+    except Exception as e:
+        raise RuntimeError("OpenAI client not available. Install 'openai' >= 1.0 and try again.") from e
+
+    prompt = f'''
+Start directly (no greeting). Write a concise, conversational overview (max two short paragraphs) of the dataset created from the uploaded Excel.
+Requirements:
+- Do NOT mention database engines, schemas, or technical column/table names.
+- For each segment, reference it as: Sheet "<sheet_name>" — Table "<title>" (use "untitled" if missing).
+- Use any provided original Excel header hints ONLY to infer friendlier human concepts; do not quote them verbatim.
+- After the overview, list 6–8 simple questions a user could ask in natural language.
+- Output Markdown with headings: "Overview" and "Try These Questions".
+
+Context (JSON):
+{json.dumps(context, ensure_ascii=False, indent=2)}
+'''
+    resp = client.chat.completions.create(
+        model=model,
+        messages=[{"role": "user", "content": prompt}],
+        temperature=0.4,
+    )
+    return resp.choices[0].message.content.strip()
+
+# ---------- Orchestration ----------
+def run_ingestion_pipeline(xlsx_path: Path, db_path: Path, log_placeholder):
+    # Combined log function
+    log_lines: List[str] = []
+    def _append(line: str):
+        log_lines.append(line)
+        log_placeholder.markdown(
+            f"<div class='logbox'><code>{'</code><br/><code>'.join(map(str, log_lines[-400:]))}</code></div>",
+            unsafe_allow_html=True,
+        )
+
+    # 1) Save (already saved by caller, but we log here for a single place)
+    _append("[app] Saving file…")
+    _append("[app] Saved.")
+    if not SCRIPT_PATH.exists():
+        _append("[app] ERROR: ingestion component not found next to the app.")
+        raise FileNotFoundError("Required ingestion component not found.")
+
+    # 2) Ingest
+    _append("[app] Ingesting…")
+    env = os.environ.copy()
+    env["PYTHONIOENCODING"] = "utf-8"
+
+    cmd = [sys.executable, str(SCRIPT_PATH), "--excel", str(xlsx_path), "--duckdb", str(db_path)]
+    try:
+        proc = subprocess.Popen(
+            cmd, cwd=str(PRIMARY_DIR),
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+            text=True, bufsize=1, universal_newlines=True, env=env
+        )
+    except Exception as e:
+        _append(f"[app] ERROR: failed to start ingestion: {e}")
+        raise
+
+    for line in iter(proc.stdout.readline, ""):
+        _append(line.rstrip("\n"))
+    proc.wait()
+    if proc.returncode != 0:
+        _append("[app] ERROR: ingestion reported a non-zero exit code.")
+        raise RuntimeError("Ingestion failed. See logs.")
+
+    _append("[app] Ingestion complete.")
+
+    # 3) Open dataset
+    _append("[app] Opening dataset…")
+    con = duckdb.connect(str(db_path))
+    _append("[app] Dataset open.")
+
+    return con, _append
+
+def analyze_and_summarize(con: duckdb.DuckDBPyConnection):
+    user_tables = list_user_tables(con)
+    preview_items = []  # list of {'table_ref': t, 'label': label} for UI
+    if not user_tables:
+        # Try to provide metadata if no tables are found
+        try:
+            meta_df = con.execute("SELECT * FROM __excel_tables").fetchdf()
+            st.warning("No user tables were discovered. Showing ingestion metadata for reference.")
+            st.dataframe(meta_df, use_container_width=True, hide_index=True)
+        except Exception:
+            st.error("No user tables were discovered and no metadata table is available.")
+        return "", []
+
+    # Build mapping + schema hints
+    mapping = table_mapping(con, user_tables)  # normalized table_name -> {sheet_name, title}
+    schema_hints = excel_schema_samples(con, mapping, max_cols=8)
+
+    # Build compact profiling context for LLM; avoid raw db table names
+    per_table = []
+    for idx, t in enumerate(user_tables, start=1):
+        prof = quick_table_profile(con, t)
+        norm = t.split('.', 1)[1].strip('"') if '.' in t else t.strip('"')
+        m = mapping.get(norm, {})
+        sheet = m.get('sheet_name')
+        title = m.get('title')
+        per_table.append({
+            "idx": idx,
+            "sheet_name": sheet,
+            "title": title or "untitled",
+            "rows": prof["rows"],
+            "n_cols": prof["n_cols"],
+            "category_fields": prof["n_cat"],
+            "numeric_measures": prof["n_num"],
+            "time_fields": prof["n_time"],
+            "example_original_headers": schema_hints.get(norm, [])
+        })
+        label = f'Sheet "{sheet}" — Table "{title or "untitled"}"' if sheet else f'Table "{title or "untitled"}"'
+        preview_items.append({'table_ref': t, 'label': label})
+
+    context = {
+        "segments": per_table
+    }
+
+    # Generate overview (LLM only)
+    overview_md = ai_overview_from_context(context)
+    return overview_md, preview_items
+
+# ---------- UI flow ----------
+file = st.file_uploader("Upload an Excel file", type=["xlsx"])
+
+if file is None and not st.session_state.last_overview_md:
+    st.info("Upload an .xlsx file to begin.")
+
+# Only show logs AFTER there is an upload or some result to show
+logs_placeholder = None
+if file is not None or st.session_state.processing or st.session_state.last_overview_md:
+    logs_exp = st.expander("Processing logs", expanded=False)
+    logs_placeholder = logs_exp.empty()
+
+if file is not None:
+    key = _file_key(file)
+    stem = Path(file.name).stem
+    saved_xlsx = UPLOAD_DIR / f"{stem}.xlsx"
+    db_path = DB_DIR / f"{stem}.duckdb"
+
+    # --- CLEAR state immediately on new upload ---
+    if st.session_state.get("processed_key") != key:
+        st.session_state["last_overview_md"] = None
+        st.session_state["last_preview_items"] = []
+        st.session_state["chat_history"] = []
+        st.session_state["schema_text"] = None
+        st.session_state["db_path"] = None
+        # Optional: clear any previous logs shown in UI on rerun
+        # (no explicit log buffer stored; the log expander will refresh)
+
+
+    # Auto-start ingestion exactly once per unique upload
+    if (st.session_state.processed_key != key) and (not st.session_state.processing):
+        st.session_state.processing = True
+        if logs_placeholder is None:
+            logs_exp = st.expander("Ingestion logs", expanded=False)
+            logs_placeholder = logs_exp.empty()
+
+        # Save uploaded file
+        with open(saved_xlsx, "wb") as f:
+            f.write(file.getbuffer())
+
+        try:
+            con, app_log = run_ingestion_pipeline(saved_xlsx, db_path, logs_placeholder)
+            # Analyze + overview
+            app_log("[app] Analyzing data…")
+            overview_md, preview_items = analyze_and_summarize(con)
+            app_log("[app] Overview complete.")
+            con.close()
+
+            st.session_state.last_overview_md = overview_md
+            st.session_state.last_preview_items = preview_items
+            st.session_state.processed_key = key
+            st.session_state.processing = False
+
+        except Exception as e:
+            st.session_state.processing = False
+            st.error(f"Ingestion failed. See logs for details. Error: {e}")
+
+# Display results if available (and avoid re-triggering ingestion)
+if st.session_state.last_overview_md:
+    #st.subheader("Overview")
+    st.markdown(st.session_state.last_overview_md)
+
+    with st.expander("Quick preview (verification only)", expanded=False):
+        try:
+            # Reconnect to current dataset path (if present)
+            if file is not None:
+                stem = Path(file.name).stem
+                db_path = DB_DIR / f"{stem}.duckdb"
+            con = duckdb.connect(str(db_path))
+            for item in st.session_state.last_preview_items:
+                t = item['table_ref']
+                label = item['label']
+                df = con.execute(f"SELECT * FROM {t} LIMIT 50").df()
+                st.caption(f"Preview — {label}")
+                st.dataframe(df, use_container_width=True, hide_index=True)
+            con.close()
+        except Exception as e:
+            st.warning(f"Could not preview tables: {e}")
+
+
+# =====================
+# Chat with your dataset
+# (Appends after overview & preview; leaves earlier logic untouched)
+# =====================
+if st.session_state.get("last_overview_md"):
+    st.divider()
+    st.subheader("Chat with your dataset")
+
+    # Lazy imports so nothing changes before preview completes
+    def _lazy_imports():
+        from duckdb_react_agent import get_schema_summary, make_llm, answer_question  # noqa: F401
+        return get_schema_summary, make_llm, answer_question
+
+    # Initialize chat memory
+    st.session_state.setdefault("chat_history", [])  # [{role, content, sql?, plot_path?}]
+
+    # --- 1) Take input first so the user's question appears immediately ---
+    user_q = st.chat_input("Ask a question about the dataset…")
+    if user_q:
+        st.session_state.chat_history.append({"role": "user", "content": user_q})
+
+    # --- 2) Render history in strict User → Assistant order ---
+    for msg in st.session_state.chat_history:
+        with st.chat_message("user" if msg["role"] == "user" else "assistant"):
+            # Always: text first
+            st.markdown(msg["content"])
+
+            # Then: optional plot (below the answer)
+            plot_path_hist = msg.get("plot_path")
+            if plot_path_hist:
+                if not os.path.isabs(plot_path_hist):
+                    plot_path_hist = str((PRIMARY_DIR / plot_path_hist).resolve())
+                if os.path.exists(plot_path_hist):
+                    st.image(plot_path_hist, caption="Chart", width=520)
+
+            # Finally: optional SQL expander
+            if msg.get("sql"):
+                with st.expander("View generated SQL", expanded=False):
+                    st.markdown(f"<div class='sqlbox'>{msg['sql']}</div>", unsafe_allow_html=True)
+
+    # --- 3) If a new question arrived, stream the assistant answer now ---
+    if user_q:
+        with st.chat_message("assistant"):
+            # Placeholders in sequence: text → plot → SQL
+            stream_placeholder = st.empty()
+            plot_placeholder = st.empty()
+            sql_placeholder = st.empty()
+
+            # Show pending immediately
+            stream_placeholder.markdown("_Answer pending…_")
+
+            partial_chunks = []
+            def on_token(t: str):
+                partial_chunks.append(t)
+                stream_placeholder.markdown("".join(partial_chunks))
+
+            # Resolve DB path
+            if 'db_path' in locals():
+                _db_path = db_path  # from the preview scope if defined
+            else:
+                if 'file' in locals() and file is not None:
+                    _stem = Path(file.name).stem
+                    _db_path = DB_DIR / f"{_stem}.duckdb"
+                else:
+                    _candidates = sorted(DB_DIR.glob("*.duckdb"), key=lambda p: p.stat().st_mtime, reverse=True)
+                    _db_path = _candidates[0] if _candidates else None
+
+            if not _db_path or not Path(_db_path).exists():
+                stream_placeholder.error("No dataset found. Please re-upload the Excel file in this session.")
+            else:
+                # Call agent lazily
+                get_schema_summary, make_llm, answer_question = _lazy_imports()
+                try:
+                    try:
+                        con2 = duckdb.connect(str(_db_path), read_only=True)
+                    except Exception:
+                        con2 = duckdb.connect(str(_db_path))
+
+                    schema_text = get_schema_summary(con2, allowed_schemas=["main"])
+                    llm = make_llm(model=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"), temperature=0.0)
+
+                    import inspect as _inspect
+                    _sig = None
+                    try:
+                        _sig = _inspect.signature(answer_question)
+                    except Exception:
+                        _sig = None
+
+                    def _call_answer():
+                        try:
+                            if _sig and "history" in _sig.parameters and "token_callback" in _sig.parameters and "stream" in _sig.parameters:
+                                return answer_question(con2, llm, schema_text, user_q, stream=True, token_callback=on_token, history=st.session_state.chat_history)
+                            elif _sig and "token_callback" in _sig.parameters and "stream" in _sig.parameters:
+                                return answer_question(con2, llm, schema_text, user_q, stream=True, token_callback=on_token)
+                            else:
+                                return answer_question(con2, llm, schema_text, user_q)
+                        except TypeError:
+                            try:
+                                return answer_question(con2, llm, schema_text, user_q, stream=True, token_callback=on_token)
+                            except TypeError:
+                                return answer_question(con2, llm, schema_text, user_q)
+
+                    result = _call_answer()
+                    con2.close()
+
+                    # Finalize text
+                    answer_text = result.get("answer") or "".join(partial_chunks) or "*No answer produced.*"
+                    stream_placeholder.markdown(answer_text)
+
+                    # Show plot next (slightly larger, below the text)
+                    plot_path = result.get("plot_path")
+                    if plot_path:
+                        if not os.path.isabs(plot_path):
+                            plot_path = str((PRIMARY_DIR / plot_path).resolve())
+                        if os.path.exists(plot_path):
+                            plot_placeholder.image(plot_path, caption="Chart", width=560)
+
+                    # Finally show SQL
+                    gen_sql = (result.get("sql") or "").strip()
+                    if gen_sql:
+                        with st.expander("View generated SQL", expanded=False):
+                            st.markdown(f"<div class='sqlbox'>{gen_sql}</div>", unsafe_allow_html=True)
+
+                    # Persist assistant message
+                    st.session_state.chat_history.append({
+                        "role": "assistant",
+                        "content": answer_text,
+                        "sql": gen_sql,
+                        "plot_path": result.get("plot_path")
+                    })
+                finally:
+                    try:
+                        con2.close()
+                    except Exception:
+                        pass