AI Data Analysis Workflow

Overview

The AI Data Analysis workflow automates the process of exploring, analyzing, and visualizing datasets. It leverages AI agents to understand data structure, perform statistical analysis, generate insights, and create visualizations — all through natural language interactions.

Difficulty: Hard

Tools Required

• LangGraph - For orchestrating the analysis pipeline
• Python - Data processing with pandas, numpy, scipy
• SQLite MCP or PostgreSQL MCP - Database access
• Filesystem MCP - CSV, Excel, JSON file access
• Matplotlib/Seaborn - Data visualization
• Jupyter - Interactive analysis environment
• Slack/Email - Report distribution

Workflow Steps

Step 1: Data Ingestion

The Data Ingestion Agent loads and validates data:

• Accept multiple data formats (CSV, Excel, JSON, SQL)
• Validate data integrity and completeness
• Detect data types and structures
• Report data quality issues

def ingest_data(source, format="auto"):
    # Detect format if auto
    if format == "auto":
        format = detect_format(source)
    
    # Load data
    if format == "csv":
        df = pd.read_csv(source)
    elif format == "excel":
        df = pd.read_excel(source)
    elif format == "json":
        df = pd.read_json(source)
    elif format == "sql":
        df = pd.read_sql(source["query"], source["connection"])
    
    # Validate
    validation = validate_dataframe(df)
    
    return {
        "dataframe": df,
        "shape": df.shape,
        "columns": list(df.columns),
        "dtypes": df.dtypes.to_dict(),
        "validation": validation
    }

Step 2: Data Profiling

The Profiler Agent understands the data:

• Generate statistical summaries
• Identify distributions and patterns
• Detect outliers and anomalies
• Analyze correlations

def profile_data(df):
    profile = {
        "summary": df.describe(include="all"),
        "missing_values": df.isnull().sum().to_dict(),
        "unique_values": df.nunique().to_dict(),
        "distributions": {},
        "outliers": {},
        "correlations": df.corr().to_dict()
    }
    
    # Analyze distributions
    for col in df.select_dtypes(include=["number"]).columns:
        profile["distributions"][col] = {
            "mean": df[col].mean(),
            "median": df[col].median(),
            "std": df[col].std(),
            "skewness": df[col].skew(),
            "kurtosis": df[col].kurtosis()
        }
    
    # Detect outliers using IQR method
    for col in df.select_dtypes(include=["number"]).columns:
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        outliers = df[(df[col] < Q1 - 1.5*IQR) | (df[col] > Q3 + 1.5*IQR)]
        profile["outliers"][col] = len(outliers)
    
    return profile

Step 3: Question Interpretation

The Interpreter Agent understands analysis requests:

• Parse natural language questions
• Identify analysis type needed
• Map questions to data columns
• Determine appropriate methods

def interpret_question(question, data_profile):
    prompt = f"""
    Data Profile:
    {data_profile}
    
    Question: {question}
    
    Analyze the question and determine:
    1. Analysis type (descriptive, diagnostic, predictive, prescriptive)
    2. Relevant columns
    3. Suggested methods
    4. Expected output format
    """
    
    interpretation = llm.parse(prompt)
    
    # Validate columns exist
    for col in interpretation["columns"]:
        if col not in data_profile["columns"]:
            interpretation["warnings"].append(f"Column '{col}' not found in data")
    
    return interpretation

Step 4: Analysis Execution

The Analyst Agent performs the analysis:

• Execute statistical tests
• Generate visualizations
• Calculate metrics and KPIs
• Identify patterns and trends

def execute_analysis(df, interpretation):
    results = {}
    
    if interpretation["type"] == "descriptive":
        # Summary statistics
        results["statistics"] = df[interpretation["columns"]].describe()
        
    elif interpretation["type"] == "diagnostic":
        # Root cause analysis
        if "group_by" in interpretation:
            results["grouped"] = df.groupby(interpretation["group_by"]).agg(interpretation["aggregations"])
        
        if "correlation" in interpretation:
            results["correlation"] = df[interpretation["columns"]].corr()
        
    elif interpretation["type"] == "predictive":
        # Predictive modeling
        from sklearn.model_selection import train_test_split
        from sklearn.ensemble import RandomForestRegressor
        
        X = df[interpretation["features"]]
        y = df[interpretation["target"]]
        
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
        model = RandomForestRegressor()
        model.fit(X_train, y_train)
        
        results["model"] = {
            "type": "RandomForestRegressor",
            "score": model.score(X_test, y_test),
            "feature_importance": dict(zip(interpretation["features"], model.feature_importances_))
        }
    
    return results

Step 5: Visualization

The Visualization Agent creates charts:

• Select appropriate chart types
• Generate publication-quality visuals
• Add annotations and insights
• Create interactive dashboards

def create_visualizations(df, analysis_results, interpretation):
    visualizations = []
    
    if interpretation["type"] == "descriptive":
        # Distribution plots
        for col in interpretation["columns"]:
            if df[col].dtype in ["int64", "float64"]:
                fig, ax = plt.subplots()
                df[col].hist(ax=ax, bins=30)
                ax.set_title(f"Distribution of {col}")
                ax.set_xlabel(col)
                ax.set_ylabel("Frequency")
                visualizations.append({"type": "histogram", "column": col, "figure": fig})
            
            else:
                fig, ax = plt.subplots()
                df[col].value_counts().head(10).plot(kind="bar", ax=ax)
                ax.set_title(f"Top 10 values of {col}")
                ax.set_xlabel(col)
                ax.set_ylabel("Count")
                visualizations.append({"type": "bar", "column": col, "figure": fig})
    
    elif interpretation["type"] == "diagnostic":
        # Correlation heatmap
        if "correlation" in analysis_results:
            fig, ax = plt.subplots(figsize=(10, 8))
            corr_matrix = df[interpretation["columns"]].corr()
            sns.heatmap(corr_matrix, annot=True, cmap="coolwarm", center=0, ax=ax)
            ax.set_title("Correlation Heatmap")
            visualizations.append({"type": "heatmap", "figure": fig})
        
        # Group comparison
        if "grouped" in analysis_results:
            fig, ax = plt.subplots()
            analysis_results["grouped"].plot(kind="bar", ax=ax)
            ax.set_title(f"{interpretation['group_by']} Comparison")
            visualizations.append({"type": "grouped_bar", "figure": fig})
    
    return visualizations

Step 6: Insight Generation

The Insight Agent extracts meaningful findings:

• Identify significant patterns
• Generate narrative explanations
• Highlight anomalies and opportunities
• Provide actionable recommendations

def generate_insights(analysis_results, visualizations, data_profile):
    insights = []
    
    # Statistical insights
    for col, stats in data_profile["distributions"].items():
        if abs(stats["skewness"]) > 1:
            direction = "right-skewed" if stats["skewness"] > 0 else "left-skewed"
            insights.append({
                "type": "distribution",
                "column": col,
                "finding": f"{col} is {direction} (skewness: {stats['skewness']:.2f})",
                "recommendation": "Consider transformation for modeling"
            })
    
    # Correlation insights
    if "correlation" in analysis_results:
        corr = analysis_results["correlation"]
        for i, col1 in enumerate(corr.columns):
            for col2 in corr.columns[i+1:]:
                value = corr.loc[col1, col2]
                if abs(value) > 0.7:
                    strength = "strong" if abs(value) > 0.8 else "moderate"
                    direction = "positive" if value > 0 else "negative"
                    insights.append({
                        "type": "correlation",
                        "columns": [col1, col2],
                        "finding": f"{col1} and {col2} have {strength} {direction} correlation ({value:.2f})",
                        "recommendation": "Consider for feature engineering"
                    })
    
    # Model insights
    if "model" in analysis_results:
        model = analysis_results["model"]
        insights.append({
            "type": "prediction",
            "finding": f"Model achieved {model['score']:.2%} accuracy",
            "recommendation": f"Top features: {list(model['feature_importance'].keys())[:3]}"
        })
    
    return insights

Step 7: Report Generation

The Reporter Agent creates the final output:

• Compile analysis summary
• Include visualizations
• Add methodology notes
• Generate downloadable formats

def generate_report(data_profile, analysis_results, visualizations, insights):
    report = {
        "title": "Data Analysis Report",
        "executive_summary": generate_summary(insights),
        "data_overview": {
            "shape": data_profile["shape"],
            "columns": data_profile["columns"],
            "missing_values": data_profile["missing_values"]
        },
        "key_findings": insights,
        "visualizations": visualizations,
        "methodology": describe_methods(analysis_results),
        "recommendations": extract_recommendations(insights)
    }
    
    # Generate formats
    formats = {
        "markdown": format_markdown(report),
        "html": format_html(report, visualizations),
        "pdf": generate_pdf(report, visualizations),
        "jupyter": generate_notebook(report, visualizations)
    }
    
    return formats

Step 8: Distribution

The Distributor Agent shares results:

• Send email reports
• Post to Slack/Teams
• Create dashboard links
• Archive for future reference

def distribute_report(report, channels):
    distribution = {}
    
    for channel in channels:
        if channel == "email":
            sendgrid.send(
                to=report["recipients"],
                subject=f"Data Analysis: {report['title']}",
                html=report["formats"]["html"]
            )
        
        elif channel == "slack":
            slack.post(
                channel="#data-analytics",
                text=f"📊 New Analysis Report: {report['title']}\n{report['executive_summary']}"
            )
        
        elif channel == "dashboard":
            # Update dashboard
            dashboard.update(report)
    
    return distribution

Example Usage

Scenario: Sales Data Analysis

Analyst: "Analyze our Q1 sales data and find trends"

AI Workflow:
1. ✅ Ingestion: Loads sales.csv (50,000 rows, 15 columns)
2. ✅ Profiling: Identifies 3 columns with missing values, detects outliers
3. ✅ Interpretation: Understands request for trend analysis
4. ✅ Analysis: Calculates monthly trends, regional comparisons, product performance
5. ✅ Visualization: Creates 6 charts (line, bar, heatmap, pie)
6. ✅ Insights: Finds 15% growth in Region A, decline in Product X
7. ✅ Report: Generates HTML, PDF, and Jupyter notebook
8. ✅ Distribution: Emails to leadership, posts summary to Slack

Scenario: Customer Churn Analysis

Analyst: "What factors contribute to customer churn?"

AI Workflow:
1. ✅ Ingestion: Loads customer_data.csv with churn labels
2. ✅ Profiling: Identifies 20 features, 5% churn rate
3. ✅ Interpretation: Understands request for predictive analysis
4. ✅ Analysis: Builds classification model (85% accuracy)
5. ✅ Visualization: Creates feature importance chart, confusion matrix
6. ✅ Insights: Identifies top 5 churn predictors
7. ✅ Report: Generates actionable recommendations
8. ✅ Distribution: Shares with product and marketing teams

LangGraph Workflow Definition

from langgraph.graph import StateGraph, END

class AnalysisState(TypedDict):
    source: str
    data_profile: dict
    interpretation: dict
    analysis_results: dict
    visualizations: list
    insights: list
    report: dict
    distribution: dict

builder = StateGraph(AnalysisState)

builder.add_node("ingestion", ingestion_node)
builder.add_node("profiling", profiling_node)
builder.add_node("interpretation", interpretation_node)
builder.add_node("analysis", analysis_node)
builder.add_node("visualization", visualization_node)
builder.add_node("insights", insights_node)
builder.add_node("reporting", reporting_node)
builder.add_node("distribution", distribution_node)

builder.add_edge("ingestion", "profiling")
builder.add_edge("profiling", "interpretation")
builder.add_edge("interpretation", "analysis")
builder.add_edge("analysis", "visualization")
builder.add_edge("visualization", "insights")
builder.add_edge("insights", "reporting")
builder.add_edge("reporting", "distribution")
builder.set_entry_point("ingestion")
builder.set_finish_point("distribution")

workflow = builder.compile()

Pros

• ✅ Natural Language: Ask questions in plain English
• ✅ Comprehensive Analysis: Statistical + ML capabilities
• ✅ Automated Visualizations: Professional charts automatically
• ✅ Actionable Insights: Recommendations included
• ✅ Multiple Formats: HTML, PDF, Jupyter, Markdown
• ✅ Reproducible: Full audit trail of analysis

Cons

• ❌ Complex Setup: Multiple tools and dependencies
• ❌ Compute Resources: Large datasets need significant memory
• ❌ Cost: Multiple AI calls for interpretation
• ❌ Accuracy: AI may misinterpret complex questions
• ❌ Domain Knowledge: May miss industry-specific nuances

When to Use

Choose this workflow when:

• You need to analyze datasets regularly
• Stakeholders ask ad-hoc analysis questions
• You want to democratize data access
• You need consistent analysis methodology

Consider alternatives when:

• You have simple one-off analysis (use pandas directly)
• You need real-time streaming analysis
• You have highly specialized domain requirements

Resources

• LangGraph Docs: https://langchain-ai.github.io/langgraph/
• Pandas Documentation: https://pandas.pydata.org/docs/
• Scikit-Learn: https://scikit-learn.org/
• Matplotlib: https://matplotlib.org/
• Seaborn: https://seaborn.pydata.org/

---

Last updated: May 2026