Strumenti per agenti basati su Pandas
Definizioni degli strumenti per le operazioni read_csv, groupby, merge e describe
Strumenti per agenti basati su Pandas è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 2 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Strumenti pandas per gli agent dei dati
Anziché generare codice arbitrario, è possibile fornire a un agent dei dati un insieme di strumenti pandas ben definiti. Ogni strumento esegue una specifica operazione sui dati e restituisce un risultato strutturato.
Questo approccio è più sicuro e prevedibile della generazione di codice libero: l'agent seleziona e concatena gli strumenti invece di scrivere il codice da zero.
Strumento: read_csv
Il punto di ingresso di qualsiasi sessione di analisi dei dati. read_csv carica un file CSV in memoria e restituisce una descrizione delle dimensioni e delle colonne, così l'agent sa su quali dati sta lavorando.
import pandas as pd
from typing import Optional
# In-memory dataframe store (shared across tool calls in one session)
dataframes = {}
def read_csv(path: str, df_name: str = 'df') -> dict:
'''Load a CSV file and return schema info.'''
df = pd.read_csv(path)
dataframes[df_name] = df
return {
'df_name': df_name,
'rows': len(df),
'columns': list(df.columns),
'dtypes': df.dtypes.astype(str).to_dict(),
'sample': df.head(3).to_dict(orient='records'),
'null_counts': df.isnull().sum().to_dict()
}
# Agent usage:
result = read_csv('orders.csv', 'orders')
print(f"Loaded {result['rows']} rows, columns: {result['columns']}")Strumento: filter_dataframe
Filtra le righe in base a una stringa condizionale. La condizione viene valutata sul dataframe usando df.query(), che supporta un sottoinsieme sicuro delle espressioni pandas.
def filter_dataframe(df_name: str, condition: str, result_name: str = None) -> dict:
'''Filter rows matching condition. Returns filtered dataframe info.'''
df = dataframes.get(df_name)
if df is None:
return {'error': f'Dataframe {df_name!r} not found. Call read_csv first.'}
try:
filtered = df.query(condition)
store_name = result_name or f'{df_name}_filtered'
dataframes[store_name] = filtered
return {
'df_name': store_name,
'rows_before': len(df),
'rows_after': len(filtered),
'sample': filtered.head(3).to_dict(orient='records')
}
except Exception as e:
return {'error': f'Filter error: {e}'}
# Example: filter orders from 2024
result = filter_dataframe('orders', 'order_date >= "2024-01-01"', 'orders_2024')
print(f"Filtered to {result['rows_after']} rows")Strumento: groupby_aggregate
Raggruppa le righe per colonna e calcola un'aggregazione. È una delle operazioni di analisi dei dati più richieste: totali, medie e conteggi per categoria.
def groupby_aggregate(df_name: str, group_column: str,
agg_column: str, agg_func: str) -> dict:
'''Group by column and aggregate. agg_func: sum, mean, count, min, max, median'''
df = dataframes.get(df_name)
if df is None:
return {'error': f'Dataframe {df_name!r} not found'}
VALID_FUNCS = {'sum', 'mean', 'count', 'min', 'max', 'median', 'std'}
if agg_func not in VALID_FUNCS:
return {'error': f'Invalid agg_func. Choose from: {VALID_FUNCS}'}
try:
result = df.groupby(group_column)[agg_column].agg(agg_func).reset_index()
result.columns = [group_column, f'{agg_column}_{agg_func}']
result = result.sort_values(f'{agg_column}_{agg_func}', ascending=False)
return {
'result': result.head(20).to_dict(orient='records'),
'total_groups': len(result)
}
except Exception as e:
return {'error': str(e)}
# Agent call: total revenue by product category
print(groupby_aggregate('orders', 'category', 'revenue', 'sum'))Strumento: merge_dataframes
Unisce due dataframe tramite una chiave comune. In questo modo l'agent può combinare dati provenienti da più file, ad esempio unendo gli ordini con una tabella dei clienti per ottenere i nomi dei clienti.
def merge_dataframes(df1_name: str, df2_name: str,
on: str, how: str = 'inner',
result_name: str = 'merged') -> dict:
'''Join two dataframes on a key column.'''
df1 = dataframes.get(df1_name)
df2 = dataframes.get(df2_name)
if df1 is None:
return {'error': f'{df1_name!r} not found'}
if df2 is None:
return {'error': f'{df2_name!r} not found'}
VALID_HOW = {'inner', 'left', 'right', 'outer'}
if how not in VALID_HOW:
return {'error': f'Invalid join type. Choose from: {VALID_HOW}'}
try:
merged = df1.merge(df2, on=on, how=how)
dataframes[result_name] = merged
return {
'df_name': result_name,
'rows': len(merged),
'columns': list(merged.columns),
'sample': merged.head(3).to_dict(orient='records')
}
except Exception as e:
return {'error': str(e)}
# Join orders with customers on customer_id
print(merge_dataframes('orders', 'customers', on='customer_id'))Strumento: describe_dataframe
Restituisce il riepilogo statistico di un dataframe. In genere è il primo strumento chiamato dall'agent dopo il caricamento dei dati, per comprenderne distribuzioni, intervalli e potenziali problemi di qualità.
def describe_dataframe(df_name: str) -> dict:
'''Return statistical summary and data quality info.'''
df = dataframes.get(df_name)
if df is None:
return {'error': f'Dataframe {df_name!r} not found'}
# Numeric stats
numeric_cols = df.select_dtypes(include='number').columns.tolist()
stats = {}
for col in numeric_cols:
s = df[col].describe()
stats[col] = {
'min': round(float(s['min']), 4),
'max': round(float(s['max']), 4),
'mean': round(float(s['mean']), 4),
'median': round(float(df[col].median()), 4),
'std': round(float(s['std']), 4),
'nulls': int(df[col].isnull().sum())
}
# Categorical columns summary
cat_cols = df.select_dtypes(include='object').columns.tolist()
cat_stats = {
col: {
'unique_values': df[col].nunique(),
'top_3': df[col].value_counts().head(3).to_dict()
}
for col in cat_cols
}
return {'numeric': stats, 'categorical': cat_stats, 'total_rows': len(df)}Strumento: sort_and_top_n
Restituisce le prime N righe ordinate in base a una colonna. È utile per domande del tipo "clienti principali", "prodotti migliori" o "ricavi più elevati".
def sort_and_top_n(df_name: str, sort_column: str,
n: int = 10, ascending: bool = False) -> dict:
'''Return top N rows sorted by column.'''
df = dataframes.get(df_name)
if df is None:
return {'error': f'{df_name!r} not found'}
if sort_column not in df.columns:
return {'error': f'Column {sort_column!r} not found. Available: {list(df.columns)}'}
sorted_df = df.sort_values(sort_column, ascending=ascending)
top = sorted_df.head(n)
return {
'rows': top.to_dict(orient='records'),
'total_rows_in_df': len(df),
'sort_column': sort_column,
'sort_order': 'ascending' if ascending else 'descending'
}
# Top 5 orders by total value
print(sort_and_top_n('orders', 'total_amount', n=5))Strumento: pivot_table
Crea una tabella pivot, ovvero una tabulazione incrociata di due colonne categoriali con un valore aggregato. È ideale per domande del tipo "ricavi per area geografica e categoria di prodotto".
def pivot_table(df_name: str, index: str, columns: str,
values: str, aggfunc: str = 'sum') -> dict:
'''Create a pivot table.'''
df = dataframes.get(df_name)
if df is None:
return {'error': f'{df_name!r} not found'}
VALID_FUNCS = {'sum', 'mean', 'count', 'min', 'max'}
if aggfunc not in VALID_FUNCS:
return {'error': f'Invalid aggfunc. Choose from: {VALID_FUNCS}'}
try:
pivot = df.pivot_table(
index=index, columns=columns,
values=values, aggfunc=aggfunc,
fill_value=0
)
# Convert to nested dict for JSON serialization
result = pivot.round(2).to_dict()
return {
'pivot': result,
'index': index,
'columns': columns,
'values': values,
'aggfunc': aggfunc
}
except Exception as e:
return {'error': str(e)}Strumento: calculate_metric
Calcola una singola metrica aziendale, come crescita dei ricavi, tasso di conversione o tasso di abbandono, a partire dalle colonne esistenti del dataframe. È più preciso che chiedere al LLM di scrivere un'espressione pandas.
def calculate_metric(df_name: str, metric: str, params: dict = None) -> dict:
'''Calculate a named business metric from the dataframe.'''
df = dataframes.get(df_name)
if df is None:
return {'error': f'{df_name!r} not found'}
params = params or {}
try:
if metric == 'conversion_rate':
total_col = params.get('total_column', 'total')
converted_col = params.get('converted_column', 'converted')
rate = df[converted_col].sum() / df[total_col].sum() * 100
return {'metric': 'conversion_rate', 'value': round(rate, 2), 'unit': '%'}
elif metric == 'average_order_value':
revenue_col = params.get('revenue_column', 'revenue')
count_col = params.get('count_column', 'order_count')
aov = df[revenue_col].sum() / df[count_col].sum()
return {'metric': 'average_order_value', 'value': round(aov, 2)}
else:
return {'error': f'Unknown metric: {metric}'}
except Exception as e:
return {'error': str(e)}Registrazione degli strumenti presso un agent LLM
Registrare tutti gli strumenti pandas presso l'agent LLM usando il formato di function calling di OpenAI. L'agent decide quale strumento chiamare in base alla domanda dell'utente.
tools = [
{
'type': 'function',
'function': {
'name': 'read_csv',
'description': 'Load a CSV file for analysis',
'parameters': {
'type': 'object',
'properties': {
'path': {'type': 'string', 'description': 'Path to CSV file'},
'df_name': {'type': 'string', 'description': 'Name to reference this dataframe'}
},
'required': ['path']
}
}
},
{
'type': 'function',
'function': {
'name': 'groupby_aggregate',
'description': 'Group data by a column and compute sum, mean, count, min, max, or median',
'parameters': {
'type': 'object',
'properties': {
'df_name': {'type': 'string'},
'group_column': {'type': 'string'},
'agg_column': {'type': 'string'},
'agg_func': {'type': 'string', 'enum': ['sum', 'mean', 'count', 'min', 'max', 'median']}
},
'required': ['df_name', 'group_column', 'agg_column', 'agg_func']
}
}
}
]
if __name__ == '__main__':
print('Registered tools:')
for t in tools:
fn = t['function']
print(f" - {fn['name']}: {fn['description']}")
Concatenazione di più strumenti
Le domande complesse richiedono la concatenazione di più strumenti. L'agent li chiama in sequenza, usando l'output di uno come input per il successivo. Ogni risultato viene restituito all'agent prima della chiamata allo strumento seguente.
import json
import openai
import os
client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
TOOL_MAP = {
'read_csv': read_csv,
'filter_dataframe': filter_dataframe,
'groupby_aggregate': groupby_aggregate,
'describe_dataframe': describe_dataframe,
'merge_dataframes': merge_dataframes,
'sort_and_top_n': sort_and_top_n
}
def run_data_agent(question, data_paths):
messages = [
{'role': 'system', 'content': f'You are a data analyst. Available data: {data_paths}'},
{'role': 'user', 'content': question}
]
for _ in range(10): # max 10 tool calls
response = client.chat.completions.create(
model='gpt-4o', messages=messages, tools=tools
)
msg = response.choices[0].message
if not msg.tool_calls:
return msg.content # final answer
messages.append(msg)
for call in msg.tool_calls:
fn = TOOL_MAP[call.function.name]
args = json.loads(call.function.arguments)
result = fn(**args)
messages.append({'role': 'tool', 'tool_call_id': call.id,
'content': json.dumps(result)})
return 'Max tool calls reached'Verifica delle conoscenze
Qual è il principale vantaggio di fornire strumenti pandas a un agent dei dati rispetto all'utilizzo del pattern dell'interprete del codice?
Riepilogo: strumenti per agent dei dati basati su pandas
Gli strumenti pandas forniscono a un agent dei dati operazioni sicure e ben definite: read_csv, filter_dataframe, groupby_aggregate, merge_dataframes, describe_dataframe, sort_and_top_n e pivot_table.
Ogni strumento convalida i propri input, gestisce gli errori in modo appropriato e restituisce risultati strutturati. Registrare gli strumenti usando il formato di function calling di OpenAI: l'agent concatenerà automaticamente le chiamate agli strumenti per rispondere a domande complesse sui dati che richiedono più passaggi.
Domande Frequenti
La lezione «Strumenti per agenti basati su Pandas» è gratuita?
Sì — il testo completo di «Strumenti per agenti basati su Pandas» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Strumenti per agenti basati su Pandas»?
Definizioni degli strumenti per le operazioni read_csv, groupby, merge e describe Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 2 di 4.
Quanto tempo richiede la lezione «Strumenti per agenti basati su Pandas»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Pattern Code Interpreter per l’analisi dei dati
- Strumenti per agenti basati su Pandas
- Generazione automatizzata di grafici e visualizzazioni
- Agenti per i riepiloghi statistici