AI Data Analyst Examples: 6 Real Setups

from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import create_sql_agent
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit

# Connect to read-only database
db = SQLDatabase.from_uri(
    "postgresql+psycopg2://readonly_user:password@localhost/analytics",
    include_tables=["orders", "customers", "products", "revenue_daily"],
    sample_rows_in_table_info=2
)

llm = ChatOpenAI(model="gpt-4o", temperature=0)
toolkit = SQLDatabaseToolkit(db=db, llm=llm)

agent = create_sql_agent(
    llm=llm,
    toolkit=toolkit,
    agent_type="openai-tools",
    verbose=True,
    prefix="""You are a data analyst assistant. When answering questions:
    1. Always check the table schema before writing SQL
    2. Write efficient queries (use LIMIT 1000 unless aggregating)
    3. After getting results, explain what they mean in business terms
    4. If results are surprising, note potential data quality issues
    5. Never modify data — SELECT only""",
    max_iterations=8
)

# Business user queries
questions = [
    "What were our top 5 products by revenue last month?",
    "How did customer acquisition cost change quarter over quarter this year?",
    "Which customer segments have the highest 90-day retention rate?"
]

for question in questions:
    print(f"\nQuestion: {question}")
    result = agent.invoke({"input": question})
    print(f"Answer: {result['output']}")

import pandas as pd
from openai import OpenAI
from langchain_experimental.tools import PythonREPLTool
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub
from langchain_openai import ChatOpenAI

client = OpenAI()

def determine_optimal_charts(df_info: dict, analysis_goal: str) -> list[dict]:
    """Determine the best chart types for the data and goal."""
    response = client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[{
            "role": "user",
            "content": f"""Given this dataset and analysis goal, recommend 3-4 charts.
            Dataset: {df_info}
            Goal: {analysis_goal}
            Return JSON: {{"charts": [{{"type": str, "x_column": str, "y_column": str,
            "title": str, "rationale": str, "insight": str}}]}}"""
        }]
    )
    import json
    return json.loads(response.choices[0].message.content)["charts"]

def generate_chart_code(df_info: dict, chart_spec: dict) -> str:
    """Generate Python code to create a specific chart."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""Write Python code to create this chart using matplotlib or plotly.
            Chart spec: {chart_spec}
            Dataset info: {df_info}

            Code requirements:
            - Load data from 'df' variable (already exists)
            - Use plotly for interactive charts if time-series, matplotlib for static
            - Include title, axis labels, and source annotation
            - Save to './charts/{chart_spec["type"]}_{chart_spec["title"][:20]}.png'
            - Print: "Chart saved: [filename]"
            Return only the Python code."""
        }]
    )
    return response.choices[0].message.content

# Load data and generate charts
df = pd.read_csv("sales_data.csv")
df_info = {
    "columns": df.dtypes.to_dict(),
    "shape": df.shape,
    "sample": df.head(3).to_dict(),
    "date_range": f"{df['date'].min()} to {df['date'].max()}" if "date" in df.columns else "N/A"
}

chart_specs = determine_optimal_charts(df_info, "Understand revenue trends and product performance")
for spec in chart_specs:
    code = generate_chart_code(df_info, spec)
    print(f"Generating: {spec['title']}")
    print(f"Insight: {spec['insight']}")
    exec(code)  # In production, use sandboxed execution

import pandas as pd
import numpy as np
from scipy import stats
from anthropic import Anthropic

client = Anthropic()

def detect_anomalies(df: pd.DataFrame, column: str, threshold_z: float = 2.5) -> list[dict]:
    """Detect statistical anomalies in a time series column."""
    # Calculate rolling mean and standard deviation
    df = df.copy()
    df["rolling_mean"] = df[column].rolling(window=14, min_periods=7).mean()
    df["rolling_std"] = df[column].rolling(window=14, min_periods=7).std()
    df["z_score"] = (df[column] - df["rolling_mean"]) / df["rolling_std"]

    anomalies = df[df["z_score"].abs() > threshold_z].copy()
    anomalies["direction"] = anomalies["z_score"].apply(lambda z: "spike" if z > 0 else "drop")
    anomalies["deviation_pct"] = ((anomalies[column] - anomalies["rolling_mean"]) / anomalies["rolling_mean"] * 100).round(1)

    return anomalies[["date", column, "rolling_mean", "z_score", "direction", "deviation_pct"]].to_dict("records")

def explain_anomaly(anomaly: dict, metric_name: str, context: str) -> str:
    """Use AI to explain a detected anomaly and suggest investigation steps."""
    response = client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Explain this metric anomaly for a business analyst:

            Metric: {metric_name}
            Date: {anomaly["date"]}
            Anomaly: {anomaly["direction"]} of {abs(anomaly["deviation_pct"])}% vs 14-day average
            Actual value: {anomaly[metric_name.lower().replace(" ", "_")]}
            Expected range: ~{anomaly["rolling_mean"]:.0f}

            Business context: {context}

            Provide:
            1. 3 most likely explanations (ranked by probability)
            2. 3 specific data sources to check to confirm root cause
            3. Recommended immediate actions
            Keep it concise — this goes to a Slack alert."""
        }]
    )
    return response.content[0].text

# Load metrics
df = pd.read_sql(
    "SELECT date, daily_revenue FROM revenue_daily WHERE date > current_date - 90 ORDER BY date",
    con=engine
)

anomalies = detect_anomalies(df, "daily_revenue", threshold_z=2.5)
context = "SaaS platform, monthly subscriptions, B2B customers, average deal $500/month"

for anomaly in anomalies[-5:]:  # Alert on last 5 anomalies
    explanation = explain_anomaly(anomaly, "daily_revenue", context)
    print(f"\n🚨 Anomaly detected: {anomaly['date']}")
    print(f"Revenue {anomaly['direction']}: {anomaly['deviation_pct']}% vs baseline")
    print(explanation)
    # post_to_slack(f"*Revenue Anomaly Detected*\n{explanation}")

import asyncio
from anthropic import Anthropic
import pandas as pd

client = Anthropic()

async def fetch_weekly_metrics() -> dict:
    """Fetch all metrics needed for weekly report in parallel."""

    async def get_revenue_metrics():
        # Query your analytics DB
        return {
            "total_revenue": 142500,
            "vs_last_week_pct": 8.3,
            "vs_last_year_pct": 34.2,
            "top_product": "Enterprise Plan",
            "top_product_revenue": 85000
        }

    async def get_customer_metrics():
        return {
            "new_customers": 47,
            "churned_customers": 12,
            "net_new": 35,
            "mrr_change": 8750,
            "nps_score": 62,
            "active_users_7d": 1847
        }

    async def get_product_metrics():
        return {
            "feature_adoption": {"AI Chat": 0.73, "API": 0.51, "Webhooks": 0.38},
            "avg_sessions_per_user": 4.2,
            "p95_api_latency_ms": 234,
            "error_rate_pct": 0.12
        }

    revenue, customers, product = await asyncio.gather(
        get_revenue_metrics(), get_customer_metrics(), get_product_metrics()
    )
    return {"revenue": revenue, "customers": customers, "product": product}

def generate_report_narrative(metrics: dict, week_ending: str) -> str:
    """Generate narrative commentary on weekly metrics."""
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": f"""Write a weekly business performance report for the week ending {week_ending}.

            Metrics: {metrics}

            Structure:
            # Weekly Performance Report — Week Ending {week_ending}

            ## Executive Summary (3 sentences max)
            ## Revenue Performance
            ## Customer Growth
            ## Product Health
            ## Key Concerns (if any)
            ## Next Week Focus

            Tone: Professional but direct. Highlight notable trends and concerns.
            Use specific numbers. Flag anything that needs leadership attention."""
        }]
    )
    return response.content[0].text

import asyncio
metrics = asyncio.run(fetch_weekly_metrics())
report = generate_report_narrative(metrics, "2026-02-28")

# Save and distribute
with open("reports/weekly_2026-02-28.md", "w") as f:
    f.write(report)

print("Weekly report generated and distributed")

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import sqlite3, pandas as pd

@tool
def run_cohort_query(months_back: int = 12) -> str:
    """Generate and run cohort retention analysis SQL."""
    sql = f"""
    WITH cohorts AS (
        SELECT
            customer_id,
            DATE_TRUNC('month', first_order_date) as cohort_month,
            DATE_TRUNC('month', order_date) as order_month
        FROM orders o
        JOIN (
            SELECT customer_id, MIN(created_at) as first_order_date
            FROM orders GROUP BY customer_id
        ) first_orders USING (customer_id)
        WHERE o.order_date >= CURRENT_DATE - INTERVAL '{months_back} months'
    )
    SELECT
        cohort_month,
        EXTRACT(MONTH FROM AGE(order_month, cohort_month)) AS months_since_first,
        COUNT(DISTINCT customer_id) AS retained_customers
    FROM cohorts
    GROUP BY 1, 2
    ORDER BY 1, 2
    """
    conn = sqlite3.connect("analytics.db")
    df = pd.read_sql(sql, conn)
    return df.to_csv(index=False)

@tool
def analyze_retention_patterns(cohort_csv: str) -> str:
    """This tool signals the agent to analyze the cohort data provided."""
    return f"Ready to analyze cohort retention patterns from: {cohort_csv[:200]}..."

from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [run_cohort_query, analyze_retention_patterns]
agent = create_react_agent(llm=llm, tools=tools, prompt=hub.pull("hwchase17/react"))

executor = AgentExecutor(agent=agent, tools=tools, max_iterations=6, verbose=True)

result = executor.invoke({
    "input": """Analyze customer retention by cohort for the past 12 months.
    Identify: 1) Average retention at M1, M3, M6, M12
    2) Whether retention is improving or declining across recent cohorts
    3) The cohort with the best M3 retention and what was different about that period
    4) Specific recommendations to improve M1 retention
    Present as a structured analysis report."""
})
print(result["output"])

import asyncio
from anthropic import Anthropic
from datetime import datetime
import json

client = Anthropic()

class DashboardNarrator:
    def __init__(self, thresholds: dict):
        self.thresholds = thresholds
        self.previous_metrics = {}

    def check_thresholds(self, current: dict) -> list[dict]:
        """Identify metrics that crossed thresholds or changed significantly."""
        alerts = []
        for metric, value in current.items():
            if metric in self.thresholds:
                threshold = self.thresholds[metric]
                if value < threshold.get("min", float("-inf")):
                    alerts.append({"metric": metric, "value": value, "type": "below_min",
                                   "threshold": threshold["min"], "severity": threshold.get("severity", "medium")})
                elif value > threshold.get("max", float("inf")):
                    alerts.append({"metric": metric, "value": value, "type": "above_max",
                                   "threshold": threshold["max"], "severity": threshold.get("severity", "medium")})

            # Check for significant change vs last reading
            if metric in self.previous_metrics:
                change_pct = (value - self.previous_metrics[metric]) / self.previous_metrics[metric] * 100
                if abs(change_pct) > 20:
                    alerts.append({"metric": metric, "value": value, "change_pct": round(change_pct, 1),
                                   "type": "significant_change", "severity": "high"})
        return alerts

    def narrate_alert(self, alert: dict, all_current_metrics: dict) -> str:
        """Generate a narrative explanation for a dashboard alert."""
        response = client.messages.create(
            model="claude-3-5-haiku-20241022",
            max_tokens=400,
            messages=[{
                "role": "user",
                "content": f"""Generate a concise, business-friendly alert for this dashboard signal:

                Alert: {json.dumps(alert)}
                Current dashboard state: {json.dumps(all_current_metrics)}
                Time: {datetime.now().strftime('%Y-%m-%d %H:%M UTC')}

                Include: What happened, likely cause, immediate action needed.
                Max 4 sentences. No jargon. Written for a business executive."""
            }]
        )
        return response.content[0].text

# Configure thresholds
narrator = DashboardNarrator(thresholds={
    "api_error_rate_pct": {"max": 1.0, "severity": "critical"},
    "checkout_conversion_pct": {"min": 2.5, "severity": "high"},
    "avg_order_value_usd": {"min": 45, "severity": "medium"},
    "active_sessions": {"min": 100, "severity": "medium"}
})

# Simulate dashboard polling
current_metrics = {
    "api_error_rate_pct": 3.2,  # Above threshold!
    "checkout_conversion_pct": 3.1,
    "avg_order_value_usd": 62,
    "active_sessions": 847,
    "revenue_today_usd": 8420
}

alerts = narrator.check_thresholds(current_metrics)
for alert in alerts:
    narrative = narrator.narrate_alert(alert, current_metrics)
    print(f"[{alert['severity'].upper()}] {alert['metric']}: {narrative}")
    # Send to Slack/PagerDuty based on severity