Corso
Lavorare con DeepSeek in Python
3 ore
1.4K
DeepSeek V3.2 Speciale è un modello di ragionamento ad alto consumo computazionale ottimizzato per workflow complessi con strumenti. Invece di limitarsi a rispondere direttamente alle domande, è progettato per pianificare, ragionare e coordinare strumenti per creare sistemi di analytics agentici.
In questo tutorial useremo DeepSeek V3.2 Speciale per costruire un Data Analyst Agent AI che si appoggia a qualsiasi CSV. Il flusso è questo:
- L'utente carica un file CSV e pone una domanda in linguaggio naturale
- Un planner agent converte la domanda in un piano di analisi JSON strutturato
- Poi alcuni strumenti Python eseguono il piano e generano tabelle/grafici.
- Infine, un explainer agent trasforma i numeri grezzi in un riepilogo di insight.
Alla fine avrai un'app Streamlit con piani espliciti e passaggi di analisi riproducibili. Se vuoi approfondire gli agenti AI, ti consiglio di dare un'occhiata al percorso di competenze AI Agent Fundamentals.
Che cos'è DeepSeek V3.2 Speciale?
DeepSeek V3.2 Speciale è la variante ad alto calcolo, potenziata con reinforcement learning, della famiglia DeepSeek V3.2, progettata specificamente per il deep reasoning e non per la chat informale.
Spinge le prestazioni fino a competere con modelli come Gemini 3 Pro su compiti complessi, e ha persino ottenuto risultati da medaglia d'oro in benchmark come IMO, CMO, ICPC World Finals e IOI 2025.
DeepSeek V3.2 Speciale è ottimizzato per:
- Matematica avanzata, logica e ragionamento multi-step
- Supporta anche workflow agentici in cui il modello pianifica azioni invece di rispondere in un colpo solo
- Come i suoi rivali, supporta anche contesti lunghi e output strutturati con ragionamento in stile chain-of-thought
V3.2 Speciale supporta solo la modalità thinking e al momento non supporta le chiamate a strumenti tramite l'API ufficiale. Non puoi semplicemente modificare il temperature e aspettarti JSON pulito al primo tentativo. Il modello Speciale ragionerà a lungo e spesso intercalerà il suo ragionamento con la risposta finale.
Progetto di esempio DeepSeek-V3.2-Speciale: AI Data Analyst Agent
In questa sezione implementeremo un AI Data Analyst Agent con un modello DeepSeek V3.2 Speciale incapsulato in un'applicazione Streamlit. In sintesi, ecco cosa fa l'app finale:
- Carica qualsiasi file CSV. Puoi trovare buoni esempi su Kaggle (ad esempio le previsioni di vendita, che ho usato in questa demo)
- L'app rileva automaticamente le colonne data e aggiunge i campi di supporto year, month e year_month.
- Scrivi una domanda e DeepSeek V3.2 Speciale agisce come Planner Agent rispondendo con un piano JSON:
- quali colonne usare per il group by
- quali metriche aggregare
- filtri opzionali (es. “Year == 2019”)
- se disegnare un bar chart, line chart o pie chart
- Un Data Worker puramente in Python esegue il piano con pandas.
- Un Chart Generator crea un grafico a barre/linee/torta con matplotlib e seaborn.
- Un Explainer Agent riassume il risultato in linguaggio semplice ed evidenzia i principali insight.
Costruiamolo passo dopo passo.
Passo 1: Prerequisiti
Per prima cosa, installa le dipendenze principali e recupera una chiave API dalla piattaforma API di DeepSeek:
pip install streamlit pandas matplotlib seaborn python-dotenv openaiUseremo alcune librerie core come streamlit per l'interfaccia web interattiva, pandas per caricare e manipolare i dati CSV, matplotlib e seaborn per creare grafici e visualizzazioni. La libreria python-dotenv serve a caricare variabili d'ambiente da un file .env, e OpenAI è il client per comunicare con l'API compatibile OpenAI di DeepSeek.
Poi, accedi alla piattaforma API di DeepSeek, vai alla scheda API Keys e clicca su Create a new API key.
Infine, assegna un nome alla tua chiave e copia la chiave generata.
Imposta la tua chiave API DeepSeek come variabile d'ambiente:
export DEEPSEEK_API_KEY="your_api_key_here"Ora l'ambiente è pronto.
Passo 2: Import e client DeepSeek
Con le dipendenze installate, il passo successivo è importarle nello script e collegare il client DeepSeek che alimenterà i nostri planner ed explainer agent.
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,
)Il blocco di codice sopra fa tre cose: importa le librerie principali, applica uno stile ai grafici e carica le variabili d'ambiente da .env. Poi legge DEEPSEEK_API_KEY, la convalida e costruisce un client OpenAI puntato sull'endpoint temporaneo V3.2 Speciale di DeepSeek, in modo che ogni chiamata LLM successiva usi il modello corretto.
Nota che DeepSeek V3.2 Speciale è attualmente servito da un endpoint temporaneo che scade il 15 dicembre 2025. Dopo tale data, l'URL di deployment potrebbe cambiare, ma il nome del modello rimarrà deepseek-reasoner. Per gli endpoint più recenti e i dettagli d'uso, fai riferimento alla documentazione ufficiale di DeepSeek.
Con il client configurato, definiamo i due agent: uno per pianificare l'analisi e un altro per spiegare i risultati.
Passo 3: Planner Agent
Con il client pronto, possiamo costruire il Planner Agent, che decide quale analisi eseguire sui dati. Invece di lasciare che il modello manipoli direttamente il DataFrame, gli chiediamo di produrre un piano JSON strutturato che il nostro “data worker” Python possa eseguire in sicurezza.
Il planner vede una descrizione testuale del dataset insieme alla domanda dell'utente e restituisce un oggetto JSON che descrive l'operazione, le colonne di raggruppamento, i filtri, la metrica e il tipo di grafico.
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)Per prima cosa, il system_prompt definisce il contratto per il Planner Agent, che converte la domanda dell'utente in un piano JSON e specifica esattamente quali campi ci aspettiamo, ad esempio un elenco di colonne group_by, filters opzionali, un target_column, una metric e un chart_type.
Aggiungiamo anche alcune euristiche (come gestire “quest'anno” o “top N”) così che il modello si comporti più come un junior data analyst che come un chatbot generico.
Poi costruiamo il messaggio di sistema che porta le istruzioni e raggruppa due elementi: una descrizione testuale dello schema del dataset (nomi colonne, dtypes, righe di esempio) e la domanda dell'utente. Questo dà al modello abbastanza contesto per scegliere nomi di colonne validi e un'aggregazione sensata per la domanda posta.
DeepSeek V3.2 Speciale viene quindi chiamato tramite client.chat.completions.create(). Poiché è un modello di ragionamento, può produrre due canali di output: una stringa di contenuto visibile e una stringa interna reasoning_content dove spesso risiedono la maggior parte del “pensiero” e la risposta finale. Estraiamo entrambi dalla risposta e teniamo traccia del finish_reason per rilevare eventuali troncamenti.
Il grosso del lavoro avviene nella funzione extract_json_from_text(). Poiché i modelli di ragionamento non restituiscono sempre JSON pulito, lo incapsuliamo in markdown. Per rendere il planner robusto, questa funzione di supporto prova diverse strategie in sequenza:
- Prima tenta di analizzare l'intera stringa con
json.loads - Poi cerca blocchi di codice JSON e analizza il contenuto all'interno
- Infine, scansiona blocchi bilanciati usando una piccola routine di matching delle parentesi e usa un decoder JSON grezzo.
Per ogni candidato, accetta solo oggetti che sembrano un piano valido, cioè devono contenere almeno uno dei campi operation, target_column, metric o group_by.
Infine, colleghiamo tutto applicando la funzione extract_json_from_text() a reasoning_content.
Insieme, questi elementi trasformano qualsiasi domanda in linguaggio naturale e uno snapshot dello schema in un piano di analisi strutturato.
Passo 4: Explainer Agent
Una volta che il Data Worker ha eseguito l'analisi con Pandas, inviamo la tabella dei risultati insieme al piano JSON a DeepSeek e trasformiamo tutto in un riepilogo con insight concreti.
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)L'explainer agent inizia impostando il system_prompt, che istruisce DeepSeek a comportarsi come un senior data analyst che parla a stakeholder di business. Specifichiamo l'esatta forma della risposta desiderata, insieme ad altre indicazioni per redigere il testo.
Poi prepariamo il user_content, che fornisce al modello tutto il contesto necessario. Passiamo la domanda originale, il piano di analisi in JSON (così sa quali calcoli sono stati eseguiti) e il numero totale di righe.
Quindi componiamo la lista dei messaggi e chiamiamo DeepSeek V3.2 Speciale tramite client.chat.completions.create(). Come nel planner, il modello può restituire testo sia in content che in reasoning_content, quindi includiamo entrambi e ricorriamo a reasoning_content se content è vuoto.
Infine, la funzione restituisce una tupla con sia la spiegazione sia la stringa di ragionamento grezza che compare nell'expander.
Con questo Explainer Agent, il ciclo è completo:
- Il Planner Agent decide cosa calcolare,
- Il Data Worker esegue il piano con pandas e matplotlib, e
- L'Explainer Agent traduce i numeri in insight azionabili
Passo 5: Funzioni di supporto
Prima di permettere a DeepSeek di pianificare qualsiasi cosa, dobbiamo descrivere chiaramente i dati e renderli adatti all'analisi. Questo si ottiene con una serie di funzioni di supporto.
Passo 5.1 Planner dello schema
Il Planner Agent non può vedere direttamente il tuo DataFrame, quindi gli inviamo un riepilogo testuale compatto del 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" + sampleQuesto helper scorre tutte le colonne. Include anche alcuni valori di esempio per elementi come nomi di prodotto, regioni o categorie. F
inalmente, aggiunge una breve tabella in markdown delle prime righe, così il modello ha esempi concreti su cui ragionare.
Questa stringa di schema è l'occhio del planner sui tuoi dati e gli permette di scegliere nomi di colonne e group by validi.
Passo 5.2 Preprocessing automatico delle date
Rileviamo automaticamente le colonne data e aggiungiamo campi di supporto come <col>_year e <col>_year_month, il che rende più facile per DeepSeek rispondere a domande come “lo scorso anno” o “per mese”.
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 dfPer prima cosa scansioniamo le colonne e, se una colonna è già datetime o il suo nome suggerisce una data (come order_date o day), proviamo a convertirla con pd.to_datetime. Per ogni colonna data, creiamo feature come <col>_year, <col>_month e <col>_year_month.
Questo preprocessing consente al planner di fare riferimento con sicurezza a colonne year e year_month senza che tu debba creare manualmente feature temporali per ogni dataset.
Passo 5.3 Applicazione dei filtri
Il planner può aggiungere semplici filtri di riga al piano e questo helper li trasforma in filtri pandas.
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 filteredPer ogni filtro, verifichiamo che la colonna esista, convertiamo il valore in numero se possibile e poi applichiamo il confronto appropriato. Tutti i filtri sono applicati in sequenza, così puoi combinare naturalmente condizioni come “Region contiene West” e “Year >= 2021”.
Questa funzione collega i vincoli in linguaggio naturale del piano ai sottoinsiemi di righe.
Passo 5.4: Piano di analisi
Ora implementiamo il Data Worker, un sottile livello che legge il piano JSON ed esegue le relative aggregazioni pandas.
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)Iniziamo scompattando il piano con operazione, colonna target, colonne di raggruppamento, filtri e metrica. T
opodiché i filtri vengono applicati tramite la funzione apply_filters(). Se il piano fa riferimento a una colonna non valida, torniamo a una semplice anteprima.
Altrimenti, per le operazioni supportate, aggreghiamo sull'intera tabella o eseguiamo un groupby seguito da operazioni come somma, media, conteggio, massimo o minimo, e ordiniamo il risultato per la colonna della metrica.
Questo è il cuore del layer di strumenti: l'LLM decide cosa calcolare e questa funzione decide come calcolarlo in sicurezza con pandas.
Passo 5.5: Generatore di grafici
Infine, trasformiamo la tabella risultante in un grafico che corrisponda al chart_type del piano, così da avere sia numeri sia visualizzazioni in output.
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 bufLeggiamo chart_type, group_by e target_column dal piano e costruamo una figura Matplotlib. La funzione generate_chart() sopra restituisce un buffer PNG in memoria che l'app Streamlit mostra accanto alle metriche chiave.
Con queste funzioni di supporto, il resto del sistema diventa molto più semplice. Il Planner usa get_schema_description per progettare l'analisi, il Data Worker usa le funzioni preprocess_dates(), apply_filters() e run_analysis_plan() per macinare i numeri, e generate_chart() trasforma la tabella finale in un grafico.
Passo 6: UI Streamlit
Ora che abbiamo planner, data worker, explainer e funzioni di supporto, possiamo incapsulare tutto in un'app Streamlit.
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")All'inizio configuriamo la pagina con st.set_page_config e CSS per stilizzare la dashboard. L'utente prima carica un CSV, che carichiamo in un DataFrame, passiamo attraverso preprocess_dates e mostriamo all'interno dell'expander Preview Data con conteggi di righe/colonne, uso di memoria e un rapido riepilogo.
Dietro le quinte, costruiamo uno schema con get_schema_description(), chiamiamo call_planner_llm per ottenere un piano JSON e mostriamo sia il piano sia il ragionamento del modello in un expander. Il Data Worker esegue quindi il piano con pandas e la sezione Results mostra un grafico da generate_chart insieme alle metriche chiave.
Infine, chiamiamo call_explainer_llm con domanda, piano e riepilogo dei risultati per generare una spiegazione per il box AI Insights, con un expander opzionale AI Reasoning Process per chi vuole sbirciare sotto il cofano.
Salva tutto come app.py ed esegui:
streamlit run app.pyNel video qui sotto, puoi vedere una versione ridotta del workflow in azione:
