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,
)
# Second: Concurrent analysis of different aspects
analysis_workflow = ConcurrentWorkflow(
name="Parallel-Analysis",
agents=[statistical_analyst, trend_analyst, anomaly_detector],
max_loops=1,
)
# Third: Sequential visualization and reporting
reporting_workflow = SequentialWorkflow(
name="Reporting-Pipeline",
agents=[visualization_expert, report_synthesizer],
max_loops=1,
)
# 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")
