Data Analysis Swarm

from swarms import Agent, ConcurrentWorkflow, SequentialWorkflow
import os

# Configure your LLM
api_key = os.getenv("OPENAI_API_KEY")

# Define data analysis agents
data_collector = Agent(
    agent_name="Data-Collector",
    system_prompt="""
    You are a data collection specialist.
    Your role is to:
    - Identify relevant data sources for the analysis
    - Extract and gather required datasets
    - Document data provenance and metadata
    - Flag data quality issues
    - Structure data for analysis
    
    Provide clean, well-documented datasets ready for analysis.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
    dynamic_temperature_enabled=True,
)

data_cleaner = Agent(
    agent_name="Data-Cleaner",
    system_prompt="""
    You are a data cleaning and validation expert.
    Your role is to:
    - Identify missing, duplicate, or invalid data
    - Apply appropriate cleaning strategies
    - Normalize and standardize data formats
    - Document all transformations
    - Validate data integrity
    
    Ensure data quality before analysis begins.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
    dynamic_temperature_enabled=True,
)

statistical_analyst = Agent(
    agent_name="Statistical-Analyst",
    system_prompt="""
    You are a statistical analyst specializing in quantitative analysis.
    Your role is to:
    - Calculate descriptive statistics (mean, median, variance, etc.)
    - Perform hypothesis testing and significance tests
    - Identify correlations and relationships
    - Assess statistical confidence and margins of error
    - Interpret statistical findings in business context
    
    Provide rigorous statistical insights.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
    dynamic_temperature_enabled=True,
)

trend_analyst = Agent(
    agent_name="Trend-Analyst",
    system_prompt="""
    You are a trend analysis expert.
    Your role is to:
    - Identify patterns and trends over time
    - Perform time-series analysis
    - Detect seasonality and cyclical patterns
    - Project future trends based on historical data
    - Assess trend strength and reliability
    
    Uncover temporal patterns and forecast future states.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
    dynamic_temperature_enabled=True,
)

anomaly_detector = Agent(
    agent_name="Anomaly-Detector",
    system_prompt="""
    You are an anomaly detection specialist.
    Your role is to:
    - Identify outliers and unusual patterns
    - Distinguish between noise and significant anomalies
    - Investigate potential causes of anomalies
    - Assess impact and urgency of anomalies
    - Recommend investigation priorities
    
    Find the unexpected insights hidden in data.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
    dynamic_temperature_enabled=True,
)

visualization_expert = Agent(
    agent_name="Visualization-Expert",
    system_prompt="""
    You are a data visualization expert.
    Your role is to:
    - Design effective charts and graphs for findings
    - Choose appropriate visualization types for data
    - Create visual narratives that tell the story
    - Ensure visualizations are accessible and clear
    - Provide specifications for implementation
    
    Transform complex data into compelling visuals.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
    dynamic_temperature_enabled=True,
)

report_synthesizer = Agent(
    agent_name="Report-Synthesizer",
    system_prompt="""
    You are a data analyst who synthesizes findings into executive reports.
    Your role is to:
    - Combine insights from all analysis streams
    - Identify key findings and actionable insights
    - Structure information for executive audience
    - Highlight business implications
    - Provide clear recommendations
    
    Create comprehensive, actionable analysis reports.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
    dynamic_temperature_enabled=True,
)

# Create the data analysis workflow
# First: Sequential data preparation
prep_workflow = SequentialWorkflow(
    name="Data-Preparation",
    agents=[data_collector, data_cleaner],
    max_loops=1,
    output_type="final",  # return a string so it can feed the next stage's task=
)

# Second: Concurrent analysis of different aspects
analysis_workflow = ConcurrentWorkflow(
    name="Parallel-Analysis",
    agents=[statistical_analyst, trend_analyst, anomaly_detector],
    max_loops=1,
    output_type="str-all-except-first",  # every analyst's answer, in agent order, as one string
)

# Third: Sequential visualization and reporting
reporting_workflow = SequentialWorkflow(
    name="Reporting-Pipeline",
    agents=[visualization_expert, report_synthesizer],
    max_loops=1,
    output_type="final",  # return the final report as a string
)

# Execute complete data analysis
if __name__ == "__main__":
    analysis_request = """
    Analyze the following dataset:
    
    Dataset: Quarterly sales data for the past 3 years
    Columns: Date, Region, Product, Revenue, Units, Customer_Segment
    
    Analysis objectives:
    1. Identify revenue trends and growth patterns
    2. Detect any anomalies or unusual performance
    3. Compare performance across regions and segments
    4. Forecast next quarter's performance
    5. Recommend strategic actions based on findings
    
    Provide comprehensive analysis with visualizations and executive summary.
    """
    
    # Stage 1: Prepare the data
    print("Stage 1: Data preparation...")
    cleaned_data = prep_workflow.run(analysis_request)
    
    # Stage 2: Run parallel analyses
    print("Stage 2: Running parallel analyses...")
    analysis_results = analysis_workflow.run(cleaned_data)
    
    # Stage 3: Visualize and synthesize report
    print("Stage 3: Creating visualizations and final report...")
    final_report = reporting_workflow.run(analysis_results)
    
    # Save the final report
    with open("sales_analysis_report.md", "w") as f:
        f.write(final_report)
    
    print("Analysis complete! Report saved to sales_analysis_report.md")

marketing_analyst = Agent(
    agent_name="Marketing-Analyst",
    system_prompt="""
    You analyze marketing-specific metrics:
    - Customer acquisition cost (CAC)
    - Lifetime value (LTV)
    - Conversion rates and funnel analysis
    - Campaign performance and ROI
    - Channel attribution
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
)

# Add to parallel analysis
analysis_workflow = ConcurrentWorkflow(
    agents=[statistical_analyst, trend_analyst, anomaly_detector, marketing_analyst],
    output_type="str-all-except-first",
)

ml_modeler = Agent(
    agent_name="ML-Modeler",
    system_prompt="""
    You are a machine learning specialist.
    Your role is to:
    - Recommend appropriate ML models for the data
    - Define features and target variables
    - Assess model performance and accuracy
    - Interpret model predictions
    - Identify key predictive factors
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
)

comparative_analyst = Agent(
    agent_name="Comparative-Analyst",
    system_prompt="""
    You perform comparative analysis:
    - Benchmark against industry standards
    - Compare performance across segments
    - Identify best and worst performers
    - Analyze competitive positioning
    - Highlight performance gaps
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
)

from swarms import Agent

real_time_monitor = Agent(
    agent_name="Real-Time-Monitor",
    system_prompt="Monitor streaming data for immediate insights...",
    max_loops=10,  # stopping_condition is checked after each loop; max_loops="auto" ignores it
    stopping_condition=lambda response: "critical_threshold_reached" in response.lower(),
)

data_cleaner = Agent(
    agent_name="Data-Cleaner",
    system_prompt="""
    Clean data and provide quality score:
    
    Quality Dimensions:
    - Completeness: % of non-null values
    - Accuracy: % of valid values
    - Consistency: % of standardized formats
    - Timeliness: Age of data
    
    Overall Quality Score: (0-100)
    
    Flag if quality score < 80.
    """,
    model_name="claude-sonnet-4-6",
    max_loops=1,
)

from swarms import Agent, batched_grid_agent_execution

chunks = ["<rows 1-25,000>", "<rows 25,001-50,000>", "<rows 50,001-75,000>", "<rows 75,001-100,000>"]

chunk_analysts = [
    Agent(agent_name=f"Chunk-Analyst-{i}", model_name="gpt-5.4-mini", max_loops=1)
    for i in range(len(chunks))
]

# Results come back in agent order; a failed agent's slot holds its exception.
chunk_results = batched_grid_agent_execution(
    agents=chunk_analysts,
    tasks=[f"Analyze this chunk of the sales data:\n{chunk}" for chunk in chunks],
)

incremental_analyst = Agent(
    agent_name="Incremental-Analyst",
    system_prompt="""
    Perform incremental analysis on new data:
    - Compare to previous analysis
    - Identify changes and deltas
    - Update trends and forecasts
    - Flag significant changes
    """,
    max_loops=1,
)

statistical_analyst = Agent(
    agent_name="Statistical-Analyst",
    persistent_memory=True,   # off by default; opt in explicitly
    context_compression=True,
    context_length=32000,
    # ... other params
)

# Sales Analysis Report - Q4 2025

## Executive Summary
- Revenue up 23% YoY, driven by Enterprise segment
- Anomaly detected: 40% spike in returns in APAC region (investigate)
- Forecast: Q1 2026 revenue projected at $4.2M (±8%)

## Statistical Analysis
[Detailed statistics...]

## Trend Analysis
[Trend charts and forecasts...]

## Anomalies & Alerts
[Unusual patterns requiring attention...]

## Recommendations
1. Investigate APAC returns spike
2. Increase investment in Enterprise sales
3. ...