DeepSeek-V3.2-Speciale Tutorial: Bouw een Data Analyst Agent

DeepSeek V3.2 Speciale is een high-compute redeneermodel dat is afgestemd op complexe, door tools ondersteunde workflows. In plaats van vragen direct te beantwoorden, is het ontworpen om te plannen, te redeneren en tools te coördineren voor het bouwen van agent-gedreven analytics-systemen.

In deze tutorial gebruiken we DeepSeek V3.2 Speciale om een AI Data Analyst Agent te bouwen boven op elke CSV. De flow ziet er zo uit:

1. Gebruiker uploadt een CSV-bestand en stelt een vraag in natuurlijke taal 
2. Een planner-agent zet de vraag om in een gestructureerd JSON-analyseplan
3. Vervolgens voeren enkele Python-tools het plan uit en genereren tabellen/grafieken.
4. Tot slot zet een uitleg-agent de ruwe cijfers om in een samenvatting met inzichten.

Aan het eind heb je een Streamlit-app met expliciete plannen en reproduceerbare analysestappen. Als je meer wilt leren over AI-agents, raad ik de AI Agent Fundamentals skill track aan.  

Wat is DeepSeek V3.2 Speciale?

DeepSeek V3.2 Speciale is de high-compute-variant uit de DeepSeek V3.2-familie met reinforcement learning-boost, speciaal ontworpen voor diepgaand redeneren in plaats van casual chat. 

Het drijft de prestaties op tot het niveau waarop het kan wedijveren met modellen zoals Gemini 3 Pro bij complexe taken, en heeft zelfs goudenmedaillescores behaald op benchmarks zoals IMO, CMO, ICPC World Finals en IOI 2025.

DeepSeek V3.2 Speciale is geoptimaliseerd voor:

• Moeilijke wiskunde, logica en meerstapsredeneren
• Het ondersteunt ook agent-gedreven workflows waarbij het model acties plant in plaats van in één keer te antwoorden
• Net als zijn rivalen ondersteunt het ook long context en gestructureerde outputs met chain-of-thought-achtig redeneren

V3.2 Speciale ondersteunt alleen thinking mode en ondersteunt momenteel geen tool-calling via de officiële API. Je kunt niet simpelweg de temperature aanpassen en verwachten dat je bij de eerste poging schone JSON krijgt. Het Speciale-model zal uitgebreid nadenken en zijn redenering vaak vermengen met het eindantwoord.

DeepSeek-V3.2-Speciale Voorbeeldproject: AI Data Analyst Agent

In deze sectie implementeren we een AI Data Analyst Agent met een DeepSeek V3.2 Speciale-model, verpakt in een Streamlit-applicatie. In grote lijnen doet de uiteindelijke app het volgende:

• Upload elke CSV. Je vindt goede voorbeelden op Kaggle (zoals sales forecasting, dat ik in deze demo gebruikte)
• De app detecteert automatisch datumkolommen en voegt hulpkolommen toe voor jaar, maand en year_month.
• Je typt een vraag, en DeepSeek V3.2 Speciale fungeert als Planner Agent die reageert met een JSON-plan:
• welke kolommen je groepeert
• welke metrieken je aggregeert
• optionele filters (bijv. “Year == 2019”)
• of er een staaf-, lijn- of cirkeldiagram moet worden getekend
• Een pure Python Data Worker voert het plan uit met pandas.
• Een Chart Generator bouwt een staaf-/lijn-/cirkeldiagram met matplotlib en seaborn.
• Een Explainer Agent vat het resultaat samen in eenvoudige taal en benadrukt de belangrijkste inzichten.

Laten we het stap voor stap bouwen.

Stap 1: Vereisten

Installeer eerst de kern-vereisten en haal een API-sleutel op via DeepSeek’s API-platform:

pip install streamlit pandas matplotlib seaborn python-dotenv openaiMogelijk gemaakt door

We gebruiken een paar kernbibliotheken zoals streamlit voor de interactieve webapp-UI, pandas om CSV-gegevens te laden en te manipuleren, matplotlib en seaborn om grafieken en visualisaties te maken. De bibliotheek python-dotenv wordt gebruikt om omgevingsvariabelen uit een .env-bestand te laden, en OpenAI als client om met DeepSeek’s OpenAI-compatibele API te praten.

Log vervolgens in op het DeepSeek API-platform, ga naar het tabblad API Keys en klik op Create a new API key.

Geef je sleutel ten slotte een naam en kopieer de gegenereerde sleutel. 

Stel je DeepSeek API-sleutel in als omgevingsvariabele:

export DEEPSEEK_API_KEY="your_api_key_here"Mogelijk gemaakt door

Onze omgeving is nu klaar.

Stap 2: Imports en DeepSeek-client

Met de dependencies geïnstalleerd, is de volgende stap om ze in ons script te importeren en de DeepSeek-client te koppelen die onze planner- en uitleg-agents aandrijft.

import io
import json
import os
import re
from datetime import datetime
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns
import streamlit as st
from dotenv import load_dotenv
from openai import OpenAI
plt.style.use('seaborn-v0_8-darkgrid')
sns.set_palette("husl")
load_dotenv()
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
if not DEEPSEEK_API_KEY:
    raise RuntimeError("Please set DEEPSEEK_API_KEY env variable.")
DEEPSEEK_BASE_URL = os.getenv(
    "DEEPSEEK_BASE_URL",
    "https://api.deepseek.com/v3.2_speciale_expires_on_20251215",
)
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)Mogelijk gemaakt door

Het bovenstaande codeblok doet drie dingen. Het importeert onze kernbibliotheken, past een plotstijl toe en laadt omgevingsvariabelen uit .env. Vervolgens leest het DEEPSEEK_API_KEY, valideert die en bouwt een OpenAI-client die verwijst naar DeepSeek’s tijdelijke V3.2 Speciale-endpoint, zodat elke volgende LLM-call via het juiste model loopt. 

Let op: DeepSeek V3.2 Speciale wordt momenteel geleverd vanaf een tijdelijk endpoint dat verloopt op 15 december 2025. Daarna kan de deployment-URL veranderen, maar de modelnaam blijft deepseek-reasoner. Raadpleeg voor de meest recente endpoints en gebruiksdetails de officiële DeepSeek-documentatie.

Met de client geconfigureerd definiëren we onze twee agents: één om de analyse te plannen en een andere om de resultaten uit te leggen.

Stap 3: Planner Agent 

Met de client op zijn plek kunnen we de Planner Agent bouwen, die beslist welke analyse op de data wordt uitgevoerd. In plaats van het model rechtstreeks de DataFrame te laten bewerken, vragen we het om een gestructureerd JSON-plan te produceren dat onze Python “data worker” veilig kan uitvoeren. 

De planner ziet een tekstuele beschrijving van de dataset samen met de vraag van de gebruiker en retourneert een JSON-object dat de operatie, group-by-kolommen, filters, metriek en grafiektype beschrijft.

def call_planner_llm(schema_text: str, question: str) -> tuple[dict, str]:
    system_prompt = """
    You are a data analysis planner. Convert user questions into JSON plans.
    JSON Structure (REQUIRED):
    {
      "operation": "group_by_summary",
      "group_by": ["column_name"],
      "filters": [],
      "target_column": "column_to_analyze",
      "metric": "sum",
      "need_chart": true,
      "chart_type": "bar"
    }
    Fields:
    - operation: Always use "group_by_summary" (most flexible)
    - group_by: Columns to group by (e.g., ["Product Name"] or ["Category", "Region"])
    - filters: Optional filters, e.g., [{"column": "Year", "op": "==", "value": "2023"}]
    - target_column: Column to aggregate (must exist in schema!)
    - metric: One of: sum, mean, count, max, min
    - need_chart: true (always show chart)
    - chart_type: bar, line, or pie
    Quick Rules:
    - Use ONLY columns from the provided schema
    - For "this year", look for year columns or use the latest year in data
    - For "top N", use appropriate group_by and metric
    - For comparisons, use filters to split groups 
    Think briefly, then output the complete JSON.
    """.strip()
    messages = [
        {"role": "system", "content": system_prompt},
        {
            "role": "user",
            "content": f"Dataset:\n{schema_text}\n\nQuestion:\n{question}\n\nReminder: After analyzing, output the complete JSON plan.",
        },
    ]
    resp = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=messages,
        max_tokens=8192,
    )
    message = resp.choices[0].message
    content = message.content or ""
    content = content.strip() 
    reasoning_content = getattr(message, 'reasoning_content', None) or ""
    finish_reason = resp.choices[0].finish_reason
    def extract_json_from_text(text: str) -> dict | None:
        if not text:
            return None
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            pass   
        code_block_match = re.search(r'```(?:json)?\s*(\{[\s\S]*?\})\s*```', text, re.DOTALL)
        if code_block_match:
            try:
                return json.loads(code_block_match.group(1))
            except json.JSONDecodeError:
                pass
        def find_all_json_objects(s):
            objects = []
            depth = 0
            start_idx = None
            in_string = False
            escape_next = False            
            for i, char in enumerate(s):
                if escape_next:
                    escape_next = False
                    continue                
                if char == '\\':
                    escape_next = True
                    continue                
                if char == '"' and not in_string:
                    in_string = True
                elif char == '"' and in_string:
                    in_string = False
                elif not in_string:
                    if char == '{':
                        if depth == 0:
                            start_idx = i
                        depth += 1
                    elif char == '}':
                        depth -= 1
                        if depth == 0 and start_idx is not None:
                            objects.append(s[start_idx:i+1])
                            start_idx = None            
            return objects 
        for candidate in find_all_json_objects(text):
            try:
                parsed = json.loads(candidate)
                if isinstance(parsed, dict) and any(k in parsed for k in ['operation', 'target_column', 'metric', 'group_by']):
                    return parsed
            except json.JSONDecodeError:
                continue
        for i in range(len(text)):
            if text[i] == '{':
                decoder = json.JSONDecoder()
                try:
                    parsed, _ = decoder.raw_decode(text[i:])
                    if isinstance(parsed, dict) and any(k in parsed for k in ['operation', 'target_column', 'metric', 'group_by']):
                        return parsed
                except json.JSONDecodeError:
                    continue        
        return None
    if reasoning_content:
        plan = extract_json_from_text(reasoning_content)
        if plan is not None:
            return plan, reasoning_content
    plan = extract_json_from_text(content)
    if plan is not None:
        return plan, reasoning_content
    error_msg = (
        f"Planner returned non-JSON content.\n"
        f"Finish reason: {finish_reason}\n"
        f"Content received: {repr(content[:500]) if content else '(empty)'}\n"
        f"Reasoning content length: {len(reasoning_content) if reasoning_content else 0}\n"
        f"Reasoning content sample: {reasoning_content[:500] if reasoning_content else '(none)'}"
    )
    if finish_reason == "length":
        error_msg += "\n\nNote: Response was truncated due to token limit. The model didn't finish generating the JSON."   
    raise ValueError(error_msg)Mogelijk gemaakt door

Eerst definieert de system_prompt het contract voor de Planner Agent, die de gebruikersvraag omzet in een JSON-plan, en specificeert het precies welke velden we verwachten, zoals een lijst met group_by-kolommen, optionele filters, een target_column, een metric en een chart_type. 

We voegen ook enkele heuristieken toe (zoals hoe om te gaan met “dit jaar” of “top N”), zodat het model zich meer gedraagt als een junior data-analist dan als een generieke chatbot.

Vervolgens bouwen we het systeembericht dat de instructies draagt en twee zaken bundelt: een tekstuele beschrijving van het dataschema (kolomnamen, dtypes, voorbeeldrijen) en de vraag van de gebruiker. Dit geeft het model genoeg context om geldige kolomnamen te kiezen en een zinnige aggregatie te bepalen voor de gestelde vraag.

DeepSeek V3.2 Speciale wordt daarna aangeroepen via client.chat.completions.create(). Omdat dit een redeneermodel is, kan het twee kanalen aan output produceren: een zichtbare content-string en een interne reasoning_content-string waarin het meeste “denken” en vaak ook het eindantwoord staan. We halen beide uit de response en onthouden de finish_reason zodat we afkapping kunnen detecteren.

Het zware werk gebeurt in de functie extract_json_from_text(). Omdat redeneermodellen niet altijd schone JSON retourneren, verpakken we het in markdown. Om de planner robuust te maken, probeert deze helperfunctie meerdere strategieën na elkaar:

• Hij probeert eerst de hele string te parsen met json.loads
• Dan zoekt hij naar JSON-codeblokken en parseert de inhoud daarin
• Tot slot scant hij op gebalanceerde blokken met een kleine accolade-matcher en gebruikt een ruwe JSON-decoder.

Bij elke kandidaat accepteert hij alleen objecten die eruitzien als een geldig plan, d.w.z. ze moeten ten minste een van de velden operation, target_column, metric of group_by bevatten. 

Tot slot verbinden we alles door de functie extract_json_from_text() toe te passen op reasoning_content.

Samen zetten deze onderdelen elke vraag in natuurlijke taal plus een schema-snapshot om in een gestructureerd analyseplan. 

Stap 4: Explainer Agent

Zodra de Data Worker de analyse met Pandas heeft uitgevoerd, sturen we de resultatentabel samen met het JSON-plan terug naar DeepSeek en maken we er een samenvatting met concrete inzichten van.

def call_explainer_llm(question: str, plan: dict, result_summary: list) -> tuple[str, str]:
    system_prompt = """
    You are a senior data analyst explaining insights to business stakeholders.
    The user asked a question about their CSV data.
    Another agent already designed an analysis plan and we executed it in Python.
    You will now explain the results in clear, engaging, non-technical language.
    Requirements:
    - Start with a direct 1-2 sentence answer to their question
    - Use bullet points (3-5 points) to highlight key insights, trends, or comparisons
    - Include specific numbers and percentages where relevant
    - Use business-friendly language (avoid technical jargon like "aggregation", "groupby")
    - If the data shows interesting patterns, point them out
    - End with a brief recommendation or next step if appropriate    
    Keep it concise and actionable.
    """.strip()
    limited_summary = result_summary[:20] if len(result_summary) > 20 else result_summary
    user_content = f"""
    User question:
    {question}
    Analysis performed:
    {json.dumps(plan, indent=2)}
    Results (top rows):
    {json.dumps(limited_summary, indent=2)}    
    Total rows in result: {len(result_summary)}
    """.strip()
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_content},
    ]
    resp = client.chat.completions.create(
        model="deepseek-reasoner",
        messages=messages,
        max_tokens=2048,
    )   
    message = resp.choices[0].message
    content = message.content or ""
    content = content.strip()
    reasoning_content = getattr(message, 'reasoning_content', None) or ""    
    if not content:
        if reasoning_content:
            content = reasoning_content.strip()  
    return (content if content else "Unable to generate explanation from the model response.", 
            reasoning_content)Mogelijk gemaakt door

De uitleg-agent begint met het instellen van de system_prompt, die DeepSeek instrueert om zich te gedragen als een senior data-analist die met business stakeholders praat. We specificeren de exacte vorm van het gewenste antwoord, plus andere aanwijzingen voor het opstellen van de reactie.

Vervolgens bereiden we user_content voor, waarmee we het model alle benodigde context geven. We geven de oorspronkelijke vraag mee, het JSON-analyseplan (zodat het weet welke berekeningen zijn uitgevoerd) en het totale aantal rijen.

Daarna stellen we de berichtenlijst samen en roepen we DeepSeek V3.2 Speciale aan via client.chat.completions.create(). Net als bij de planner kan het model tekst teruggeven in zowel content als reasoning_content, dus we voegen beide toe en vallen terug op reasoning_content als content leeg is. 

Tot slot retourneert de functie een tuple met zowel de uitleg als de ruwe redeneer-string die in de expander verschijnt.

Met deze Explainer Agent is de loop rond:

• De Planner Agent beslist wat er berekend wordt,
• De Data Worker voert het plan uit met pandas en matplotlib, en
• De Explainer Agent vertaalt de cijfers naar actiegerichte inzichten

Stap 5: Helperfuncties

Voordat we DeepSeek iets kunnen laten plannen, moeten we de data duidelijk beschrijven en analysevriendelijk maken. Dat bereiken we met een reeks helperfuncties.

Stap 5.1 Schema-planner

De Planner Agent kan je DataFrame niet direct zien, dus sturen we een compacte tekstuele samenvatting van de dataset.

def get_schema_description(df: pd.DataFrame, max_rows: int = 3) -> str:
    schema_lines = ["Columns:"]
    for col, dtype in df.dtypes.items():
        if dtype == 'object' and df[col].nunique() < 20:
            unique_vals = df[col].dropna().unique()[:5]
            schema_lines.append(f"- {col} ({dtype}) - sample values: {', '.join(map(str, unique_vals))}")
        else:
            schema_lines.append(f"- {col} ({dtype})")
    sample = df.head(max_rows).to_markdown(index=False)
    return "\n".join(schema_lines) + "\n\nSample rows:\n" + sampleMogelijk gemaakt door

Deze helper loopt alle kolommen langs. Hij geeft ook enkele voorbeeldwaarden voor zaken als productnamen, regio’s of categorieën. F

inaal voegt hij een korte markdown-tabel toe met de eerste paar rijen, zodat het model concrete voorbeelden heeft om over te redeneren.

Deze schematekst is het oog van de planner op je data, waardoor die geldige kolomnamen en group-bys kan kiezen.

Stap 5.2 Automatische datum-preprocessing

We detecteren automatisch datumkolommen en voegen hulpkolommen toe zoals <col>_year en <col>_year_month, wat het voor DeepSeek makkelijker maakt om vragen als “vorig jaar” of “per maand” te beantwoorden.

def preprocess_dates(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    date_columns = []    
    for col in df.columns:
        if pd.api.types.is_datetime64_any_dtype(df[col]):
            date_columns.append(col)
            continue
        if any(x in col.lower() for x in ['date', 'time', 'day']):
            try:
                df[col] = pd.to_datetime(df[col], errors='coerce')
                if df[col].notna().sum() > 0: 
                    date_columns.append(col)
            except:
                pass
    for col in date_columns:
        if col in df.columns and pd.api.types.is_datetime64_any_dtype(df[col]):
            df[f'{col}_year'] = df[col].dt.year
            df[f'{col}_month'] = df[col].dt.month
            df[f'{col}_year_month'] = df[col].dt.to_period('M').astype(str) 
    return dfMogelijk gemaakt door

Eerst scannen we de kolommen, en als een kolom al datetime is of de naam op een datum wijst (zoals order_date of day), proberen we die te parsen met pd.to_datetime. Voor elke datumkolom maken we features zoals <col>_year, <col>_month en <col>_year_month.

Deze preprocessing betekent dat de planner veilig kan verwijzen naar year- en year_month-kolommen zonder dat jij voor elke dataset tijdsfeatures handmatig hoeft te engineeren.

Stap 5.3 Filtertoepassing

De planner kan eenvoudige rijf filters aan het plan toevoegen, en deze helper zet die om in pandas-filters.

def apply_filters(df: pd.DataFrame, filters: list) -> pd.DataFrame:
    if not filters:
        return df
    filtered = df.copy()
    for f in filters:
        col = f.get("column")
        op = f.get("op")
        val = f.get("value")
        if col not in filtered.columns:
            continue
        try:
            val_cast = pd.to_numeric(val)
        except Exception:
            val_cast = val
        if op == "==":
            filtered = filtered[filtered[col] == val_cast]
        elif op == "!=":
            filtered = filtered[filtered[col] != val_cast]
        elif op == ">":
            filtered = filtered[filtered[col] > val_cast]
        elif op == "<":
            filtered = filtered[filtered[col] < val_cast]
        elif op == ">=":
            filtered = filtered[filtered[col] >= val_cast]
        elif op == "<=":
            filtered = filtered[filtered[col] <= val_cast]
        elif op == "contains":
            filtered = filtered[filtered[col].astype(str).str.contains(str(val_cast), case=False, na=False)]
    return filteredMogelijk gemaakt door

Voor elk filter controleren we of de kolom bestaat, en we casten de waarde indien mogelijk naar een getal, waarna we de juiste vergelijking toepassen. Alle filters worden sequentieel toegepast, zodat je voorwaarden als “Region contains West” en “Year >= 2021” natuurlijk kunt combineren.

Deze functie verbindt de natuurlijke-taalbeperkingen uit het plan met rijsubsets.

Stap 5.4: Analyseplan 

Nu implementeren we de Data Worker, een dunne laag die het JSON-plan leest en de overeenkomstige pandas-aggregaties uitvoert.

def run_analysis_plan(df: pd.DataFrame, plan: dict) -> pd.DataFrame:
    op = plan.get("operation")
    target_col = plan.get("target_column")
    group_by = plan.get("group_by") or []
    filters = plan.get("filters") or []
    metric = plan.get("metric", "sum")
    if filters:
        df = apply_filters(df, filters)
    if not op or not target_col or target_col not in df.columns:
        return df.head(20)
    if op in ("aggregate_compare", "group_by_summary", "filter_then_aggregate"):
        if not group_by:
            if metric == "sum":
                value = df[target_col].sum()
            elif metric == "mean":
                value = df[target_col].mean()
            elif metric == "count":
                value = df[target_col].count()
            elif metric == "max":
                value = df[target_col].max()
            elif metric == "min":
                value = df[target_col].min()
            else:
                raise ValueError(f"Unsupported metric: {metric}")
            return pd.DataFrame({f"{metric}_{target_col}": [value]})
        if metric == "sum":
            agg_df = df.groupby(group_by)[target_col].sum().reset_index()
        elif metric == "mean":
            agg_df = df.groupby(group_by)[target_col].mean().reset_index()
        elif metric == "count":
            agg_df = df.groupby(group_by)[target_col].count().reset_index()
        elif metric == "max":
            agg_df = df.groupby(group_by)[target_col].max().reset_index()
        elif metric == "min":
            agg_df = df.groupby(group_by)[target_col].min().reset_index()
        else:
            raise ValueError(f"Unsupported metric: {metric}")
        agg_df = agg_df.sort_values(by=target_col, ascending=False)       
        return agg_df
    return df.head(20)Mogelijk gemaakt door

We beginnen met het uitpakken van het plan met parameters voor operatie, doelkolom, groepeer-kolommen, filters en metriek. T

oen passen we filters toe via de functie apply_filters(). Als het plan naar een ongeldige kolom verwijst, vallen we terug op een eenvoudige preview. 

Anders voeren we voor ondersteunde operaties ofwel een aggregatie over de hele tabel uit, of een groupby gevolgd door bewerkingen zoals sum, mean, count, max of min, en sorteren we het resultaat op de metriek-kolom.

Dit is de kern van de toolslaag: de LLM beslist wat er berekend moet worden, en deze functie beslist hoe je het veilig met pandas uitvoert.

Stap 5.5: Grafiekgenerator 

Tot slot zetten we de resulterende tabel om in een grafiek die overeenkomt met het chart_type uit het plan, zodat we zowel cijfers als visuals in de output hebben.

def generate_chart(df: pd.DataFrame, plan: dict):
    if not plan.get("need_chart") or df.empty:
        return None
    chart_type = plan.get("chart_type", "bar")
    group_by = plan.get("group_by") or []
    target_col = plan.get("target_column")
    if not target_col or target_col not in df.columns:
        return None
    fig, ax = plt.subplots(figsize=(10, 6))    
    if group_by and len(group_by) > 0:
        x_col = group_by[0]
        if x_col not in df.columns:
            return None
        plot_df = df.head(15)        
        if chart_type == "bar":
            bars = ax.bar(range(len(plot_df)), plot_df[target_col], color=sns.color_palette("husl", len(plot_df)))
            ax.set_xticks(range(len(plot_df)))
            ax.set_xticklabels(plot_df[x_col], rotation=45, ha='right')
            ax.set_ylabel(target_col, fontsize=12)
            ax.set_xlabel(x_col, fontsize=12)
            for i, (bar, val) in enumerate(zip(bars, plot_df[target_col])):
                height = bar.get_height()
                ax.text(bar.get_x() + bar.get_width()/2., height,
                       f'{val:,.0f}' if abs(val) > 100 else f'{val:.2f}',
                       ha='center', va='bottom', fontsize=9)                 
        elif chart_type == "line":
            ax.plot(range(len(plot_df)), plot_df[target_col], marker='o', linewidth=2, markersize=8)
            ax.set_xticks(range(len(plot_df)))
            ax.set_xticklabels(plot_df[x_col], rotation=45, ha='right')
            ax.set_ylabel(target_col, fontsize=12)
            ax.set_xlabel(x_col, fontsize=12)
            ax.grid(True, alpha=0.3)    
        elif chart_type == "pie" and len(plot_df) <= 10:
            ax.pie(plot_df[target_col], labels=plot_df[x_col], autopct='%1.1f%%', startangle=90)
            ax.axis('equal')
        else:
            bars = ax.bar(range(len(plot_df)), plot_df[target_col])
            ax.set_xticks(range(len(plot_df)))
            ax.set_xticklabels(plot_df[x_col], rotation=45, ha='right')
        title = f"{target_col} by {x_col}"
    else:
        ax.bar([target_col], [df[target_col].iloc[0]])
        title = f"{target_col}"
    ax.set_title(title, fontsize=14, fontweight='bold', pad=20)
    fig.tight_layout()
    buf = io.BytesIO()
    fig.savefig(buf, format="png", dpi=150, bbox_inches='tight')
    buf.seek(0)
    plt.close(fig)
    return bufMogelijk gemaakt door

We lezen chart_type, group_by en target_column uit het plan en bouwen een Matplotlib-figuur. De bovenstaande functie generate_chart() retourneert een in-memory PNG-buffer die de Streamlit-app naast kernmetrieken toont.

Met deze helperfuncties wordt de rest van het systeem veel eenvoudiger. De Planner gebruikt get_schema_description om de analyse te ontwerpen, de Data Worker gebruikt de functies preprocess_dates(), apply_filters() en run_analysis_plan() om te rekenen, en de functie generate_chart() zet de uiteindelijke tabel om in een grafiek.

Stap 6: Streamlit-UI

Nu we de planner, data worker, uitlegger en helperfuncties hebben, kunnen we alles in een Streamlit-app gieten. 

st.set_page_config(
    page_title="AI Data Analyst Agent",
    page_icon="📊",
    layout="wide",
    initial_sidebar_state="expanded"
)
st.markdown("""
<style>
    .main-header {
        
        font-weight: 700;
        
        
        text-align: center;
    }
    .sub-header {
        
        
        
        text-align: center;
    }
    .metric-card {
        background-
        
        border-radius: 0.5rem;
        border-left: 4px solid #1f77b4;
    }
    .insight-box {
        background-
        
        border-radius: 0.5rem;
        border-left: 4px solid #2ecc71;
        
    }
</style>
""", unsafe_allow_html=True)
st.markdown("<h1 class='main-header'>AI Data Analyst Agent</h1>", unsafe_allow_html=True)
uploaded_file = st.file_uploader("Upload CSV file", type=["csv"], help="Upload any CSV file to get started")
if uploaded_file is not None:
    try:
        df = pd.read_csv(uploaded_file)
        df = preprocess_dates(df)
    except Exception as e:
        st.error(f"Error loading CSV: {e}")
        st.stop()
    with st.expander("Preview Data", expanded=False):
        col1, col2, col3 = st.columns(3)
        col1.metric("Total Rows", f"{len(df):,}")
        col2.metric("Total Columns", len(df.columns))
        col3.metric("Memory Usage", f"{df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")  
        st.dataframe(df.head(100), width='stretch', height=300)
        st.markdown("**Numeric Columns Summary:**")
        numeric_cols = df.select_dtypes(include=['number']).columns
        if len(numeric_cols) > 0:
            st.dataframe(df[numeric_cols].describe(), width='stretch')
    st.markdown("---")
    col1, col2 = st.columns([4, 1])
    with col1:
        question = st.text_input(
            " Ask a question about your data",
            placeholder="e.g., Which product category had the highest sales last year?",
            label_visibility="collapsed"
        )
    with col2:
        analyze_button = st.button("Analyze", type="primary", width='stretch')
    if analyze_button and question.strip():
        with st.spinner("Analyzing data..."):
            schema_text = get_schema_description(df)
            try:
                plan, planner_reasoning = call_planner_llm(schema_text, question)
            except Exception as e:
                st.error(f" Planner agent failed: {e}")
                st.stop()
            with st.expander(" Analysis Plan", expanded=False):
                col1, col2 = st.columns(2)
                with col1:
                    st.json(plan)
                with col2:
                    if planner_reasoning:
                        st.markdown("**AI Reasoning Process:**")
                        st.text_area("Planner Reasoning", planner_reasoning[:1000] + "..." if len(planner_reasoning) > 1000 else planner_reasoning, 
                                   height=200, label_visibility="collapsed")
            try:
                result_df = run_analysis_plan(df, plan)
            except Exception as e:
                st.error(f" Error running analysis: {e}")
                st.info(" Tip: Try rephrasing your question or be more specific about the columns you want to analyze.")
                st.stop()
        st.markdown("---")
        st.markdown("## Results")
        chart_buf = generate_chart(result_df, plan) 
        if chart_buf is not None:
            col1, col2 = st.columns([2, 1])
            with col1:
                st.image(chart_buf)
            with col2:
                st.markdown("### Key Metrics")
                target_col = plan.get("target_column")
                if target_col in result_df.columns:
                    st.metric("Total", f"{result_df[target_col].sum():,.2f}")
                    st.metric("Average", f"{result_df[target_col].mean():,.2f}")
                    st.metric("Max", f"{result_df[target_col].max():,.2f}")
                    st.metric("Min", f"{result_df[target_col].min():,.2f}")
        else:
            st.info("Chart generation skipped (not applicable for this query)")
        with st.expander("Results Table", expanded=True):
            st.dataframe(result_df, width='stretch', height=400)
            csv_data = result_df.to_csv(index=False)
            st.download_button(
                label="Download Results as CSV",
                data=csv_data,
                file_name=f"analysis_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv",
                mime="text/csv"
            )
        with st.spinner("Generating insights..."):
            result_df_serializable = result_df.copy()
            for col in result_df_serializable.columns:
                if pd.api.types.is_datetime64_any_dtype(result_df_serializable[col]):
                    result_df_serializable[col] = result_df_serializable[col].astype(str)
                elif pd.api.types.is_period_dtype(result_df_serializable[col]):
                    result_df_serializable[col] = result_df_serializable[col].astype(str)
            result_summary = result_df_serializable.to_dict(orient="records")
            explanation, explainer_reasoning = call_explainer_llm(
                question=question,
                plan=plan,
                result_summary=result_summary,
            )
        st.markdown("---")
        st.markdown("## AI Insights")
        st.markdown(f'<div class="insight-box">{explanation}</div>', unsafe_allow_html=True)
        if explainer_reasoning:
            with st.expander(" AI Reasoning Process", expanded=False):
                st.text_area("Explainer Reasoning", explainer_reasoning[:1500] + "..." if len(explainer_reasoning) > 1500 else explainer_reasoning,
                           height=300, label_visibility="collapsed")Mogelijk gemaakt door

Bovenaan configureren we de pagina met st.set_page_config en CSS om het dashboard te stylen. De gebruiker uploadt eerst een CSV, die we inladen in een DataFrame, door preprocess_dates halen en tonen in een Preview Data-expander met rijen-/kolomtellingen, geheugengebruik en een snelle samenvatting.

Achter de schermen bouwen we een schema met de functie get_schema_description(), roepen we call_planner_llm aan om een JSON-plan te krijgen en tonen we zowel het plan als de redenering van het model in een expander. De Data Worker voert vervolgens het plan uit met pandas, en de sectie Results toont een grafiek van generate_chart samen met de kernmetrieken.

Tot slot roepen we call_explainer_llm aan met de vraag, het plan en de resultaatsamenvatting om een uitleg te genereren voor de AI Insights-box, met een optionele AI Reasoning Process-expander voor wie onder de motorkap wil kijken.

Sla alles op als app.py en voer uit:

streamlit run app.pyMogelijk gemaakt door

In de video hieronder zie je een verkorte versie van de workflow in actie: