Conversational data analysis using natural language queries on DataFrames. Chat with your data using LLMs to generate insights, create visualizations, and explain code.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
Instruções da origem · Visualização somente leitura
name
pandasai
description
Conversational data analysis using natural language queries on DataFrames. Chat with your data using LLMs to generate insights, create visualizations, and explain code.
# Core installation
pip install pandasai>=2.0.0
# For OpenAI backend (recommended)
pip install pandasai[openai]
# For visualization
pip install pandasai matplotlib seaborn plotly
# For Excel/CSV handling
pip install pandasai openpyxl xlrd
# For database connections
pip install pandasai sqlalchemy psycopg2-binary pymysql
# Environment setupexport OPENAI_API_KEY="sk-..."# Or for other providersexport ANTHROPIC_API_KEY="sk-ant-..."export GOOGLE_API_KEY=
"..."
Verify Installation
import pandasai
from pandasai import SmartDataframe
import pandas as pd
print(f"PandasAI version: {pandasai.__version__}")
# Quick test
df = pd.DataFrame({"name": ["Alice", "Bob"], "age": [25, 30]})
smart_df = SmartDataframe(df)
print("PandasAI installed successfully!")
Core Capabilities
1. Basic Natural Language Queries
Simple DataFrame Conversations:
"""
Query DataFrames using natural language with PandasAI.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
defcreate_smart_dataframe(
df: pd.DataFrame,
model: str = "gpt-4",
temperature: float = 0.0) -> SmartDataframe:
"""
Create a SmartDataframe for natural language queries.
Args:
df: Input pandas DataFrame
model: LLM model to use
temperature: Sampling temperature (0 for deterministic)
Returns:
SmartDataframe instance ready for queries
"""# Initialize LLM
llm = OpenAI(
model=model,
temperature=temperature
)
# Create SmartDataframe
smart_df = SmartDataframe(
df,
config={
"llm": llm,
"verbose": True,
"enable_cache": True,
"conversational": False
}
)
return smart_df
defquery_dataframe(smart_df: SmartDataframe, question: str) -> any:
"""
Ask a natural language question about the data.
Args:
smart_df: SmartDataframe instance
question: Natural language question
Returns:
Query result (can be value, DataFrame, or visualization)
"""try:
result = smart_df.chat(question)
return result
except Exception as e:
print(f"Query error: {e}")
returnNone# Usage Example# Create sample sales data
sales_data = pd.DataFrame({
"date": pd.date_range("2025-01-01", periods=100, freq="D"),
"product": ["Widget A", "Widget B", "Widget C", "Widget D"] * 25,
"region": ["North", "South", "East", "West"] * 25,
"units_sold": [50, 75, 100, 125] * 25,
"revenue": [500, 1125, 2000, 3125] * 25,
"cost": [300, 600, 1200, 1800] * 25
})
# Create SmartDataframe
smart_sales = create_smart_dataframe(sales_data)
# Ask questions in natural language
questions = [
"What is the total revenue?",
"Which product has the highest average revenue?",
"What is the profit margin by product?",
"Show me the top 5 days by units sold"
]
for question in questions:
print(f"\nQ: {question}")
result = query_dataframe(smart_sales, question)
print(f"A: {result}")
Aggregation and Statistical Queries:
"""
Perform aggregations and statistical analysis using natural language.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
defcreate_analytics_interface(df: pd.DataFrame) -> SmartDataframe:
"""Create an analytics interface for statistical queries."""
llm = OpenAI(model="gpt-4", temperature=0)
return SmartDataframe(
df,
config={
"llm": llm,
"verbose": False,
"enable_cache": True,
"custom_whitelisted_dependencies": ["scipy", "numpy"]
}
)
# Create sample employee performance data
employee_data = pd.DataFrame({
"employee_id": range(1, 101),
"department": ["Engineering", "Sales", "Marketing", "HR", "Finance"] * 20,
"experience_years": [1, 2, 3, 5, 7, 10, 12, 15, 18, 20] * 10,
"performance_score": [round(3.0 + i * 0.02, 2) for i inrange(100)],
"salary": [50000 + i * 500for i inrange(100)],
"projects_completed": [5 + i % 20for i inrange(100)]
})
smart_employees = create_analytics_interface(employee_data)
# Statistical queries
statistical_questions = [
"What is the average salary by department?",
"Calculate the correlation between experience_years and salary",
"What is the standard deviation of performance scores?",
"Find the median projects completed by department",
"Which department has the highest variance in salary?",
"Show the salary distribution statistics (mean, median, std, min, max)"
]
print("Statistical Analysis Results:")
print("=" * 50)
for question in statistical_questions:
print(f"\nQuestion: {question}")
result = smart_employees.chat(question)
print(f"Answer: {result}")
2. Chart Generation and Visualization
Creating Charts from Natural Language:
"""
Generate visualizations using natural language descriptions.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
import matplotlib.pyplot as plt
from pathlib import Path
defcreate_visualization_interface(
df: pd.DataFrame,
save_charts: bool = True,
charts_path: str = "./charts") -> SmartDataframe:
"""
Create interface optimized for chart generation.
Args:
df: Input DataFrame
save_charts: Whether to save generated charts
charts_path: Directory to save charts
Returns:
SmartDataframe configured for visualization
"""
llm = OpenAI(model="gpt-4", temperature=0)
# Create charts directory
Path(charts_path).mkdir(parents=True, exist_ok=True)
return SmartDataframe(
df,
config={
"llm": llm,
"save_charts": save_charts,
"save_charts_path": charts_path,
"verbose": True,
"custom_whitelisted_dependencies": [
"matplotlib",
"seaborn",
"plotly"
]
}
)
# Create sample time series dataimport numpy as np
np.random.seed(42)
dates = pd.date_range("2024-01-01", periods=365, freq="D")
time_series_data = pd.DataFrame({
"date": dates,
"sales": np.random.normal(1000, 200, 365).cumsum(),
"customers": np.random.poisson(50, 365),
"category": ["Electronics", "Clothing", "Food", "Home"] * 91 + ["Electronics"],
"region": ["North", "South", "East", "West", "Central"] * 73
})
smart_viz = create_visualization_interface(time_series_data)
# Generate various chart types
chart_prompts = [
"Create a line chart showing sales trend over time",
"Plot a bar chart of average sales by category",
"Generate a pie chart showing customer distribution by region",
"Create a scatter plot of sales vs customers",
"Make a heatmap showing sales by category and region",
"Plot monthly sales with a trend line"
]
print("Generating visualizations...")
for prompt in chart_prompts:
print(f"\nGenerating: {prompt}")
result = smart_viz.chat(prompt)
print(f"Result: {result}")
defgenerate_dashboard_charts(
df: pd.DataFrame,
output_dir: str = "./dashboard") -> dict:
"""
Generate multiple charts for a dashboard.
Args:
df: Input DataFrame
output_dir: Directory to save charts
Returns:
Dictionary with chart paths and descriptions
"""
smart_df = create_visualization_interface(df, charts_path=output_dir)
dashboard_charts = {
"trend": "Create a line chart showing the main trend over time",
"distribution": "Create a histogram showing the distribution of the primary metric",
"comparison": "Create a bar chart comparing categories",
"relationship": "Create a scatter plot showing key relationships",
"composition": "Create a pie chart showing composition breakdown"
}
results = {}
for chart_name, prompt in dashboard_charts.items():
try:
result = smart_df.chat(prompt)
results[chart_name] = {
"prompt": prompt,
"result": result,
"status": "success"
}
except Exception as e:
results[chart_name] = {
"prompt": prompt,
"error": str(e),
"status": "failed"
}
return results
# Generate dashboard
dashboard_results = generate_dashboard_charts(time_series_data)
for name, info in dashboard_results.items():
print(f"\n{name}: {info['status']}")
Advanced Visualization Customization:
"""
Customize chart appearance and style using natural language.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
defcreate_styled_visualization(
df: pd.DataFrame,
style_config: dict = None) -> SmartDataframe:
"""
Create visualization interface with custom styling.
Args:
df: Input DataFrame
style_config: Custom style configuration
Returns:
SmartDataframe with styling configuration
"""
default_style = {
"figure.figsize": (12, 6),
"axes.titlesize": 14,
"axes.labelsize": 12,
"xtick.labelsize": 10,
"ytick.labelsize": 10,
"legend.fontsize": 10
}
style = {**default_style, **(style_config or {})}
# Apply matplotlib styleimport matplotlib.pyplot as plt
plt.rcParams.update(style)
llm = OpenAI(model="gpt-4", temperature=0)
return SmartDataframe(
df,
config={
"llm": llm,
"verbose": True,
"custom_whitelisted_dependencies": [
"matplotlib",
"seaborn"
]
}
)
# Sample financial data
financial_data = pd.DataFrame({
"quarter": ["Q1", "Q2", "Q3", "Q4"] * 3,
"year": [2023, 2023, 2023, 2023, 2024, 2024, 2024, 2024, 2025, 2025, 2025, 2025],
"revenue": [100, 120, 115, 140, 150, 165, 180, 200, 210, 225, 240, 260],
"expenses": [80, 90, 85, 100, 110, 120, 130, 145, 150, 160, 170, 185],
"profit": [20, 30, 30, 40, 40, 45, 50, 55, 60, 65, 70, 75]
})
smart_finance = create_styled_visualization(financial_data)
# Detailed visualization requests
visualization_requests = [
"""Create a grouped bar chart comparing revenue and expenses by quarter,
with a secondary line showing profit trend. Use blue for revenue,
red for expenses, and green for profit.""",
"""Generate a stacked area chart showing revenue, expenses, and profit
over time with year-quarter on x-axis. Use a professional color palette.""",
"""Create a dual-axis chart with revenue bars on the left axis and
profit margin percentage line on the right axis."""
]
for request in visualization_requests:
print(f"\nRequest: {request[:50]}...")
result = smart_finance.chat(request)
print(f"Result type: {type(result)}")
3. Code Explanation and Generation
Understanding Generated Code:
"""
Get explanations of generated code and data transformations.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
defcreate_explanatory_interface(df: pd.DataFrame) -> SmartDataframe:
"""Create interface that provides code explanations."""
llm = OpenAI(model="gpt-4", temperature=0)
return SmartDataframe(
df,
config={
"llm": llm,
"verbose": True,
"enable_cache": False, # Disable to always show code"explain": True# Enable explanations
}
)
defquery_with_explanation(
smart_df: SmartDataframe,
question: str) -> dict:
"""
Query data and get both result and explanation.
Args:
smart_df: SmartDataframe instance
question: Natural language question
Returns:
Dictionary with result, code, and explanation
"""# Get the result
result = smart_df.chat(question)
# Get the last generated code
last_code = smart_df.last_code_generated
# Ask for explanation
explanation_prompt = f"""
Explain what this code does in simple terms:
```python
{last_code}
```
"""
explanation = smart_df.chat(explanation_prompt)
return {
"question": question,
"result": result,
"code": last_code,
"explanation": explanation
}
# Sample inventory data
inventory_data = pd.DataFrame({
"sku": [f"SKU-{i:04d}"for i inrange(1, 51)],
"product_name": [f"Product {i}"for i inrange(1, 51)],
"category": ["Electronics", "Clothing", "Home", "Sports", "Books"] * 10,
"quantity": [100, 50, 200, 75, 150] * 10,
"unit_price": [29.99, 49.99, 19.99, 79.99, 14.99] * 10,
"reorder_level": [20, 10, 40, 15, 30] * 10,
"supplier_id": ["SUP-A", "SUP-B", "SUP-C", "SUP-D", "SUP-E"] * 10
})
smart_inventory = create_explanatory_interface(inventory_data)
# Complex queries with explanations
complex_queries = [
"Which products are below their reorder level?",
"Calculate the total inventory value by category",
"Find suppliers with the most products and their total inventory value",
"What percentage of products in each category are below reorder level?"
]
print("Queries with Code Explanations")
print("=" * 60)
for query in complex_queries:
result = query_with_explanation(smart_inventory, query)
print(f"\nQuestion: {result['question']}")
print(f"\nGenerated Code:\n{result['code']}")
print(f"\nResult: {result['result']}")
print("-" * 40)
Custom Code Generation:
"""
Generate custom analysis code using natural language specifications.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
defgenerate_analysis_code(
df: pd.DataFrame,
analysis_description: str) -> str:
"""
Generate analysis code from natural language description.
Args:
df: Sample DataFrame for context
analysis_description: Natural language description of desired analysis
Returns:
Generated Python code as string
"""
llm = OpenAI(model="gpt-4", temperature=0)
smart_df = SmartDataframe(
df,
config={
"llm": llm,
"verbose": True,
"response_parser": None# Get raw code
}
)
# Generate the code
smart_df.chat(analysis_description)
return smart_df.last_code_generated
defcreate_analysis_function(
df: pd.DataFrame,
function_spec: str) -> callable:
"""
Create a reusable analysis function from specification.
Args:
df: Sample DataFrame for context
function_spec: Natural language specification
Returns:
Callable function
"""
code = generate_analysis_code(df, function_spec)
# Extract and wrap the code
wrapped_code = f"""
def analysis_function(df):
import pandas as pd
import numpy as np
{code}
return result
"""# Execute to define function
local_vars = {}
exec(wrapped_code, {"pd": pd, "np": __import__("numpy")}, local_vars)
return local_vars["analysis_function"]
# Example: Generate reusable analysis code
sample_df = pd.DataFrame({
"date": pd.date_range("2024-01-01", periods=30),
"metric_a": range(30),
"metric_b": range(30, 60)
})
# Generate code for specific analysis
code = generate_analysis_code(
sample_df,
"""Create a rolling 7-day average of metric_a and identify
days where metric_b exceeds 2 standard deviations from its mean"""
)
print("Generated Analysis Code:")
print("=" * 40)
print(code)
4. Multiple LLM Backends
Configuring Different LLM Providers:
"""
Use various LLM backends with PandasAI.
"""import pandas as pd
from pandasai import SmartDataframe
from typing importOptionaldefcreate_openai_backend(
model: str = "gpt-4",
temperature: float = 0.0,
api_key: Optional[str] = None):
"""Create OpenAI LLM backend."""from pandasai.llm import OpenAI
return OpenAI(
model=model,
temperature=temperature,
api_key=api_key # Uses env var if None
)
defcreate_anthropic_backend(
model: str = "claude-3-opus-20240229",
temperature: float = 0.0,
api_key: Optional[str] = None):
"""Create Anthropic Claude LLM backend."""from pandasai.llm import Anthropic
return Anthropic(
model=model,
temperature=temperature,
api_key=api_key
)
defcreate_google_backend(
model: str = "gemini-pro",
temperature: float = 0.0,
api_key: Optional[str] = None):
"""Create Google Gemini LLM backend."""from pandasai.llm import GoogleGemini
return GoogleGemini(
model=model,
temperature=temperature,
api_key=api_key
)
defcreate_local_backend(
model_path: str,
model_type: str = "llama"):
"""Create local LLM backend using Ollama or similar."""from pandasai.llm import LocalLLM
return LocalLLM(
model_path=model_path,
model_type=model_type
)
defcreate_azure_openai_backend(
deployment_name: str,
api_base: str,
api_version: str = "2024-02-15-preview",
api_key: Optional[str] = None):
"""Create Azure OpenAI LLM backend."""from pandasai.llm import AzureOpenAI
return AzureOpenAI(
deployment_name=deployment_name,
api_base=api_base,
api_version=api_version,
api_key=api_key
)
classMultiBackendAnalyzer:
"""
Analyzer supporting multiple LLM backends with fallback.
"""def__init__(self, df: pd.DataFrame):
self.df = df
self.backends = {}
self.current_backend = Nonedefadd_backend(self, name: str, llm):
"""Add an LLM backend."""self.backends[name] = llm
ifself.current_backend isNone:
self.current_backend = name
defset_backend(self, name: str):
"""Set the active backend."""if name notinself.backends:
raise ValueError(f"Backend '{name}' not found")
self.current_backend = name
defquery(self, question: str, backend: str = None) -> any:
"""
Query using specified or current backend.
Args:
question: Natural language question
backend: Backend name (uses current if None)
Returns:
Query result
"""
backend_name = backend orself.current_backend
llm = self.backends[backend_name]
smart_df = SmartDataframe(
self.df,
config={"llm": llm, "verbose": False}
)
return smart_df.chat(question)
defquery_with_fallback(
self,
question: str,
backend_order: list = None) -> tuple:
"""
Query with automatic fallback to other backends on failure.
Args:
question: Natural language question
backend_order: Order of backends to try
Returns:
Tuple of (result, backend_used)
"""
backends_to_try = backend_order orlist(self.backends.keys())
for backend_name in backends_to_try:
try:
result = self.query(question, backend_name)
return result, backend_name
except Exception as e:
print(f"Backend '{backend_name}' failed: {e}")
continueraise RuntimeError("All backends failed")
# Usage Example
sample_data = pd.DataFrame({
"product": ["A", "B", "C"],
"sales": [100, 200, 150]
})
# Create multi-backend analyzer
analyzer = MultiBackendAnalyzer(sample_data)
# Add backends (uncomment based on available API keys)
analyzer.add_backend("openai", create_openai_backend("gpt-4"))
# analyzer.add_backend("anthropic", create_anthropic_backend())# analyzer.add_backend("google", create_google_backend())# Query with specific backend
result = analyzer.query("What is the total sales?", backend="openai")
print(f"Result: {result}")
# Query with fallback# result, used_backend = analyzer.query_with_fallback("What is the average sales?")# print(f"Result: {result} (using {used_backend})")
5. Multi-DataFrame Analysis
Analyzing Multiple Related DataFrames:
"""
Work with multiple related DataFrames using PandasAI.
"""import pandas as pd
from pandasai import SmartDatalake
from pandasai.llm import OpenAI
defcreate_smart_datalake(
dataframes: dict,
model: str = "gpt-4") -> SmartDatalake:
"""
Create a SmartDatalake for multi-DataFrame analysis.
Args:
dataframes: Dictionary of name: DataFrame pairs
model: LLM model to use
Returns:
SmartDatalake instance
"""
llm = OpenAI(model=model, temperature=0)
# Convert dict to list with names
df_list = []
for name, df in dataframes.items():
df.name = name # Assign name attribute
df_list.append(df)
return SmartDatalake(
df_list,
config={
"llm": llm,
"verbose": True,
"enable_cache": True
}
)
# Create related business DataFrames
customers = pd.DataFrame({
"customer_id": range(1, 101),
"customer_name": [f"Customer {i}"for i inrange(1, 101)],
"segment": ["Enterprise", "SMB", "Startup"] * 33 + ["Enterprise"],
"region": ["North", "South", "East", "West"] * 25,
"signup_date": pd.date_range("2023-01-01", periods=100, freq="3D")
})
customers.name = "customers"
orders = pd.DataFrame({
"order_id": range(1, 501),
"customer_id": [i % 100 + 1for i inrange(500)],
"order_date": pd.date_range("2024-01-01", periods=500, freq="6H"),
"product_id": [i % 20 + 1for i inrange(500)],
"quantity": [1, 2, 3, 5, 10] * 100,
"unit_price": [29.99, 49.99, 99.99, 149.99, 199.99] * 100
})
orders.name = "orders"
products = pd.DataFrame({
"product_id": range(1, 21),
"product_name": [f"Product {i}"for i inrange(1, 21)],
"category": ["Electronics", "Software", "Hardware", "Services"] * 5,
"cost": [15, 25, 50, 75, 100] * 4
})
products.name = "products"# Create SmartDatalake
datalake = create_smart_datalake({
"customers": customers,
"orders": orders,
"products": products
})
# Cross-DataFrame queries
cross_df_queries = [
"What is the total revenue by customer segment?",
"Which products are most popular among Enterprise customers?",
"Show the monthly order trend by product category",
"Find customers who have never ordered",
"What is the average order value by region?",
"Which product category has the highest profit margin?"
]
print("Multi-DataFrame Analysis")
print("=" * 50)
for query in cross_df_queries:
print(f"\nQuery: {query}")
result = datalake.chat(query)
print(f"Result: {result}")
Building a Data Warehouse Interface:
"""
Create a conversational interface for data warehouse queries.
"""import pandas as pd
from pandasai import SmartDatalake
from pandasai.llm import OpenAI
from typing importDict, List, Optionalfrom dataclasses import dataclass
@dataclassclassTableSchema:
"""Schema definition for a table."""
name: str
description: str
columns: Dict[str, str] # column_name: description
primary_key: str
foreign_keys: Dict[str, str] = None# column: referenced_table.columnclassDataWarehouseInterface:
"""
Conversational interface for data warehouse exploration.
"""def__init__(self, model: str = "gpt-4"):
self.llm = OpenAI(model=model, temperature=0)
self.tables: Dict[str, pd.DataFrame] = {}
self.schemas: Dict[str, TableSchema] = {}
self.datalake: Optional[SmartDatalake] = Nonedefadd_table(
self,
name: str,
df: pd.DataFrame,
schema: TableSchema = None):
"""Add a table to the data warehouse."""
df.name = name
self.tables[name] = df
if schema:
self.schemas[name] = schema
# Rebuild datalakeself._rebuild_datalake()
def_rebuild_datalake(self):
"""Rebuild the SmartDatalake with current tables."""ifself.tables:
self.datalake = SmartDatalake(
list(self.tables.values()),
config={
"llm": self.llm,
"verbose": True,
"enable_cache": True
}
)
defget_table_info(self) -> str:
"""Get information about available tables."""
info_lines = ["Available Tables:"]
for name, schema inself.schemas.items():
info_lines.append(f"\n{name}: {schema.description}")
info_lines.append(f" Primary Key: {schema.primary_key}")
info_lines.append(" Columns:")
for col, desc in schema.columns.items():
info_lines.append(f" - {col}: {desc}")
return"\n".join(info_lines)
defquery(self, question: str) -> any:
"""
Query the data warehouse using natural language.
Args:
question: Natural language question
Returns:
Query result
"""ifnotself.datalake:
raise ValueError("No tables loaded")
returnself.datalake.chat(question)
defsuggest_queries(self, table_name: str = None) -> List[str]:
"""
Suggest useful queries based on available data.
Args:
table_name: Specific table to focus on (optional)
Returns:
List of suggested queries
"""
suggestions = []
if table_name and table_name inself.tables:
df = self.tables[table_name]
# Numeric column suggestions
numeric_cols = df.select_dtypes(include=['number']).columns
for col in numeric_cols[:3]:
suggestions.append(f"What is the average {col} in {table_name}?")
# Categorical column suggestions
cat_cols = df.select_dtypes(include=['object', 'category']).columns
for col in cat_cols[:2]:
suggestions.append(f"Show distribution of {col} in {table_name}")
else:
# Cross-table suggestionsiflen(self.tables) > 1:
tables = list(self.tables.keys())
suggestions.append(f"How are {tables[0]} and {tables[1]} related?")
suggestions.append("What are the key metrics across all tables?")
return suggestions
# Usage
warehouse = DataWarehouseInterface()
# Add tables with schemas
warehouse.add_table(
"customers",
customers,
TableSchema(
name="customers",
description="Customer master data",
columns={
"customer_id": "Unique customer identifier",
"customer_name": "Customer company name",
"segment": "Business segment (Enterprise/SMB/Startup)",
"region": "Geographic region"
},
primary_key="customer_id"
)
)
warehouse.add_table(
"orders",
orders,
TableSchema(
name="orders",
description="Customer orders",
columns={
"order_id": "Unique order identifier",
"customer_id": "Reference to customer",
"order_date": "Date of order",
"quantity": "Quantity ordered",
"unit_price": "Price per unit"
},
primary_key="order_id",
foreign_keys={"customer_id": "customers.customer_id"}
)
)
# Get table informationprint(warehouse.get_table_info())
# Query the warehouse
result = warehouse.query("What is the total revenue by customer segment?")
print(f"\nResult: {result}")
# Get query suggestions
suggestions = warehouse.suggest_queries()
print(f"\nSuggested queries: {suggestions}")
6. Privacy and Security Modes
Configuring Privacy-Preserving Analysis:
"""
Configure PandasAI for privacy-sensitive data analysis.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
from typing importList, Dict, Optionalimport hashlib
defanonymize_column(series: pd.Series, method: str = "hash") -> pd.Series:
"""
Anonymize a column for privacy.
Args:
series: Pandas Series to anonymize
method: Anonymization method ('hash', 'mask', 'generalize')
Returns:
Anonymized Series
"""if method == "hash":
return series.apply(
lambda x: hashlib.md5(str(x).encode()).hexdigest()[:8]
if pd.notna(x) else x
)
elif method == "mask":
return series.apply(
lambda x: str(x)[:2] + "*" * (len(str(x)) - 2)
if pd.notna(x) andlen(str(x)) > 2else"**"
)
elif method == "generalize":
if pd.api.types.is_numeric_dtype(series):
return pd.cut(series, bins=5, labels=["Very Low", "Low", "Medium", "High", "Very High"])
return series
else:
raise ValueError(f"Unknown method: {method}")
defcreate_privacy_safe_interface(
df: pd.DataFrame,
sensitive_columns: List[str],
anonymization_method: str = "hash") -> SmartDataframe:
"""
Create SmartDataframe with anonymized sensitive columns.
Args:
df: Input DataFrame
sensitive_columns: List of columns to anonymize
anonymization_method: Method for anonymization
Returns:
SmartDataframe with anonymized data
"""# Create copy and anonymize
safe_df = df.copy()
for col in sensitive_columns:
if col in safe_df.columns:
safe_df[col] = anonymize_column(safe_df[col], anonymization_method)
llm = OpenAI(model="gpt-4", temperature=0)
return SmartDataframe(
safe_df,
config={
"llm": llm,
"verbose": False,
"enforce_privacy": True,
"enable_cache": False# Don't cache sensitive queries
}
)
classPrivacyAwareAnalyzer:
"""
Analyzer with privacy controls and audit logging.
"""def__init__(
self,
df: pd.DataFrame,
sensitive_columns: List[str],
allowed_operations: List[str] = None):
self.original_df = df
self.sensitive_columns = sensitive_columns
self.allowed_operations = allowed_operations or [
"aggregate", "count", "average", "sum", "distribution"
]
self.query_log = []
# Create anonymized versionself.safe_df = self._create_safe_df()
# Create SmartDataframe
llm = OpenAI(model="gpt-4", temperature=0)
self.smart_df = SmartDataframe(
self.safe_df,
config={
"llm": llm,
"verbose": True
}
)
def_create_safe_df(self) -> pd.DataFrame:
"""Create anonymized DataFrame."""
safe_df = self.original_df.copy()
for col inself.sensitive_columns:
if col in safe_df.columns:
safe_df[col] = anonymize_column(safe_df[col], "hash")
return safe_df
def_check_query_safety(self, query: str) -> bool:
"""
Check if query is safe to execute.
Args:
query: Natural language query
Returns:
True if query is safe
"""
query_lower = query.lower()
# Check for attempts to access sensitive columns directlyfor col inself.sensitive_columns:
iff"show {col}"in query_lower orf"list {col}"in query_lower:
returnFalse# Check if query uses allowed operations
has_allowed_op = any(
op in query_lower for op inself.allowed_operations
)
return has_allowed_op
defquery(self, question: str, user_id: str = "anonymous") -> any:
"""
Execute a privacy-safe query.
Args:
question: Natural language question
user_id: User identifier for audit
Returns:
Query result or error message
"""# Log query attemptself.query_log.append({
"user_id": user_id,
"question": question,
"timestamp": pd.Timestamp.now()
})
# Check query safetyifnotself._check_query_safety(question):
return"Query denied: Attempting to access sensitive data directly"# Execute querytry:
result = self.smart_df.chat(question)
return result
except Exception as e:
returnf"Query error: {e}"defget_audit_log(self) -> pd.DataFrame:
"""Get query audit log."""return pd.DataFrame(self.query_log)
# Example with sensitive data
sensitive_data = pd.DataFrame({
"employee_id": range(1, 101),
"name": [f"Employee {i}"for i inrange(1, 101)], # Sensitive"email": [f"emp{i}@company.com"for i inrange(1, 101)], # Sensitive"ssn": [f"XXX-XX-{i:04d}"for i inrange(1, 101)], # Highly sensitive"department": ["Engineering", "Sales", "Marketing", "HR", "Finance"] * 20,
"salary": [50000 + i * 1000for i inrange(100)],
"performance_score": [3.0 + (i % 20) * 0.1for i inrange(100)]
})
# Create privacy-aware analyzer
analyzer = PrivacyAwareAnalyzer(
sensitive_data,
sensitive_columns=["name", "email", "ssn"],
allowed_operations=["aggregate", "average", "count", "distribution", "compare"]
)
# Safe queries
safe_queries = [
"What is the average salary by department?",
"How many employees are in each department?",
"What is the distribution of performance scores?"
]
print("Privacy-Safe Analysis Results:")
print("=" * 50)
for query in safe_queries:
print(f"\nQuery: {query}")
result = analyzer.query(query, user_id="analyst_001")
print(f"Result: {result}")
# Unsafe query attempt
unsafe_query = "Show me all employee names and emails"print(f"\nUnsafe Query: {unsafe_query}")
result = analyzer.query(unsafe_query, user_id="analyst_001")
print(f"Result: {result}")
# View audit logprint("\nAudit Log:")
print(analyzer.get_audit_log())
Integration Examples
Streamlit Dashboard Integration
"""
Build an interactive data analysis dashboard with Streamlit and PandasAI.
"""import streamlit as st
import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
defcreate_streamlit_dashboard():
"""Create a Streamlit dashboard for conversational data analysis."""
st.title("Conversational Data Analysis")
st.write("Ask questions about your data in natural language")
# File upload
uploaded_file = st.file_uploader("Upload CSV file", type=["csv"])
if uploaded_file isnotNone:
# Load data
df = pd.read_csv(uploaded_file)
# Display data preview
st.subheader("Data Preview")
st.dataframe(df.head())
# Display data info
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Rows", len(df))
with col2:
st.metric("Columns", len(df.columns))
with col3:
st.metric("Memory", f"{df.memory_usage(deep=True).sum() / 1024:.1f} KB")
# Initialize PandasAI
api_key = st.sidebar.text_input("OpenAI API Key", type="password")
if api_key:
llm = OpenAI(api_key=api_key, model="gpt-4", temperature=0)
smart_df = SmartDataframe(df, config={"llm": llm, "verbose": False})
# Query interface
st.subheader("Ask a Question")
# Suggested queries
st.write("**Suggested queries:**")
suggestions = [
f"What is the average of {df.select_dtypes(include='number').columns[0]}?"iflen(df.select_dtypes(include='number').columns) > 0elseNone,
f"Show the distribution of {df.columns[0]}",
"How many rows are there?",
"What are the column names and their types?"
]
suggestions = [s for s in suggestions if s]
selected_suggestion = st.selectbox("Or choose a suggestion:", [""] + suggestions)
# Query input
query = st.text_area(
"Your question:",
value=selected_suggestion,
placeholder="e.g., What is the total sales by region?"
)
if st.button("Ask"):
if query:
with st.spinner("Analyzing..."):
try:
result = smart_df.chat(query)
st.subheader("Result")
# Display based on result typeifisinstance(result, pd.DataFrame):
st.dataframe(result)
elifhasattr(result, 'savefig'):
st.pyplot(result)
else:
st.write(result)
# Show generated codewith st.expander("View Generated Code"):
st.code(smart_df.last_code_generated, language="python")
except Exception as e:
st.error(f"Error: {e}")
else:
st.warning("Please enter your OpenAI API key in the sidebar")
# To run: streamlit run dashboard.py# if __name__ == "__main__":# create_streamlit_dashboard()
FastAPI Service Integration
"""
Create a REST API for PandasAI queries.
"""from fastapi import FastAPI, HTTPException, UploadFile, File
from pydantic import BaseModel
from typing importOptional, List, Dict, Anyimport pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
import io
import json
app = FastAPI(title="PandasAI Query Service", version="1.0.0")
# In-memory storage for datasets
datasets: Dict[str, pd.DataFrame] = {}
smart_dfs: Dict[str, SmartDataframe] = {}
classQueryRequest(BaseModel):
"""Request model for queries."""
dataset_id: str
question: str
include_code: bool = FalseclassQueryResponse(BaseModel):
"""Response model for queries."""
result: Any
code: Optional[str] = None
result_type: strclassDatasetInfo(BaseModel):
"""Information about a dataset."""
dataset_id: str
rows: int
columns: int
column_names: List[str]
dtypes: Dict[str, str]
@app.post("/datasets/upload")asyncdefupload_dataset(
file: UploadFile = File(...),
dataset_id: Optional[str] = None) -> Dict[str, str]:
"""Upload a CSV dataset."""ifnot file.filename.endswith('.csv'):
raise HTTPException(400, "Only CSV files are supported")
# Read CSV
content = await file.read()
df = pd.read_csv(io.BytesIO(content))
# Generate ID if not providedif dataset_id isNone:
dataset_id = file.filename.replace('.csv', '')
# Store dataset
datasets[dataset_id] = df
# Create SmartDataframe
llm = OpenAI(model="gpt-4", temperature=0)
smart_dfs[dataset_id] = SmartDataframe(df, config={"llm": llm, "verbose": False})
return {"dataset_id": dataset_id, "message": "Dataset uploaded successfully"}
@app.get("/datasets/{dataset_id}", response_model=DatasetInfo)asyncdefget_dataset_info(dataset_id: str) -> DatasetInfo:
"""Get information about a dataset."""if dataset_id notin datasets:
raise HTTPException(404, f"Dataset '{dataset_id}' not found")
df = datasets[dataset_id]
return DatasetInfo(
dataset_id=dataset_id,
rows=len(df),
columns=len(df.columns),
column_names=list(df.columns),
dtypes={col: str(dtype) for col, dtype in df.dtypes.items()}
)
@app.post("/query", response_model=QueryResponse)asyncdefquery_dataset(request: QueryRequest) -> QueryResponse:
"""Query a dataset using natural language."""if request.dataset_id notin smart_dfs:
raise HTTPException(404, f"Dataset '{request.dataset_id}' not found")
smart_df = smart_dfs[request.dataset_id]
try:
result = smart_df.chat(request.question)
# Determine result typeifisinstance(result, pd.DataFrame):
result_type = "dataframe"
result = result.to_dict(orient="records")
elifisinstance(result, (int, float)):
result_type = "number"else:
result_type = "text"
result = str(result)
response = QueryResponse(
result=result,
result_type=result_type
)
if request.include_code:
response.code = smart_df.last_code_generated
return response
except Exception as e:
raise HTTPException(500, f"Query failed: {str(e)}")
@app.get("/datasets")asyncdeflist_datasets() -> List[str]:
"""List all available datasets."""returnlist(datasets.keys())
@app.delete("/datasets/{dataset_id}")asyncdefdelete_dataset(dataset_id: str) -> Dict[str, str]:
"""Delete a dataset."""if dataset_id notin datasets:
raise HTTPException(404, f"Dataset '{dataset_id}' not found")
del datasets[dataset_id]
del smart_dfs[dataset_id]
return {"message": f"Dataset '{dataset_id}' deleted"}
# Run with: uvicorn api:app --reload
Jupyter Notebook Integration
"""
Enhanced PandasAI usage in Jupyter notebooks.
"""import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm import OpenAI
from IPython.display import display, Markdown, HTML
classJupyterPandasAI:
"""
Enhanced PandasAI interface for Jupyter notebooks.
"""def__init__(self, df: pd.DataFrame, model: str = "gpt-4"):
self.df = df
self.llm = OpenAI(model=model, temperature=0)
self.smart_df = SmartDataframe(
df,
config={
"llm": self.llm,
"verbose": False,
"save_charts": True,
"save_charts_path": "./notebook_charts"
}
)
self.history = []
defask(self, question: str, show_code: bool = True) -> any:
"""
Ask a question with rich Jupyter output.
Args:
question: Natural language question
show_code: Whether to display generated code
Returns:
Query result
"""# Display question
display(Markdown(f"**Question:** {question}"))
# Execute query
result = self.smart_df.chat(question)
# Store in historyself.history.append({
"question": question,
"result": result,
"code": self.smart_df.last_code_generated
})
# Display result
display(Markdown("**Result:**"))
ifisinstance(result, pd.DataFrame):
display(result)
else:
display(result)
# Optionally show codeif show_code:
display(Markdown("**Generated Code:**"))
display(Markdown(f"```python\n{self.smart_df.last_code_generated}\n```"))
return result
defexplain(self, question: str) -> str:
"""
Ask a question and get a detailed explanation.
Args:
question: Natural language question
Returns:
Explanation string
"""# First get the result
result = self.smart_df.chat(question)
# Then ask for explanation
explanation = self.smart_df.chat(
f"Explain in detail how you calculated: {question}"
)
display(Markdown(f"**Question:** {question}"))
display(Markdown(f"**Result:** {result}"))
display(Markdown(f"**Explanation:** {explanation}"))
return explanation
defcompare(self, questions: list) -> pd.DataFrame:
"""
Run multiple queries and compare results.
Args:
questions: List of questions to compare
Returns:
DataFrame with questions and results
"""
results = []
for q in questions:
result = self.smart_df.chat(q)
results.append({
"Question": q,
"Result": result
})
comparison_df = pd.DataFrame(results)
display(comparison_df)
return comparison_df
defshow_history(self) -> None:
"""Display query history."""for i, item inenumerate(self.history, 1):
display(Markdown(f"### Query {i}"))
display(Markdown(f"**Q:** {item['question']}"))
display(Markdown(f"**A:** {item['result']}"))
display(Markdown("---"))
# Usage in Jupyter:# jpa = JupyterPandasAI(my_dataframe)# jpa.ask("What is the average sales by region?")# jpa.explain("Why did Q3 have lower sales?")# jpa.compare(["Total sales", "Average sales", "Max sales"])
Best Practices
1. Query Optimization
"""Best practices for optimizing PandasAI queries."""# DO: Be specific in your questions
good_queries = [
"What is the total revenue for Q1 2024?", # Specific time frame"Show the top 5 products by units sold", # Clear limit"Calculate the average order value by customer segment"# Clear metric
]
# DON'T: Ask vague questions
bad_queries = [
"Tell me about the data", # Too vague"What's interesting?", # No clear objective"Analyze everything"# No focus
]
# DO: Break complex questions into stepsdefmulti_step_analysis(smart_df, questions):
"""Execute analysis in logical steps."""
results = {}
for i, q inenumerate(questions, 1):
print(f"Step {i}: {q}")
results[f"step_{i}"] = smart_df.chat(q)
return results
# Example multi-step analysis
steps = [
"First, what are the column names and their data types?",
"What is the date range of the data?",
"What is the total revenue by month?",
"Which month had the highest growth rate?"
]
2. Error Handling
"""Robust error handling for PandasAI operations."""from typing importOptional, Tupleimport logging
logger = logging.getLogger(__name__)
defsafe_query(
smart_df: SmartDataframe,
question: str,
max_retries: int = 3) -> Tuple[Optional[any], Optional[str]]:
"""
Execute query with error handling and retries.
Args:
smart_df: SmartDataframe instance
question: Natural language question
max_retries: Maximum retry attempts
Returns:
Tuple of (result, error_message)
"""for attempt inrange(max_retries):
try:
result = smart_df.chat(question)
return result, Noneexcept Exception as e:
logger.warning(f"Query attempt {attempt + 1} failed: {e}")
if attempt == max_retries - 1:
returnNone, str(e)
returnNone, "Max retries exceeded"defvalidate_result(result: any, expected_type: type = None) -> bool:
"""Validate query result."""if result isNone:
returnFalseif expected_type andnotisinstance(result, expected_type):
returnFalsereturnTrue
3. Caching Strategy
"""Implement caching for repeated queries."""from functools import lru_cache
import hashlib
classCachedPandasAI:
"""PandasAI wrapper with query caching."""def__init__(self, df: pd.DataFrame):
self.df = df
self.smart_df = SmartDataframe(df, config={"enable_cache": True})
self._result_cache = {}
def_get_cache_key(self, question: str) -> str:
"""Generate cache key for question."""
df_hash = hashlib.md5(
pd.util.hash_pandas_object(self.df).values.tobytes()
).hexdigest()[:8]
q_hash = hashlib.md5(question.encode()).hexdigest()[:8]
returnf"{df_hash}_{q_hash}"defquery(self, question: str, use_cache: bool = True) -> any:
"""Query with caching."""
cache_key = self._get_cache_key(question)
if use_cache and cache_key inself._result_cache:
returnself._result_cache[cache_key]
result = self.smart_df.chat(question)
self._result_cache[cache_key] = result
return result
defclear_cache(self):
"""Clear the result cache."""self._result_cache.clear()
# Problem: API key not found or invalid# Solution: Verify environment variable or explicit keyimport os
defverify_api_key():
"""Verify API key is configured."""
key = os.getenv("OPENAI_API_KEY")
ifnot key:
print("Error: OPENAI_API_KEY not set")
print("Set it with: export OPENAI_API_KEY='your-key'")
returnFalseifnot key.startswith("sk-"):
print("Warning: API key format may be incorrect")
returnFalsereturnTrue# Alternative: Pass key explicitlyfrom pandasai.llm import OpenAI
llm = OpenAI(api_key="your-api-key-here")
2. Memory Issues with Large DataFrames
# Problem: Out of memory with large DataFrames# Solution: Sample or aggregate before queryingdefcreate_efficient_smart_df(df: pd.DataFrame, max_rows: int = 10000):
"""Create SmartDataframe with large data handling."""iflen(df) > max_rows:
print(f"Sampling {max_rows} rows from {len(df)} total rows")
sample_df = df.sample(n=max_rows, random_state=42)
else:
sample_df = df
return SmartDataframe(
sample_df,
config={
"llm": OpenAI(model="gpt-4"),
"verbose": False,
"max_retries": 3
}
)
3. Incorrect Results
# Problem: PandasAI returns incorrect or unexpected results# Solution: Verify with explicit code and add contextdefverify_result(smart_df, question, expected_type=None):
"""Verify PandasAI result against pandas."""# Get PandasAI result
ai_result = smart_df.chat(question)
# Get generated code
code = smart_df.last_code_generated
print(f"Question: {question}")
print(f"AI Result: {ai_result}")
print(f"Generated Code:\n{code}")
# Manual verification promptprint("\nPlease verify the generated code produces the expected result.")
return ai_result, code
4. Chart Generation Failures
# Problem: Charts not generating or displaying# Solution: Check dependencies and configurationdefdiagnose_chart_issues():
"""Diagnose chart generation issues."""
issues = []
# Check matplotlibtry:
import matplotlib
print(f"Matplotlib version: {matplotlib.__version__}")
except ImportError:
issues.append("matplotlib not installed")
# Check backendtry:
import matplotlib.pyplot as plt
backend = plt.get_backend()
print(f"Matplotlib backend: {backend}")
except Exception as e:
issues.append(f"Backend issue: {e}")
# Check save pathfrom pathlib import Path
charts_path = Path("./charts")
ifnot charts_path.exists():
print("Creating charts directory...")
charts_path.mkdir(parents=True)
if issues:
print("Issues found:")
for issue in issues:
print(f" - {issue}")
else:
print("No issues detected")
return issues
