diff --git a/docs/swarms/structs/swarm_templates.md b/docs/swarms/structs/swarm_templates.md new file mode 100644 index 00000000..8c963e24 --- /dev/null +++ b/docs/swarms/structs/swarm_templates.md @@ -0,0 +1,796 @@ +# SwarmTemplates + +A comprehensive library of pre-built, production-ready swarm templates for common use cases. Each template includes professionally designed agents with optimized prompts and proven orchestration patterns. + +## Overview + +SwarmTemplates provides instant access to enterprise-grade multi-agent systems that solve real-world business problems. Instead of building agent teams from scratch, you can leverage battle-tested templates that have been optimized for specific use cases. + +### Key Benefits + +| Benefit | Description | +|---------|-------------| +| **Instant Deployment** | Get production-ready swarms in seconds, not hours | +| **Best Practices Built-in** | Templates use proven prompts and orchestration patterns | +| **Customizable** | Easy to customize with your own models and parameters | +| **Production-Tested** | Each template has been tested and optimized for real use | +| **Comprehensive Coverage** | Templates for research, content, code, finance, and more | + +## Available Templates + +### 1. Research-Analysis-Synthesis + +**Use Case:** Academic research, market research, competitive analysis + +**Workflow:** Research Specialist → Analysis Expert → Synthesis Specialist + +**Best For:** +- Comprehensive topic investigation +- Market research reports +- Competitive analysis +- Academic literature reviews + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.research_analysis_synthesis( + model_name="gpt-4o-mini" +) + +result = swarm.run("Analyze the impact of AI on healthcare") +print(result) +``` + +--- + +### 2. Content Creation Pipeline + +**Use Case:** Blog posts, articles, marketing content, documentation + +**Workflow:** Content Strategist → Content Writer → Editor → SEO Specialist + +**Best For:** +- Blog post creation +- Article writing +- Marketing content +- Technical documentation +- SEO-optimized content + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.content_creation_pipeline( + model_name="gpt-4o-mini" +) + +result = swarm.run("Create a blog post about AI agents in customer service") +print(result) +``` + +--- + +### 3. Code Development Team + +**Use Case:** Software development, feature implementation, code review + +**Workflow:** Requirements Analyst → Software Developer → QA Engineer → Code Reviewer + +**Best For:** +- Feature development +- Code generation +- Technical implementation +- Code quality review +- Bug fixing + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.code_development_team( + model_name="gpt-4o-mini" +) + +result = swarm.run("Create a REST API for user authentication") +print(result) +``` + +--- + +### 4. Financial Analysis Team + +**Use Case:** Investment analysis, financial reporting, risk assessment + +**Workflow:** Data Collector → Financial Analyst → Risk Assessor → Report Writer + +**Best For:** +- Investment analysis +- Financial performance evaluation +- Risk assessment +- Financial reporting +- Due diligence + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.financial_analysis_team( + model_name="gpt-4o-mini" +) + +result = swarm.run("Analyze Tesla's Q4 2024 financial performance") +print(result) +``` + +--- + +### 5. Marketing Campaign Team + +**Use Case:** Marketing campaigns, product launches, brand awareness + +**Workflow:** Marketing Strategist → Content Creator → Creative Director → Distribution Specialist + +**Best For:** +- Campaign planning +- Product launches +- Marketing strategy +- Content marketing +- Multi-channel campaigns + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.marketing_campaign_team( + model_name="gpt-4o-mini" +) + +result = swarm.run("Create a product launch campaign for an AI fitness app") +print(result) +``` + +--- + +### 6. Customer Support Team + +**Use Case:** Customer service, technical support, issue resolution + +**Workflow:** Support Triage → Technical Support → Customer Success + +**Best For:** +- Customer inquiries +- Technical troubleshooting +- Issue resolution +- Customer satisfaction +- Support ticket handling + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.customer_support_team( + model_name="gpt-4o-mini" +) + +result = swarm.run("Customer reports login issues after password reset") +print(result) +``` + +--- + +### 7. Legal Document Review + +**Use Case:** Contract review, legal compliance, risk analysis + +**Workflow:** Legal Analyst → Compliance Specialist → Risk Assessor → Summary Writer + +**Best For:** +- Contract review +- Legal compliance checking +- Risk assessment +- Due diligence +- Legal document analysis + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.legal_document_review( + model_name="gpt-4o-mini" +) + +result = swarm.run("Review this SaaS service agreement") +print(result) +``` + +--- + +### 8. Data Science Pipeline + +**Use Case:** Data analysis, ML pipelines, business intelligence + +**Workflow:** Data Collector → Data Cleaner → Data Analyst → Visualization Specialist + +**Best For:** +- Data analysis projects +- ML pipeline design +- Business intelligence +- Predictive analytics +- Data visualization + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.data_science_pipeline( + model_name="gpt-4o-mini" +) + +result = swarm.run("Analyze customer churn data and identify key factors") +print(result) +``` + +--- + +## Class Reference + +### SwarmTemplates + +Main class providing access to all swarm templates. + +#### Class Methods + +##### `list_templates()` + +Get a list of all available swarm templates with descriptions. + +**Returns:** +- `List[Dict[str, str]]`: List of templates with metadata + +**Example:** + +```python +from swarms import SwarmTemplates + +templates = SwarmTemplates.list_templates() + +for template in templates: + print(f"{template['name']}: {template['description']}") +``` + +**Output:** + +``` +research_analysis_synthesis: Research → Analysis → Synthesis workflow +content_creation_pipeline: Planning → Writing → Editing → SEO workflow +code_development_team: Requirements → Development → Testing → Review workflow +... +``` + +--- + +##### `get_template_info(template_name: str)` + +Get detailed information about a specific template. + +**Parameters:** +- `template_name` (str): Name of the template + +**Returns:** +- `Dict[str, Any]`: Detailed template information + +**Raises:** +- `ValueError`: If template name is not found + +**Example:** + +```python +from swarms import SwarmTemplates + +info = SwarmTemplates.get_template_info("research_analysis_synthesis") + +print(f"Description: {info['description']}") +print(f"Use Case: {info['use_case']}") +print(f"Agents: {info['agents']}") +``` + +--- + +##### `create(template_name, model_name, max_loops, verbose, custom_params)` + +Universal method to create any swarm template. + +**Parameters:** +- `template_name` (str): Name of the template to create +- `model_name` (str): LLM model to use for all agents (default: "gpt-4o-mini") +- `max_loops` (int): Maximum loops for agent execution (default: 1) +- `verbose` (bool): Enable verbose logging (default: True) +- `custom_params` (Optional[Dict[str, Any]]): Additional custom parameters + +**Returns:** +- Configured swarm workflow (type depends on template) + +**Raises:** +- `ValueError`: If template name is not found + +**Example:** + +```python +from swarms import SwarmTemplates + +# Create a template by name +swarm = SwarmTemplates.create( + template_name="research_analysis_synthesis", + model_name="gpt-4o-mini", + max_loops=1, + verbose=True +) + +result = swarm.run("Analyze quantum computing trends") +print(result) +``` + +--- + +##### Template-Specific Methods + +Each template also has its own dedicated method: + +**Method Signatures:** + +```python +@classmethod +def research_analysis_synthesis( + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, +) -> SequentialWorkflow: + ... + +@classmethod +def content_creation_pipeline( + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, +) -> SequentialWorkflow: + ... + +# Similar signatures for all other templates +``` + +**Common Parameters:** +- `model_name` (str): LLM model to use +- `max_loops` (int): Maximum loops per agent +- `verbose` (bool): Enable verbose logging + +**Returns:** +- Configured SequentialWorkflow instance + +--- + +## Usage Patterns + +### Pattern 1: Quick Start + +Use the universal `create()` method for quick instantiation: + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.create("content_creation_pipeline") +result = swarm.run("Write a blog post about AI") +``` + +### Pattern 2: Direct Method Call + +Use template-specific methods for type hints and IDE support: + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.research_analysis_synthesis( + model_name="gpt-4o", + verbose=True +) + +result = swarm.run("Research renewable energy trends") +``` + +### Pattern 3: Discovery and Selection + +List templates and select based on use case: + +```python +from swarms import SwarmTemplates + +# List all templates +templates = SwarmTemplates.list_templates() + +# Find template for your use case +for template in templates: + if "financial" in template['use_case'].lower(): + print(f"Found: {template['name']}") + + # Get detailed info + info = SwarmTemplates.get_template_info(template['name']) + + # Create and use + swarm = SwarmTemplates.create(template['name']) +``` + +### Pattern 4: Custom Model Configuration + +Use different models for different needs: + +```python +from swarms import SwarmTemplates + +# High-quality model for critical analysis +financial_swarm = SwarmTemplates.financial_analysis_team( + model_name="gpt-4o" # Premium model +) + +# Cost-effective model for routine tasks +content_swarm = SwarmTemplates.content_creation_pipeline( + model_name="gpt-4o-mini" # Economy model +) +``` + +### Pattern 5: Batch Processing + +Process multiple tasks with the same template: + +```python +from swarms import SwarmTemplates + +# Create template once +swarm = SwarmTemplates.customer_support_team() + +# Process multiple support tickets +tickets = [ + "Customer can't reset password", + "Billing question about invoice", + "Feature request for dark mode" +] + +for ticket in tickets: + result = swarm.run(ticket) + print(f"Resolution: {result}\n") +``` + +--- + +## Customization + +### Custom Models + +All templates support any LLM model that works with Swarms: + +```python +from swarms import SwarmTemplates + +# OpenAI models +swarm = SwarmTemplates.create("research_analysis_synthesis", + model_name="gpt-4o" +) + +# Anthropic models +swarm = SwarmTemplates.create("code_development_team", + model_name="claude-sonnet-4-5" +) + +# Local models +swarm = SwarmTemplates.create("content_creation_pipeline", + model_name="ollama/llama3" +) +``` + +### Max Loops Configuration + +Control agent iteration depth: + +```python +from swarms import SwarmTemplates + +# Single-pass workflow (default) +swarm = SwarmTemplates.research_analysis_synthesis( + max_loops=1 +) + +# Multi-iteration with refinement +swarm = SwarmTemplates.code_development_team( + max_loops=3 # Allows for iteration and improvement +) +``` + +### Verbose Logging + +Control logging verbosity: + +```python +from swarms import SwarmTemplates + +# Detailed logging (default) +swarm = SwarmTemplates.financial_analysis_team( + verbose=True +) + +# Silent operation +swarm = SwarmTemplates.content_creation_pipeline( + verbose=False +) +``` + +--- + +## Best Practices + +### 1. Choose the Right Template + +Select templates based on your specific use case: + +```python +# For research and analysis +research_swarm = SwarmTemplates.research_analysis_synthesis() + +# For content generation +content_swarm = SwarmTemplates.content_creation_pipeline() + +# For technical tasks +code_swarm = SwarmTemplates.code_development_team() +``` + +### 2. Provide Clear, Detailed Tasks + +Give agents comprehensive context: + +```python +# ❌ Too vague +result = swarm.run("Analyze the company") + +# ✅ Clear and detailed +result = swarm.run(""" +Analyze Company XYZ's financial performance including: +- Revenue growth trends (last 3 years) +- Profitability metrics +- Cash flow analysis +- Key risk factors +- Competitive positioning +""") +``` + +### 3. Use Appropriate Models + +Match model capability to task complexity: + +```python +# Complex analysis → Premium model +financial_swarm = SwarmTemplates.financial_analysis_team( + model_name="gpt-4o" +) + +# Routine tasks → Economy model +support_swarm = SwarmTemplates.customer_support_team( + model_name="gpt-4o-mini" +) +``` + +### 4. Handle Outputs Appropriately + +Process and validate results: + +```python +from swarms import SwarmTemplates + +swarm = SwarmTemplates.data_science_pipeline() + +result = swarm.run("Analyze customer data") + +# Validate output +if result and len(result) > 100: + # Process valid result + with open("analysis_report.txt", "w") as f: + f.write(result) +else: + # Handle insufficient output + print("Warning: Analysis incomplete") +``` + +### 5. Reuse Templates for Similar Tasks + +Create once, use multiple times: + +```python +from swarms import SwarmTemplates + +# Create template instance +content_swarm = SwarmTemplates.content_creation_pipeline() + +# Reuse for multiple content pieces +topics = ["AI in Healthcare", "Future of Work", "Climate Tech"] + +for topic in topics: + article = content_swarm.run(f"Write article about {topic}") + # Save or process article +``` + +--- + +## Production Deployment + +### Error Handling + +Implement robust error handling: + +```python +from swarms import SwarmTemplates + +def run_swarm_safely(template_name, task): + """ + Safely execute swarm with error handling + """ + try: + swarm = SwarmTemplates.create( + template_name=template_name, + model_name="gpt-4o-mini", + verbose=True + ) + + result = swarm.run(task) + + if not result: + raise ValueError("Empty result from swarm") + + return {"success": True, "result": result} + + except ValueError as e: + return {"success": False, "error": str(e)} + except Exception as e: + return {"success": False, "error": f"Unexpected error: {str(e)}"} + +# Usage +response = run_swarm_safely( + "research_analysis_synthesis", + "Analyze market trends" +) + +if response["success"]: + print(response["result"]) +else: + print(f"Error: {response['error']}") +``` + +### Performance Monitoring + +Track template performance: + +```python +import time +from swarms import SwarmTemplates + +def run_with_monitoring(template_name, task): + """ + Run swarm with performance tracking + """ + start_time = time.time() + + swarm = SwarmTemplates.create(template_name) + result = swarm.run(task) + + execution_time = time.time() - start_time + + return { + "result": result, + "execution_time": execution_time, + "template": template_name + } + +# Track performance +metrics = run_with_monitoring( + "financial_analysis_team", + "Analyze Q4 results" +) + +print(f"Completed in {metrics['execution_time']:.2f}s") +``` + +### Logging and Auditing + +Implement comprehensive logging: + +```python +import logging +from swarms import SwarmTemplates + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +def run_with_logging(template_name, task, user_id=None): + """ + Run swarm with comprehensive logging + """ + logger.info(f"Starting swarm: {template_name}") + logger.info(f"User: {user_id}, Task: {task[:100]}") + + try: + swarm = SwarmTemplates.create(template_name) + result = swarm.run(task) + + logger.info(f"Swarm completed successfully") + logger.info(f"Result length: {len(result)} characters") + + return result + + except Exception as e: + logger.error(f"Swarm failed: {str(e)}") + raise + +# Usage with logging +result = run_with_logging( + template_name="research_analysis_synthesis", + task="Market analysis", + user_id="user_123" +) +``` + +--- + +## FAQ + +### Q: Can I modify the agent prompts in templates? + +**A:** Templates use pre-optimized prompts. For custom prompts, create agents manually using the `Agent` class. + +### Q: Which template should I use for my use case? + +**A:** Use `SwarmTemplates.list_templates()` to see all options with use cases, or refer to the template descriptions above. + +### Q: Can templates work with local models? + +**A:** Yes! Templates support any model compatible with Swarms: + +```python +swarm = SwarmTemplates.create( + "content_creation_pipeline", + model_name="ollama/llama3" +) +``` + +### Q: How do I handle long-running tasks? + +**A:** Increase `max_loops` for iterative refinement: + +```python +swarm = SwarmTemplates.create( + "code_development_team", + max_loops=3 # Allow multiple iterations +) +``` + +### Q: Are templates suitable for production? + +**A:** Yes! Templates are production-tested and optimized. Implement proper error handling and monitoring for production deployments. + +### Q: Can I mix agents from different templates? + +**A:** Templates return workflow objects. For custom agent combinations, use `Agent` and workflow classes directly. + +### Q: How much do template operations cost? + +**A:** Cost depends on your chosen model and task complexity. Templates use sequential workflows, so expect 3-4 agent calls per task. + +--- + +## Examples + +See comprehensive examples in `/examples/swarm_templates_examples.py` + +Run examples: + +```bash +cd examples +python swarm_templates_examples.py +``` + +--- + +## Support + +For issues, questions, or feature requests: +- GitHub Issues: https://github.com/kyegomez/swarms/issues +- Discord: https://discord.gg/EamjgSaEQf +- Documentation: https://docs.swarms.world + diff --git a/examples/guides/x402_examples/README.md b/examples/guides/x402_examples/README.md new file mode 100644 index 00000000..21a4bca3 --- /dev/null +++ b/examples/guides/x402_examples/README.md @@ -0,0 +1,102 @@ +# X402 Research Agent Example + +A simple example showing how to build a research agent with Swarms and monetize it using the x402 payment framework. + +## What is x402? + +x402 is a payment protocol that enables you to monetize your AI agents and APIs with crypto payments. It works by adding payment middleware to your endpoints, allowing you to charge users per request. + +## Quick Start + +### 1. Install Dependencies + +```bash +pip install swarms swarms-tools x402 fastapi uvicorn python-dotenv +``` + +### 2. Set Environment Variables + +Create a `.env` file: + +```bash +OPENAI_API_KEY=your-openai-api-key +EXA_API_KEY=your-exa-api-key +``` + +### 3. Update the Wallet Address + +Edit `research_agent_x402_example.py` and replace: + +```python +pay_to_address="0xYourWalletAddressHere" +``` + +with your actual EVM-compatible wallet address (e.g., MetaMask, Coinbase Wallet). + +### 4. Run the Server + +```bash +python research_agent_x402_example.py +``` + +The server will start at `http://localhost:8000` + +### 5. Test the Integration + +#### Free Health Check Endpoint +```bash +curl http://localhost:8000/ +``` + +#### Paid Research Endpoint (requires payment) +```bash +curl http://localhost:8000/research?query="What are the latest breakthroughs in quantum computing?" +``` + +This will return a 402 Payment Required response with payment instructions. + +## How It Works + +1. **Agent Creation**: The code creates a Swarms agent with the `exa_search` tool for conducting research +2. **Payment Middleware**: x402's `require_payment` middleware protects the `/research` endpoint +3. **Payment Flow**: + - Client requests the endpoint + - Server responds with 402 Payment Required + payment instructions + - Client makes payment (handled by x402 client SDK) + - Client retries request with payment proof + - Server verifies payment and returns research results + +## Going to Production (Mainnet) + +For testnet (current setup): +- Network: `base-sepolia` +- Uses free facilitator: `https://x402.org/facilitator` +- Test USDC only + +For mainnet: +1. Get CDP API credentials from [cdp.coinbase.com](https://cdp.coinbase.com) +2. Update the code: + ```python + from cdp.x402 import create_facilitator_config + + facilitator_config = create_facilitator_config( + api_key_id=os.getenv("CDP_API_KEY_ID"), + api_key_secret=os.getenv("CDP_API_KEY_SECRET"), + ) + + # Change network to "base" and add facilitator_config + require_payment( + path="/research", + price="$0.01", + pay_to_address="0xYourWalletAddress", + network_id="base", # Changed from base-sepolia + facilitator_config=facilitator_config, + ... + ) + ``` + +## Learn More + +- [X402 Documentation](https://docs.cdp.coinbase.com/x402) +- [Swarms Documentation](https://docs.swarms.world) +- [X402 GitHub Examples](https://github.com/coinbase/x402/tree/main/examples) diff --git a/examples/guides/x402_examples/env.example b/examples/guides/x402_examples/env.example new file mode 100644 index 00000000..93bfcef0 --- /dev/null +++ b/examples/guides/x402_examples/env.example @@ -0,0 +1,17 @@ +# OpenAI API Key (required) +OPENAI_API_KEY=sk-your-openai-api-key-here + +# Your wallet address for receiving payments (required) +# This should be an EVM-compatible address (Ethereum, Base, etc.) +WALLET_ADDRESS=0xYourWalletAddressHere + +# For mainnet only (optional during testnet testing) +# Get these from https://cdp.coinbase.com +CDP_API_KEY_ID=your-cdp-api-key-id +CDP_API_KEY_SECRET=your-cdp-api-key-secret + +# Optional: Configure the agent +AGENT_MODEL_NAME=gpt-4o +AGENT_TEMPERATURE=0.5 +AGENT_MAX_LOOPS=1 + diff --git a/examples/guides/x402_examples/memecoin_agent_x402.py b/examples/guides/x402_examples/memecoin_agent_x402.py new file mode 100644 index 00000000..43f68037 --- /dev/null +++ b/examples/guides/x402_examples/memecoin_agent_x402.py @@ -0,0 +1,281 @@ +import os +from typing import Any, Dict + +from dotenv import load_dotenv +from fastapi import FastAPI +from pydantic import BaseModel, Field + +from swarms import Agent +from x402.fastapi.middleware import require_payment + +# Load environment variables +load_dotenv() + + +# Custom system prompt for memecoin analysis +MEMECOIN_ANALYSIS_PROMPT = """You are an expert cryptocurrency analyst specializing in memecoin evaluation and risk assessment. + +Your role is to provide comprehensive, data-driven analysis of memecoins to help investors make informed decisions. + +For each memecoin analysis, you should evaluate: + +1. **Market Overview**: + - Current market cap and trading volume + - Price trends and volatility + - Liquidity assessment + +2. **Community & Social Sentiment**: + - Social media presence and engagement + - Community size and activity + - Influencer endorsements or warnings + - Trending status on platforms + +3. **Technical Analysis**: + - Price patterns and trends + - Support and resistance levels + - Trading volume analysis + - Key technical indicators (RSI, MACD, etc.) + +4. **Risk Assessment**: + - Rug pull indicators + - Liquidity risks + - Smart contract security + - Team transparency + - Concentration of holdings + +5. **Investment Recommendation**: + - Risk level (Low/Medium/High/Extreme) + - Potential upside and downside + - Time horizon considerations + - Position sizing recommendations + +Always provide balanced, objective analysis. Emphasize risks clearly and never guarantee returns. +Be honest about the speculative nature of memecoin investments. + +Format your analysis in a clear, structured manner with actionable insights. +""" + + +# Initialize FastAPI app +app = FastAPI( + title="Memecoin Analysis Agent", + description="AI-powered memecoin analysis service monetized with x402", + version="1.0.0", +) + + +def create_memecoin_agent( + agent_name: str = "Memecoin-Analyzer", + model_name: str = "gpt-4o", + max_loops: int = 1, + temperature: float = 0.5, +) -> Agent: + """ + Create a specialized agent for memecoin analysis. + + Args: + agent_name: Name identifier for the agent + model_name: The LLM model to use for analysis + max_loops: Maximum number of reasoning loops + temperature: Temperature for response generation (0.0-1.0) + + Returns: + Configured Agent instance for memecoin analysis + """ + agent = Agent( + agent_name=agent_name, + system_prompt=MEMECOIN_ANALYSIS_PROMPT, + model_name=model_name, + max_loops=max_loops, + autosave=True, + dashboard=False, + verbose=True, + dynamic_temperature_enabled=True, + temperature=temperature, + ) + return agent + + +# Initialize the memecoin analysis agent +memecoin_agent = create_memecoin_agent() + + +# Request model +class MemecoinRequest(BaseModel): + """ + Request model for memecoin analysis. + + Attributes: + symbol: The memecoin ticker symbol (e.g., 'DOGE', 'PEPE', 'SHIB') + include_social: Whether to include social media sentiment analysis + include_technicals: Whether to include technical analysis + """ + + symbol: str = Field(..., description="Memecoin ticker symbol") + include_social: bool = Field( + default=True, description="Include social media sentiment" + ) + include_technicals: bool = Field( + default=True, description="Include technical analysis" + ) + + +# Apply x402 payment middleware to the analysis endpoint +# For testnet (Base Sepolia), use the free facilitator +# For mainnet, see configuration in README.md +app.middleware("http")( + require_payment( + path="/analyze-memecoin", + price="$0.10", # 10 cents per analysis + pay_to_address=os.getenv( + "WALLET_ADDRESS", "0xYourWalletAddress" + ), + network_id="base-sepolia", # Use "base" for mainnet + description="Get comprehensive AI-powered memecoin analysis including market sentiment, risk assessment, and investment recommendations", + input_schema={ + "type": "object", + "properties": { + "symbol": { + "type": "string", + "description": "Memecoin ticker symbol (e.g., DOGE, PEPE, SHIB)", + }, + "include_social": { + "type": "boolean", + "description": "Include social media sentiment analysis", + "default": True, + }, + "include_technicals": { + "type": "boolean", + "description": "Include technical analysis indicators", + "default": True, + }, + }, + "required": ["symbol"], + }, + output_schema={ + "type": "object", + "properties": { + "symbol": {"type": "string"}, + "analysis": {"type": "string"}, + "risk_level": {"type": "string"}, + "recommendation": {"type": "string"}, + "timestamp": {"type": "string"}, + }, + }, + ) +) + + +@app.get("/") +async def root() -> Dict[str, str]: + """ + Root endpoint providing API information. + + Returns: + Dictionary with API details and usage instructions + """ + return { + "service": "Memecoin Analysis Agent", + "version": "1.0.0", + "description": "AI-powered memecoin analysis monetized with x402", + "endpoints": { + "/analyze-memecoin": "POST - Analyze a memecoin (requires payment)", + "/health": "GET - Health check endpoint (free)", + }, + "pricing": "$0.10 per analysis", + "network": "base-sepolia (testnet)", + } + + +@app.get("/health") +async def health_check() -> Dict[str, str]: + """ + Health check endpoint (free, no payment required). + + Returns: + Dictionary with service status + """ + return {"status": "healthy", "agent": "ready"} + + +@app.post("/analyze-memecoin") +async def analyze_memecoin( + request: MemecoinRequest, +) -> Dict[str, Any]: + """ + Analyze a memecoin using the AI agent (requires x402 payment). + + This endpoint is protected by x402 payment middleware. Users must + pay $0.10 in USDC to access the analysis. + + Args: + request: MemecoinRequest with symbol and analysis options + + Returns: + Dictionary containing comprehensive memecoin analysis + """ + # Build the analysis query + query_parts = [f"Analyze the memecoin {request.symbol.upper()}."] + + if request.include_social: + query_parts.append( + "Include detailed social media sentiment analysis." + ) + + if request.include_technicals: + query_parts.append( + "Include comprehensive technical analysis with key indicators." + ) + + query_parts.append( + "Provide a clear risk assessment and investment recommendation." + ) + + query = " ".join(query_parts) + + # Run the agent analysis + analysis_result = memecoin_agent.run(query) + + # Return structured response + return { + "symbol": request.symbol.upper(), + "analysis": analysis_result, + "risk_level": "See analysis for details", + "recommendation": "See analysis for details", + "timestamp": "2025-10-29", + "disclaimer": "This analysis is for informational purposes only. Not financial advice.", + } + + +@app.post("/batch-analyze") +async def batch_analyze(symbols: list[str]) -> Dict[str, Any]: + """ + Analyze multiple memecoins in a batch (free endpoint for demo). + + Args: + symbols: List of memecoin ticker symbols + + Returns: + Dictionary with analysis for multiple coins + """ + results = {} + for symbol in symbols[:3]: # Limit to 3 for demo + query = ( + f"Provide a brief overview of {symbol.upper()} memecoin." + ) + analysis = memecoin_agent.run(query) + results[symbol.upper()] = { + "brief_analysis": analysis[:200] + "...", + "full_analysis_endpoint": "/analyze-memecoin", + } + + return { + "batch_results": results, + "note": "For full detailed analysis, use /analyze-memecoin (requires payment)", + } + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/examples/guides/x402_examples/research_agent_x402_example.py b/examples/guides/x402_examples/research_agent_x402_example.py new file mode 100644 index 00000000..d301fe66 --- /dev/null +++ b/examples/guides/x402_examples/research_agent_x402_example.py @@ -0,0 +1,83 @@ +from dotenv import load_dotenv +from fastapi import FastAPI +from swarms_tools import exa_search + +from swarms import Agent +from x402.fastapi.middleware import require_payment + +# Load environment variables +load_dotenv() + +app = FastAPI(title="Research Agent API") + +# Initialize the research agent +research_agent = Agent( + agent_name="Research-Agent", + system_prompt="You are an expert research analyst. Conduct thorough research on the given topic and provide comprehensive, well-structured insights with citations.", + model_name="gpt-4o-mini", + max_loops=1, + tools=[exa_search], +) + + +# Apply x402 payment middleware to the research endpoint +app.middleware("http")( + require_payment( + path="/research", + price="$0.01", + pay_to_address="0xYourWalletAddressHere", + network_id="base-sepolia", + description="AI-powered research agent that conducts comprehensive research on any topic", + input_schema={ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Research topic or question", + } + }, + "required": ["query"], + }, + output_schema={ + "type": "object", + "properties": { + "research": { + "type": "string", + "description": "Comprehensive research results", + } + }, + }, + ) +) + + +@app.get("/research") +async def conduct_research(query: str): + """ + Conduct research on a given topic using the research agent. + + Args: + query: The research topic or question + + Returns: + Research results from the agent + """ + result = research_agent.run(query) + return {"research": result} + + +@app.get("/") +async def root(): + """Health check endpoint (free, no payment required)""" + return { + "message": "Research Agent API with x402 payments", + "endpoints": { + "/research": "Paid endpoint - $0.01 per request", + }, + } + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/examples/single_agent/tools/exa_search_agent.py b/examples/single_agent/tools/exa_search_agent.py index ca023cb1..714f790a 100644 --- a/examples/single_agent/tools/exa_search_agent.py +++ b/examples/single_agent/tools/exa_search_agent.py @@ -3,8 +3,8 @@ from swarms_tools import exa_search agent = Agent( - name="Exa Search Agent", - llm="gpt-4o-mini", + agent_name="Exa Search Agent", + model_name="gpt-4o-mini", tools=[exa_search], ) diff --git a/swarms/structs/batched_grid_workflow.py b/swarms/structs/batched_grid_workflow.py index 66e6d71f..239b55dc 100644 --- a/swarms/structs/batched_grid_workflow.py +++ b/swarms/structs/batched_grid_workflow.py @@ -8,6 +8,7 @@ from swarms.structs.multi_agent_exec import ( ) from swarms.structs.omni_agent_types import AgentType from swarms.structs.swarm_id import swarm_id +from swarms.utils.output_types import OutputType class BatchedGridWorkflow: @@ -30,6 +31,7 @@ class BatchedGridWorkflow: description (str): Description of the workflow's purpose. agents (List[AgentType]): List of agents to execute tasks. max_loops (int): Maximum number of execution loops to perform. + output_type (OutputType): Type of output to return. Example: >>> from swarms.structs.batched_grid_workflow import BatchedGridWorkflow @@ -48,6 +50,7 @@ class BatchedGridWorkflow: description: str = "For every agent, run the task on a different task", agents: List[AgentType] = None, max_loops: int = 1, + output_type: OutputType = "dict", ): """ Initialize a BatchedGridWorkflow instance. diff --git a/swarms/structs/swarm_templates.py b/swarms/structs/swarm_templates.py new file mode 100644 index 00000000..ea9d0f36 --- /dev/null +++ b/swarms/structs/swarm_templates.py @@ -0,0 +1,1351 @@ +from typing import Any, Dict, List, Optional + +from swarms.structs.agent import Agent +from swarms.structs.sequential_workflow import SequentialWorkflow +from swarms.utils.loguru_logger import initialize_logger + +logger = initialize_logger(log_folder="swarm_templates") + + +class SwarmTemplates: + """ + A comprehensive library of pre-built, production-ready swarm templates for common use cases. + + This class provides easy access to professionally designed multi-agent systems with + optimized prompts and orchestration patterns. Each template is production-tested and + ready for deployment. + + Attributes: + None (all methods are class methods) + + Methods: + list_templates: Get a list of all available templates + get_template_info: Get detailed information about a specific template + create: Create a swarm from a template name + research_analysis_synthesis: Research and analysis workflow + content_creation_pipeline: Content creation workflow + code_development_team: Software development workflow + financial_analysis_team: Financial analysis workflow + marketing_campaign_team: Marketing campaign workflow + customer_support_team: Customer support workflow + legal_document_review: Legal document review workflow + data_science_pipeline: Data science workflow + + Example: + >>> from swarms.structs.swarm_templates import SwarmTemplates + >>> + >>> # Create a research workflow + >>> swarm = SwarmTemplates.create( + ... "research_analysis_synthesis", + ... model_name="gpt-4o-mini" + ... ) + >>> result = swarm.run("Analyze the impact of AI on healthcare") + >>> print(result) + >>> + >>> # List all available templates + >>> templates = SwarmTemplates.list_templates() + >>> for template in templates: + ... print(f"{template['name']}: {template['description']}") + """ + + @classmethod + def list_templates(cls) -> List[Dict[str, str]]: + """ + Get a list of all available swarm templates with descriptions. + + Returns: + List[Dict[str, str]]: List of templates with name, description, and use_case + + Example: + >>> templates = SwarmTemplates.list_templates() + >>> for template in templates: + ... print(f"{template['name']}: {template['description']}") + """ + return [ + { + "name": "research_analysis_synthesis", + "description": "Research → Analysis → Synthesis workflow for comprehensive topic investigation", + "use_case": "Academic research, market research, competitive analysis", + "orchestration": "Sequential", + "agents": 3, + }, + { + "name": "content_creation_pipeline", + "description": "Planning → Writing → Editing → SEO workflow for content production", + "use_case": "Blog posts, articles, marketing content, documentation", + "orchestration": "Sequential", + "agents": 4, + }, + { + "name": "code_development_team", + "description": "Requirements → Development → Testing → Review workflow for software development", + "use_case": "Software projects, feature development, code review", + "orchestration": "Sequential", + "agents": 4, + }, + { + "name": "financial_analysis_team", + "description": "Data Collection → Analysis → Risk Assessment → Reporting workflow", + "use_case": "Investment analysis, financial reporting, risk assessment", + "orchestration": "Sequential", + "agents": 4, + }, + { + "name": "marketing_campaign_team", + "description": "Strategy → Content → Design → Distribution workflow", + "use_case": "Marketing campaigns, product launches, brand awareness", + "orchestration": "Sequential", + "agents": 4, + }, + { + "name": "customer_support_team", + "description": "Triage → Resolution → Follow-up workflow", + "use_case": "Customer service, technical support, issue resolution", + "orchestration": "Sequential", + "agents": 3, + }, + { + "name": "legal_document_review", + "description": "Analysis → Compliance → Risk Assessment → Summary workflow", + "use_case": "Contract review, legal compliance, risk analysis", + "orchestration": "Sequential", + "agents": 4, + }, + { + "name": "data_science_pipeline", + "description": "Collection → Cleaning → Analysis → Visualization workflow", + "use_case": "Data analysis, ML pipelines, business intelligence", + "orchestration": "Sequential", + "agents": 4, + }, + ] + + @classmethod + def get_template_info(cls, template_name: str) -> Dict[str, Any]: + """ + Get detailed information about a specific template. + + Args: + template_name (str): Name of the template + + Returns: + Dict[str, Any]: Detailed template information + + Raises: + ValueError: If template name is not found + + Example: + >>> info = SwarmTemplates.get_template_info("research_analysis_synthesis") + >>> print(info['description']) + """ + templates = cls.list_templates() + for template in templates: + if template["name"] == template_name: + return template + + available = [t["name"] for t in templates] + raise ValueError( + f"Template '{template_name}' not found. Available templates: {available}" + ) + + @classmethod + def create( + cls, + template_name: str, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + custom_params: Optional[Dict[str, Any]] = None, + ) -> Any: + """ + Create a swarm from a template name. + + Args: + template_name (str): Name of the template to create + model_name (str): Name of the LLM model to use for all agents + max_loops (int): Maximum number of loops for agent execution + verbose (bool): Enable verbose logging + custom_params (Optional[Dict[str, Any]]): Additional custom parameters + + Returns: + Any: Configured swarm workflow (SequentialWorkflow, AgentRearrange, etc.) + + Raises: + ValueError: If template name is not found + + Example: + >>> swarm = SwarmTemplates.create( + ... "research_analysis_synthesis", + ... model_name="gpt-4o-mini", + ... max_loops=1 + ... ) + >>> result = swarm.run("Analyze quantum computing trends") + """ + custom_params = custom_params or {} + + # Map template names to creation methods + template_methods = { + "research_analysis_synthesis": cls.research_analysis_synthesis, + "content_creation_pipeline": cls.content_creation_pipeline, + "code_development_team": cls.code_development_team, + "financial_analysis_team": cls.financial_analysis_team, + "marketing_campaign_team": cls.marketing_campaign_team, + "customer_support_team": cls.customer_support_team, + "legal_document_review": cls.legal_document_review, + "data_science_pipeline": cls.data_science_pipeline, + } + + if template_name not in template_methods: + available = list(template_methods.keys()) + raise ValueError( + f"Template '{template_name}' not found. Available templates: {available}" + ) + + logger.info(f"Creating swarm template: {template_name}") + + return template_methods[template_name]( + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + **custom_params, + ) + + @classmethod + def research_analysis_synthesis( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a research, analysis, and synthesis workflow. + + This template creates a three-agent sequential workflow optimized for + comprehensive research and analysis tasks. The workflow follows: + 1. Researcher: Gathers information and creates comprehensive research + 2. Analyst: Analyzes the research and extracts key insights + 3. Synthesizer: Synthesizes findings into actionable conclusions + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured research workflow + + Example: + >>> swarm = SwarmTemplates.research_analysis_synthesis( + ... model_name="gpt-4o-mini" + ... ) + >>> result = swarm.run("Research the latest trends in renewable energy") + >>> print(result) + """ + researcher = Agent( + agent_name="Research-Specialist", + agent_description="Expert in comprehensive research and information gathering across multiple sources", + system_prompt="""You are a world-class research specialist with expertise in gathering, organizing, and synthesizing information from diverse sources. + +Your responsibilities: +- Conduct thorough research on the given topic +- Identify key facts, statistics, and relevant information +- Organize information in a logical, structured manner +- Cite sources and provide context for findings +- Identify knowledge gaps and areas requiring deeper investigation +- Present research in a clear, comprehensive format + +Guidelines: +- Be thorough and exhaustive in your research +- Cross-reference multiple sources for accuracy +- Highlight contradictions or debates in the field +- Provide context and background information +- Use bullet points and structured formats for clarity +- Include relevant data, statistics, and examples""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + analyst = Agent( + agent_name="Analysis-Expert", + agent_description="Specialist in analyzing research data and extracting meaningful insights", + system_prompt="""You are an expert analyst with deep experience in interpreting research, identifying patterns, and extracting actionable insights. + +Your responsibilities: +- Analyze the research provided by the Research Specialist +- Identify key patterns, trends, and correlations +- Evaluate the significance and implications of findings +- Assess strengths and limitations of the research +- Provide critical analysis and multiple perspectives +- Connect findings to broader contexts and trends + +Guidelines: +- Use analytical frameworks and methodologies +- Support conclusions with evidence from the research +- Identify cause-and-effect relationships +- Consider alternative interpretations +- Highlight the most important insights +- Present analysis in a structured, logical manner""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + synthesizer = Agent( + agent_name="Synthesis-Specialist", + agent_description="Expert in synthesizing research and analysis into actionable conclusions", + system_prompt="""You are a synthesis expert who excels at combining research and analysis into clear, actionable conclusions and recommendations. + +Your responsibilities: +- Synthesize the research and analysis into a coherent narrative +- Identify the most important takeaways and insights +- Provide actionable recommendations based on findings +- Create executive summaries and key points +- Connect insights to practical applications +- Present a balanced, well-reasoned conclusion + +Guidelines: +- Integrate findings from both research and analysis +- Prioritize the most significant insights +- Provide clear, actionable recommendations +- Consider implications and next steps +- Present conclusions in an accessible format +- Balance depth with clarity and conciseness""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Research-Analysis-Synthesis-Workflow", + description="A comprehensive workflow for research, analysis, and synthesis", + agents=[researcher, analyst, synthesizer], + max_loops=1, + verbose=verbose, + ) + + logger.info( + "Research-Analysis-Synthesis workflow created successfully" + ) + return workflow + + @classmethod + def content_creation_pipeline( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a content creation workflow with planning, writing, editing, and SEO optimization. + + This template creates a four-agent sequential workflow optimized for + professional content creation. The workflow follows: + 1. Content Strategist: Plans content structure and strategy + 2. Content Writer: Writes engaging, high-quality content + 3. Editor: Refines and polishes the content + 4. SEO Specialist: Optimizes content for search engines + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured content creation workflow + + Example: + >>> swarm = SwarmTemplates.content_creation_pipeline() + >>> result = swarm.run("Create a blog post about AI in healthcare") + >>> print(result) + """ + strategist = Agent( + agent_name="Content-Strategist", + agent_description="Expert in content strategy and planning", + system_prompt="""You are a senior content strategist with expertise in creating compelling content strategies and outlines. + +Your responsibilities: +- Analyze the content topic and target audience +- Create a detailed content outline with key sections +- Define the content's purpose, tone, and messaging +- Identify key points and arguments to cover +- Research audience needs and pain points +- Establish content goals and success metrics + +Guidelines: +- Create comprehensive, structured outlines +- Define clear objectives for each section +- Consider audience perspective and needs +- Include keyword opportunities and SEO considerations +- Provide guidance on tone and style +- Identify potential hooks and compelling angles""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + writer = Agent( + agent_name="Content-Writer", + agent_description="Professional writer specializing in engaging, high-quality content", + system_prompt="""You are an expert content writer with a talent for creating engaging, informative, and persuasive content. + +Your responsibilities: +- Write content based on the strategist's outline +- Create compelling headlines and subheadings +- Write in an engaging, accessible style +- Include examples, anecdotes, and supporting evidence +- Maintain consistent tone and voice throughout +- Ensure logical flow and strong transitions + +Guidelines: +- Follow the provided outline and strategy +- Write clear, concise, and engaging prose +- Use active voice and varied sentence structure +- Include specific examples and data where appropriate +- Create compelling introductions and conclusions +- Maintain reader interest throughout the piece""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + editor = Agent( + agent_name="Content-Editor", + agent_description="Professional editor focused on clarity, quality, and impact", + system_prompt="""You are a professional editor with expertise in refining content for clarity, impact, and quality. + +Your responsibilities: +- Review and refine the written content +- Improve clarity, flow, and readability +- Eliminate redundancy and strengthen weak passages +- Ensure grammatical accuracy and proper style +- Enhance storytelling and narrative structure +- Verify factual accuracy and consistency + +Guidelines: +- Focus on improving clarity and impact +- Tighten prose and eliminate unnecessary words +- Strengthen transitions and logical flow +- Ensure consistent tone and voice +- Check for grammatical and stylistic errors +- Preserve the author's voice while improving quality""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + seo_specialist = Agent( + agent_name="SEO-Specialist", + agent_description="SEO expert focused on optimizing content for search engines", + system_prompt="""You are an SEO specialist with expertise in optimizing content for search engines while maintaining quality and readability. + +Your responsibilities: +- Optimize content for target keywords +- Improve meta descriptions and title tags +- Enhance content structure for SEO (headings, lists, etc.) +- Add internal linking opportunities +- Ensure proper keyword density and placement +- Improve readability and user engagement signals + +Guidelines: +- Maintain natural, readable content while optimizing +- Use keywords strategically in headings and body +- Suggest metadata improvements (title, description) +- Recommend internal linking opportunities +- Ensure mobile-friendly formatting +- Focus on user intent and search intent alignment +- Provide final SEO recommendations and checklist""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Content-Creation-Pipeline", + description="Professional content creation workflow from planning to SEO optimization", + agents=[strategist, writer, editor, seo_specialist], + max_loops=1, + verbose=verbose, + ) + + logger.info("Content-Creation-Pipeline created successfully") + return workflow + + @classmethod + def code_development_team( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a software development workflow with requirements, development, testing, and review. + + This template creates a four-agent sequential workflow optimized for + software development projects. The workflow follows: + 1. Requirements Analyst: Analyzes requirements and creates specifications + 2. Software Developer: Implements the code based on requirements + 3. QA Engineer: Tests the code and identifies issues + 4. Code Reviewer: Reviews code quality and best practices + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured software development workflow + + Example: + >>> swarm = SwarmTemplates.code_development_team() + >>> result = swarm.run("Create a REST API for user authentication") + >>> print(result) + """ + requirements_analyst = Agent( + agent_name="Requirements-Analyst", + agent_description="Expert in analyzing requirements and creating technical specifications", + system_prompt="""You are a senior requirements analyst with expertise in translating business needs into clear technical specifications. + +Your responsibilities: +- Analyze the project requirements and objectives +- Break down requirements into specific, actionable tasks +- Identify technical constraints and dependencies +- Define acceptance criteria and success metrics +- Create detailed technical specifications +- Identify potential risks and challenges + +Guidelines: +- Write clear, unambiguous requirements +- Use specific, measurable acceptance criteria +- Consider edge cases and error scenarios +- Define data models and API contracts when relevant +- Include non-functional requirements (performance, security) +- Prioritize requirements and identify MVP features""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + developer = Agent( + agent_name="Software-Developer", + agent_description="Expert software developer specializing in clean, maintainable code", + system_prompt="""You are an expert software developer with deep knowledge of best practices, design patterns, and clean code principles. + +Your responsibilities: +- Implement code based on the technical specifications +- Write clean, maintainable, well-documented code +- Follow best practices and design patterns +- Include error handling and edge case management +- Write modular, reusable components +- Add inline comments and documentation + +Guidelines: +- Follow the requirements and specifications precisely +- Use clear naming conventions and code organization +- Implement proper error handling and validation +- Write self-documenting code with clear comments +- Consider performance and scalability +- Include usage examples and documentation""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + qa_engineer = Agent( + agent_name="QA-Engineer", + agent_description="Quality assurance engineer focused on testing and validation", + system_prompt="""You are a QA engineer with expertise in testing strategies, test case design, and quality assurance. + +Your responsibilities: +- Review the implemented code against requirements +- Identify potential bugs, issues, and edge cases +- Design test cases and testing strategies +- Verify acceptance criteria are met +- Test error handling and edge cases +- Provide detailed feedback on issues found + +Guidelines: +- Test against all specified requirements +- Consider edge cases and boundary conditions +- Test error handling and validation +- Verify performance and security considerations +- Document all issues found with reproduction steps +- Suggest improvements and optimizations +- Provide a comprehensive testing report""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + code_reviewer = Agent( + agent_name="Code-Reviewer", + agent_description="Senior code reviewer focused on code quality and best practices", + system_prompt="""You are a senior code reviewer with expertise in code quality, security, and best practices. + +Your responsibilities: +- Review code quality and adherence to best practices +- Identify potential security vulnerabilities +- Assess code maintainability and readability +- Verify proper error handling and logging +- Check for performance optimization opportunities +- Provide constructive feedback and recommendations + +Guidelines: +- Focus on code quality, not just functionality +- Check for security vulnerabilities (injection, XSS, etc.) +- Assess code organization and architecture +- Verify proper documentation and comments +- Identify opportunities for refactoring +- Provide specific, actionable recommendations +- Highlight both strengths and areas for improvement""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Code-Development-Team", + description="Complete software development workflow from requirements to code review", + agents=[ + requirements_analyst, + developer, + qa_engineer, + code_reviewer, + ], + max_loops=1, + verbose=verbose, + ) + + logger.info( + "Code-Development-Team workflow created successfully" + ) + return workflow + + @classmethod + def financial_analysis_team( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a financial analysis workflow for comprehensive financial evaluation. + + This template creates a four-agent sequential workflow optimized for + financial analysis and investment decisions. The workflow follows: + 1. Data Collector: Gathers financial data and metrics + 2. Financial Analyst: Analyzes financial performance + 3. Risk Assessor: Evaluates risks and vulnerabilities + 4. Report Writer: Creates comprehensive financial reports + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured financial analysis workflow + + Example: + >>> swarm = SwarmTemplates.financial_analysis_team() + >>> result = swarm.run("Analyze Tesla's Q4 2024 financial performance") + >>> print(result) + """ + data_collector = Agent( + agent_name="Financial-Data-Collector", + agent_description="Expert in gathering and organizing financial data and metrics", + system_prompt="""You are a financial data specialist with expertise in collecting, organizing, and presenting financial information. + +Your responsibilities: +- Gather relevant financial data and metrics +- Organize data in a structured, accessible format +- Identify key financial indicators and ratios +- Collect industry benchmarks and comparisons +- Verify data accuracy and consistency +- Present data with proper context + +Guidelines: +- Focus on relevant, material financial data +- Use standard financial metrics and ratios +- Provide historical context and trends +- Include industry comparisons when relevant +- Organize data in clear tables and formats +- Cite sources for all financial data""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + analyst = Agent( + agent_name="Financial-Analyst", + agent_description="Expert financial analyst specializing in performance evaluation", + system_prompt="""You are a senior financial analyst with expertise in analyzing financial statements, ratios, and business performance. + +Your responsibilities: +- Analyze financial data and performance metrics +- Evaluate profitability, liquidity, and solvency +- Identify trends, patterns, and anomalies +- Assess competitive position and market standing +- Provide insights on financial health and performance +- Make data-driven conclusions and observations + +Guidelines: +- Use comprehensive financial analysis techniques +- Calculate and interpret key financial ratios +- Identify strengths and weaknesses in performance +- Compare against industry benchmarks +- Highlight significant trends and changes +- Support conclusions with quantitative evidence""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + risk_assessor = Agent( + agent_name="Risk-Assessment-Specialist", + agent_description="Expert in identifying and evaluating financial risks", + system_prompt="""You are a risk assessment specialist with expertise in identifying, analyzing, and quantifying financial risks. + +Your responsibilities: +- Identify key financial risks and vulnerabilities +- Assess market, credit, and operational risks +- Evaluate risk mitigation strategies +- Quantify potential impact of identified risks +- Provide risk ratings and prioritization +- Recommend risk management approaches + +Guidelines: +- Consider multiple risk categories (market, credit, operational, etc.) +- Assess both likelihood and potential impact +- Identify early warning indicators +- Consider macro and micro risk factors +- Provide specific risk mitigation recommendations +- Use risk matrices and frameworks where appropriate""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + report_writer = Agent( + agent_name="Financial-Report-Writer", + agent_description="Expert in creating comprehensive, professional financial reports", + system_prompt="""You are a financial report writer with expertise in creating clear, comprehensive, and actionable financial reports. + +Your responsibilities: +- Synthesize data, analysis, and risk assessment into a cohesive report +- Create executive summary with key findings +- Present findings in a clear, professional format +- Include actionable recommendations +- Provide balanced perspective on opportunities and risks +- Structure report for different stakeholder audiences + +Guidelines: +- Start with executive summary of key findings +- Use clear sections and logical flow +- Include both quantitative data and qualitative insights +- Present balanced view of strengths and concerns +- Provide specific, actionable recommendations +- Use professional financial reporting standards +- Make the report accessible to non-financial stakeholders""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Financial-Analysis-Team", + description="Comprehensive financial analysis workflow from data collection to reporting", + agents=[ + data_collector, + analyst, + risk_assessor, + report_writer, + ], + max_loops=1, + verbose=verbose, + ) + + logger.info( + "Financial-Analysis-Team workflow created successfully" + ) + return workflow + + @classmethod + def marketing_campaign_team( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a marketing campaign workflow from strategy to distribution. + + This template creates a four-agent sequential workflow optimized for + marketing campaign development. The workflow follows: + 1. Marketing Strategist: Develops campaign strategy and positioning + 2. Content Creator: Creates campaign content and messaging + 3. Creative Director: Designs creative elements and visuals + 4. Distribution Specialist: Plans distribution and channels + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured marketing campaign workflow + + Example: + >>> swarm = SwarmTemplates.marketing_campaign_team() + >>> result = swarm.run("Create a product launch campaign for an AI-powered fitness app") + >>> print(result) + """ + strategist = Agent( + agent_name="Marketing-Strategist", + agent_description="Expert in marketing strategy and campaign planning", + system_prompt="""You are a senior marketing strategist with expertise in developing comprehensive marketing campaigns and strategies. + +Your responsibilities: +- Define campaign objectives and success metrics +- Identify target audience and customer personas +- Develop positioning and key messaging +- Create campaign strategy and timeline +- Identify competitive advantages and unique value propositions +- Establish budget recommendations and resource allocation + +Guidelines: +- Start with clear, measurable objectives +- Define specific target audience segments +- Develop compelling positioning and messaging +- Consider the full customer journey +- Identify key channels and tactics +- Create detailed campaign brief for team execution""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + content_creator = Agent( + agent_name="Campaign-Content-Creator", + agent_description="Expert in creating compelling marketing content and copy", + system_prompt="""You are a marketing content specialist with expertise in creating persuasive, engaging campaign content. + +Your responsibilities: +- Create compelling headlines and taglines +- Write persuasive ad copy and content +- Develop email marketing content +- Create social media posts and content +- Write landing page copy +- Ensure consistent messaging across channels + +Guidelines: +- Follow the strategic positioning and messaging +- Write for the specific target audience +- Create urgency and compelling calls-to-action +- Use persuasive copywriting techniques +- Adapt tone for different channels +- Focus on benefits and value propositions +- Keep messaging clear and concise""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + creative_director = Agent( + agent_name="Creative-Director", + agent_description="Expert in creative direction and visual campaign design", + system_prompt="""You are a creative director with expertise in developing compelling visual concepts and creative campaigns. + +Your responsibilities: +- Develop creative concepts and visual themes +- Design visual hierarchy and layout recommendations +- Specify imagery, colors, and design elements +- Create mood boards and creative briefs +- Ensure brand consistency across materials +- Develop creative guidelines for execution + +Guidelines: +- Create distinctive, memorable visual concepts +- Ensure alignment with brand identity +- Design for target audience preferences +- Consider cross-channel consistency +- Specify concrete design elements (colors, fonts, imagery) +- Provide detailed creative direction for designers +- Focus on visual impact and emotional resonance""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + distribution_specialist = Agent( + agent_name="Distribution-Specialist", + agent_description="Expert in marketing channel strategy and campaign distribution", + system_prompt="""You are a distribution specialist with expertise in multi-channel marketing and campaign execution. + +Your responsibilities: +- Develop channel strategy and mix +- Create distribution timeline and schedule +- Recommend budget allocation across channels +- Identify targeting and audience parameters +- Plan A/B tests and optimization strategies +- Define tracking and measurement approach + +Guidelines: +- Select channels based on audience and objectives +- Create detailed distribution timeline +- Recommend specific tactics for each channel +- Include targeting parameters and criteria +- Plan for testing and optimization +- Define KPIs and success metrics for each channel +- Provide implementation roadmap and next steps""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Marketing-Campaign-Team", + description="Complete marketing campaign workflow from strategy to distribution", + agents=[ + strategist, + content_creator, + creative_director, + distribution_specialist, + ], + max_loops=1, + verbose=verbose, + ) + + logger.info( + "Marketing-Campaign-Team workflow created successfully" + ) + return workflow + + @classmethod + def customer_support_team( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a customer support workflow for handling customer inquiries. + + This template creates a three-agent sequential workflow optimized for + customer support and issue resolution. The workflow follows: + 1. Support Triage: Categorizes and prioritizes customer issues + 2. Technical Support: Resolves technical issues and provides solutions + 3. Customer Success: Ensures satisfaction and follows up + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured customer support workflow + + Example: + >>> swarm = SwarmTemplates.customer_support_team() + >>> result = swarm.run("Customer reports login issues after password reset") + >>> print(result) + """ + triage_specialist = Agent( + agent_name="Support-Triage-Specialist", + agent_description="Expert in categorizing and prioritizing customer issues", + system_prompt="""You are a support triage specialist with expertise in quickly assessing and categorizing customer issues. + +Your responsibilities: +- Analyze the customer issue and extract key details +- Categorize the issue type (technical, billing, feature request, etc.) +- Assess urgency and priority level +- Identify required information and clarifying questions +- Route to appropriate support tier +- Document issue details in structured format + +Guidelines: +- Extract all relevant details from customer inquiry +- Categorize issues using standard taxonomy +- Assess priority based on impact and urgency +- Identify any missing information needed +- Provide clear summary of the issue +- Recommend appropriate support path""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + technical_support = Agent( + agent_name="Technical-Support-Specialist", + agent_description="Expert in resolving technical issues and providing solutions", + system_prompt="""You are a senior technical support specialist with deep product knowledge and problem-solving expertise. + +Your responsibilities: +- Analyze the triaged issue thoroughly +- Identify root cause and potential solutions +- Provide step-by-step resolution instructions +- Offer workarounds if immediate fix unavailable +- Explain technical concepts in accessible language +- Escalate complex issues when necessary + +Guidelines: +- Provide clear, actionable resolution steps +- Use simple language avoiding jargon +- Offer multiple solutions when available +- Include screenshots or examples when helpful +- Verify solution addresses the root cause +- Document any known issues or limitations +- Be empathetic and customer-focused""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + customer_success = Agent( + agent_name="Customer-Success-Specialist", + agent_description="Expert in ensuring customer satisfaction and long-term success", + system_prompt="""You are a customer success specialist focused on ensuring customer satisfaction and building long-term relationships. + +Your responsibilities: +- Review the issue resolution and ensure completeness +- Create professional, empathetic follow-up communication +- Identify opportunities to enhance customer experience +- Suggest relevant resources or features +- Check for related issues or concerns +- Ensure customer feels valued and supported + +Guidelines: +- Acknowledge any inconvenience experienced +- Confirm the issue is fully resolved +- Provide additional helpful resources +- Suggest proactive measures to prevent future issues +- Invite feedback and further questions +- End on a positive, supportive note +- Include clear next steps if any action required""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Customer-Support-Team", + description="Customer support workflow from triage to resolution and follow-up", + agents=[ + triage_specialist, + technical_support, + customer_success, + ], + max_loops=1, + verbose=verbose, + ) + + logger.info( + "Customer-Support-Team workflow created successfully" + ) + return workflow + + @classmethod + def legal_document_review( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a legal document review workflow for contracts and agreements. + + This template creates a four-agent sequential workflow optimized for + legal document analysis and review. The workflow follows: + 1. Legal Analyst: Analyzes document structure and key terms + 2. Compliance Specialist: Reviews regulatory compliance + 3. Risk Assessor: Identifies legal risks and liabilities + 4. Summary Writer: Creates executive summary and recommendations + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured legal review workflow + + Example: + >>> swarm = SwarmTemplates.legal_document_review() + >>> result = swarm.run("Review this software licensing agreement") + >>> print(result) + """ + legal_analyst = Agent( + agent_name="Legal-Analyst", + agent_description="Expert in analyzing legal documents and contracts", + system_prompt="""You are a legal analyst with expertise in contract analysis and legal document review. + +Your responsibilities: +- Analyze document structure and key provisions +- Identify all parties and their obligations +- Extract key terms, dates, and conditions +- Identify definitions and legal terminology +- Note any unusual or non-standard clauses +- Summarize main contractual obligations + +Guidelines: +- Read the document thoroughly and systematically +- Identify all key terms and provisions +- Note defined terms and their usage +- Extract important dates and deadlines +- Identify rights, obligations, and restrictions +- Flag any ambiguous or unclear language +- Provide structured summary of key provisions""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + compliance_specialist = Agent( + agent_name="Compliance-Specialist", + agent_description="Expert in regulatory compliance and legal requirements", + system_prompt="""You are a compliance specialist with expertise in regulatory requirements and legal compliance. + +Your responsibilities: +- Review document for regulatory compliance +- Identify applicable laws and regulations +- Verify required clauses and disclosures +- Check for industry-specific requirements +- Identify any compliance gaps or issues +- Recommend necessary compliance additions + +Guidelines: +- Consider all relevant regulatory frameworks +- Verify required legal disclosures are present +- Check for industry-specific compliance requirements +- Identify any missing required clauses +- Note jurisdiction-specific requirements +- Recommend compliance improvements +- Highlight any regulatory red flags""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + risk_assessor = Agent( + agent_name="Legal-Risk-Assessor", + agent_description="Expert in identifying legal risks and liabilities", + system_prompt="""You are a legal risk specialist with expertise in identifying and assessing contractual risks and liabilities. + +Your responsibilities: +- Identify potential legal risks and liabilities +- Assess indemnification and liability clauses +- Evaluate termination and breach provisions +- Analyze dispute resolution mechanisms +- Identify unfavorable or one-sided terms +- Quantify risk levels and potential exposure + +Guidelines: +- Identify all risk factors in the agreement +- Assess liability caps and limitations +- Evaluate indemnification obligations +- Review warranty and representation clauses +- Identify potential breach scenarios +- Assess intellectual property risks +- Provide risk ratings and recommendations""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + summary_writer = Agent( + agent_name="Legal-Summary-Writer", + agent_description="Expert in creating clear legal summaries and recommendations", + system_prompt="""You are a legal summary specialist with expertise in distilling complex legal analysis into clear, actionable summaries. + +Your responsibilities: +- Create executive summary of the document +- Summarize key findings from analysis, compliance, and risk review +- Highlight critical issues and concerns +- Provide clear, actionable recommendations +- Organize information for decision-making +- Present legal concepts in accessible language + +Guidelines: +- Start with brief executive summary +- Highlight most critical issues and risks +- Organize findings by importance +- Provide specific recommendations for each issue +- Use clear language avoiding excessive jargon +- Include decision-making criteria +- Summarize recommended next steps""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Legal-Document-Review", + description="Comprehensive legal document review from analysis to recommendations", + agents=[ + legal_analyst, + compliance_specialist, + risk_assessor, + summary_writer, + ], + max_loops=1, + verbose=verbose, + ) + + logger.info( + "Legal-Document-Review workflow created successfully" + ) + return workflow + + @classmethod + def data_science_pipeline( + cls, + model_name: str = "gpt-4o-mini", + max_loops: int = 1, + verbose: bool = True, + ) -> SequentialWorkflow: + """ + Create a data science workflow from collection to visualization. + + This template creates a four-agent sequential workflow optimized for + data science projects. The workflow follows: + 1. Data Collector: Gathers and documents data sources + 2. Data Cleaner: Cleans and preprocesses data + 3. Data Analyst: Performs analysis and modeling + 4. Visualization Specialist: Creates visualizations and reports + + Args: + model_name (str): Name of the LLM model to use + max_loops (int): Maximum loops for each agent + verbose (bool): Enable verbose logging + + Returns: + SequentialWorkflow: Configured data science workflow + + Example: + >>> swarm = SwarmTemplates.data_science_pipeline() + >>> result = swarm.run("Analyze customer churn data and identify key factors") + >>> print(result) + """ + data_collector = Agent( + agent_name="Data-Collection-Specialist", + agent_description="Expert in data collection and documentation", + system_prompt="""You are a data collection specialist with expertise in gathering, documenting, and organizing data for analysis. + +Your responsibilities: +- Identify required data sources and datasets +- Document data collection methodology +- Describe data structure and schema +- Identify data quality issues and limitations +- Create data dictionary and documentation +- Provide data context and background + +Guidelines: +- Clearly describe all data sources +- Document data collection process +- Identify key variables and features +- Note any data limitations or biases +- Describe data format and structure +- Provide sample data or examples +- Document data quality observations""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + data_cleaner = Agent( + agent_name="Data-Cleaning-Specialist", + agent_description="Expert in data cleaning and preprocessing", + system_prompt="""You are a data cleaning specialist with expertise in data preprocessing, transformation, and quality improvement. + +Your responsibilities: +- Identify data quality issues (missing, duplicates, outliers) +- Design data cleaning strategy +- Recommend preprocessing steps +- Handle missing data appropriately +- Detect and treat outliers +- Transform and normalize data as needed + +Guidelines: +- Systematically identify all data quality issues +- Provide specific cleaning recommendations +- Explain rationale for each preprocessing step +- Consider impact on analysis and modeling +- Document all transformations applied +- Provide before/after data quality metrics +- Create reproducible cleaning pipeline""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + data_analyst = Agent( + agent_name="Data-Analysis-Specialist", + agent_description="Expert in statistical analysis and machine learning", + system_prompt="""You are a data analyst with expertise in statistical analysis, machine learning, and data modeling. + +Your responsibilities: +- Perform exploratory data analysis +- Identify patterns, trends, and relationships +- Select appropriate analytical methods +- Build and validate models if applicable +- Interpret results and findings +- Assess statistical significance + +Guidelines: +- Start with exploratory data analysis +- Use appropriate statistical methods +- Identify key insights and patterns +- Validate findings with multiple approaches +- Consider causation vs correlation +- Assess confidence and uncertainty +- Provide clear interpretation of results +- Document methodology and assumptions""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + visualization_specialist = Agent( + agent_name="Data-Visualization-Specialist", + agent_description="Expert in data visualization and reporting", + system_prompt="""You are a data visualization specialist with expertise in creating clear, insightful visualizations and reports. + +Your responsibilities: +- Design effective visualizations for key findings +- Create comprehensive data reports +- Present insights in accessible format +- Recommend specific chart types and designs +- Provide actionable recommendations +- Create executive summary of findings + +Guidelines: +- Select appropriate visualization types for each insight +- Design clear, intuitive visualizations +- Use color and design principles effectively +- Create layered reporting (executive summary → details) +- Highlight the most important findings +- Provide clear recommendations and next steps +- Make insights actionable for stakeholders +- Include supporting data and methodology notes""", + model_name=model_name, + max_loops=max_loops, + verbose=verbose, + ) + + workflow = SequentialWorkflow( + name="Data-Science-Pipeline", + description="Complete data science workflow from collection to visualization", + agents=[ + data_collector, + data_cleaner, + data_analyst, + visualization_specialist, + ], + max_loops=1, + verbose=verbose, + ) + + logger.info( + "Data-Science-Pipeline workflow created successfully" + ) + return workflow diff --git a/tests/structs/test_batched_grid_workflow.py b/tests/structs/test_batched_grid_workflow.py new file mode 100644 index 00000000..ee17f57f --- /dev/null +++ b/tests/structs/test_batched_grid_workflow.py @@ -0,0 +1,498 @@ +import pytest +from swarms import Agent +from swarms.structs.batched_grid_workflow import BatchedGridWorkflow + + +def test_batched_grid_workflow_basic_execution(): + """Test basic BatchedGridWorkflow execution with multiple agents and tasks""" + # Create specialized agents for different tasks + research_agent = Agent( + agent_name="Research-Analyst", + agent_description="Agent specializing in research and data collection", + model_name="gpt-4o", + max_loops=1, + ) + + analysis_agent = Agent( + agent_name="Data-Analyst", + agent_description="Agent specializing in data analysis", + model_name="gpt-4o", + max_loops=1, + ) + + reporting_agent = Agent( + agent_name="Report-Writer", + agent_description="Agent specializing in report writing", + model_name="gpt-4o", + max_loops=1, + ) + + # Create workflow with multiple agents + workflow = BatchedGridWorkflow( + name="Basic-Batched-Grid-Workflow", + description="Testing basic batched grid execution", + agents=[research_agent, analysis_agent, reporting_agent], + max_loops=1, + ) + + # Run workflow with different tasks for each agent + tasks = [ + "Research the latest trends in artificial intelligence", + "Analyze the impact of AI on job markets", + "Write a summary report on AI developments", + ] + result = workflow.run(tasks) + + # Verify results + assert result is not None + assert isinstance(result, list) + assert len(result) == 1 # max_loops=1 + assert result[0] is not None + + +def test_batched_grid_workflow_multiple_loops(): + """Test BatchedGridWorkflow with multiple execution loops""" + # Create agents for iterative processing + agent1 = Agent( + agent_name="Iteration-Agent-1", + agent_description="Agent for iterative task processing", + model_name="gpt-4o", + max_loops=1, + ) + + agent2 = Agent( + agent_name="Iteration-Agent-2", + agent_description="Agent for iterative task refinement", + model_name="gpt-4o", + max_loops=1, + ) + + workflow = BatchedGridWorkflow( + name="Multi-Loop-Workflow", + description="Testing multiple execution loops", + agents=[agent1, agent2], + max_loops=3, + ) + + tasks = [ + "Generate ideas for a new product", + "Evaluate the feasibility of product ideas", + ] + results = workflow.run(tasks) + + # Verify multiple loop execution + assert results is not None + assert isinstance(results, list) + assert len(results) == 3 # max_loops=3 + for result in results: + assert result is not None + + +def test_batched_grid_workflow_single_agent_single_task(): + """Test BatchedGridWorkflow with a single agent and single task""" + agent = Agent( + agent_name="Solo-Agent", + agent_description="Single agent for solo task execution", + model_name="gpt-4o", + max_loops=1, + ) + + workflow = BatchedGridWorkflow( + name="Single-Agent-Workflow", + description="Testing single agent execution", + agents=[agent], + max_loops=1, + ) + + tasks = ["Analyze the current state of renewable energy"] + result = workflow.run(tasks) + + assert result is not None + assert isinstance(result, list) + assert len(result) == 1 + + +def test_batched_grid_workflow_four_agents(): + """Test BatchedGridWorkflow with four agents in grid pattern""" + agents = [ + Agent( + agent_name=f"Grid-Agent-{i+1}", + agent_description=f"Agent {i+1} for grid execution pattern", + model_name="gpt-4o", + max_loops=1, + ) + for i in range(4) + ] + + workflow = BatchedGridWorkflow( + name="Four-Agent-Grid-Workflow", + description="Testing grid execution with four agents", + agents=agents, + max_loops=2, + ) + + tasks = [ + "Evaluate market trends in technology", + "Assess competitive landscape", + "Analyze customer sentiment", + "Review financial performance", + ] + results = workflow.run(tasks) + + assert results is not None + assert isinstance(results, list) + assert len(results) == 2 # max_loops=2 + for result in results: + assert result is not None + + +def test_batched_grid_workflow_max_loops_validation(): + """Test that BatchedGridWorkflow validates max_loops parameter""" + agent = Agent( + agent_name="Validation-Agent", + agent_description="Agent for validation testing", + model_name="gpt-4o", + max_loops=1, + ) + + # Test with invalid max_loops (zero) + with pytest.raises(ValueError) as exc_info: + BatchedGridWorkflow( + name="Invalid-Loops-Workflow", + agents=[agent], + max_loops=0, + ) + assert "max_loops must be a positive integer" in str( + exc_info.value + ) + + # Test with invalid max_loops (negative) + with pytest.raises(ValueError) as exc_info: + BatchedGridWorkflow( + name="Invalid-Loops-Workflow", + agents=[agent], + max_loops=-5, + ) + assert "max_loops must be a positive integer" in str( + exc_info.value + ) + + +def test_batched_grid_workflow_step_method(): + """Test the step method of BatchedGridWorkflow""" + agent1 = Agent( + agent_name="Step-Agent-1", + agent_description="First agent for step testing", + model_name="gpt-4o", + max_loops=1, + ) + + agent2 = Agent( + agent_name="Step-Agent-2", + agent_description="Second agent for step testing", + model_name="gpt-4o", + max_loops=1, + ) + + workflow = BatchedGridWorkflow( + name="Step-Test-Workflow", + description="Testing step method", + agents=[agent1, agent2], + max_loops=1, + ) + + tasks = [ + "Generate a business plan outline", + "Create a financial projection model", + ] + + # Test single step execution + step_result = workflow.step(tasks) + + assert step_result is not None + assert isinstance(step_result, list) + + +def test_batched_grid_workflow_with_complex_tasks(): + """Test BatchedGridWorkflow with complex multi-step tasks""" + # Create agents with specific expertise + financial_agent = Agent( + agent_name="Financial-Analyst", + agent_description="Expert in financial analysis and forecasting", + model_name="gpt-4o", + max_loops=1, + ) + + technical_agent = Agent( + agent_name="Technical-Expert", + agent_description="Expert in technical analysis and evaluation", + model_name="gpt-4o", + max_loops=1, + ) + + strategic_agent = Agent( + agent_name="Strategic-Planner", + agent_description="Expert in strategic planning and execution", + model_name="gpt-4o", + max_loops=1, + ) + + workflow = BatchedGridWorkflow( + name="Complex-Analysis-Workflow", + description="Workflow for complex multi-perspective analysis", + agents=[financial_agent, technical_agent, strategic_agent], + max_loops=2, + ) + + tasks = [ + "Perform a comprehensive financial analysis of a SaaS company with $10M ARR. " + "Include revenue projections, burn rate analysis, and unit economics.", + "Evaluate the technical architecture and scalability of a cloud-based platform. " + "Assess security measures, infrastructure costs, and technical debt.", + "Develop a strategic expansion plan for entering European markets. " + "Consider regulatory requirements, competition, and go-to-market strategy.", + ] + + results = workflow.run(tasks) + + assert results is not None + assert isinstance(results, list) + assert len(results) == 2 + for result in results: + assert result is not None + assert isinstance(result, list) + assert len(result) == 3 # Three agents + + +def test_batched_grid_workflow_default_parameters(): + """Test BatchedGridWorkflow with default parameters""" + agent = Agent( + agent_name="Default-Test-Agent", + agent_description="Agent for testing default parameters", + model_name="gpt-4o", + max_loops=1, + ) + + # Create workflow with mostly default parameters + workflow = BatchedGridWorkflow(agents=[agent]) + + # Verify default values + assert workflow.name == "BatchedGridWorkflow" + assert ( + workflow.description + == "For every agent, run the task on a different task" + ) + assert workflow.max_loops == 1 + assert workflow.id is not None + + +def test_batched_grid_workflow_custom_id_and_name(): + """Test BatchedGridWorkflow with custom ID and name""" + agent = Agent( + agent_name="Custom-Agent", + agent_description="Agent for custom parameter testing", + model_name="gpt-4o", + max_loops=1, + ) + + custom_id = "custom-workflow-id-12345" + custom_name = "My-Custom-Workflow" + custom_description = "This is a custom workflow for testing" + + workflow = BatchedGridWorkflow( + id=custom_id, + name=custom_name, + description=custom_description, + agents=[agent], + max_loops=1, + ) + + assert workflow.id == custom_id + assert workflow.name == custom_name + assert workflow.description == custom_description + + +def test_batched_grid_workflow_large_scale(): + """Test BatchedGridWorkflow with multiple agents for scalability""" + # Create 6 agents to test larger scale operations + agents = [ + Agent( + agent_name=f"Scale-Agent-{i+1}", + agent_description=f"Agent {i+1} for scalability testing", + model_name="gpt-4o", + max_loops=1, + ) + for i in range(6) + ] + + workflow = BatchedGridWorkflow( + name="Large-Scale-Workflow", + description="Testing workflow scalability", + agents=agents, + max_loops=1, + ) + + tasks = [ + "Task 1: Market research for technology sector", + "Task 2: Competitor analysis in cloud computing", + "Task 3: Customer needs assessment", + "Task 4: Pricing strategy development", + "Task 5: Risk assessment and mitigation", + "Task 6: Implementation roadmap planning", + ] + + results = workflow.run(tasks) + + assert results is not None + assert isinstance(results, list) + assert len(results) == 1 + + +def test_batched_grid_workflow_sequential_runs(): + """Test running BatchedGridWorkflow multiple times sequentially""" + agent1 = Agent( + agent_name="Sequential-Agent-1", + agent_description="Agent for sequential execution testing", + model_name="gpt-4o", + max_loops=1, + ) + + agent2 = Agent( + agent_name="Sequential-Agent-2", + agent_description="Agent for sequential execution testing", + model_name="gpt-4o", + max_loops=1, + ) + + workflow = BatchedGridWorkflow( + name="Sequential-Runs-Workflow", + description="Testing multiple sequential runs", + agents=[agent1, agent2], + max_loops=1, + ) + + tasks_batch1 = [ + "Analyze Q1 business performance", + "Review Q1 customer feedback", + ] + + tasks_batch2 = [ + "Analyze Q2 business performance", + "Review Q2 customer feedback", + ] + + # Run workflow twice with different tasks + results1 = workflow.run(tasks_batch1) + results2 = workflow.run(tasks_batch2) + + assert results1 is not None + assert results2 is not None + assert isinstance(results1, list) + assert isinstance(results2, list) + + +def test_batched_grid_workflow_specialized_agents(): + """Test BatchedGridWorkflow with highly specialized agents""" + # Create agents with very specific roles + seo_agent = Agent( + agent_name="SEO-Specialist", + agent_description="Expert in search engine optimization and content strategy", + model_name="gpt-4o", + max_loops=1, + ) + + content_agent = Agent( + agent_name="Content-Creator", + agent_description="Expert in content creation and copywriting", + model_name="gpt-4o", + max_loops=1, + ) + + social_media_agent = Agent( + agent_name="Social-Media-Manager", + agent_description="Expert in social media marketing and engagement", + model_name="gpt-4o", + max_loops=1, + ) + + workflow = BatchedGridWorkflow( + name="Marketing-Campaign-Workflow", + description="Specialized workflow for marketing campaign execution", + agents=[seo_agent, content_agent, social_media_agent], + max_loops=1, + ) + + tasks = [ + "Develop an SEO strategy for a new e-commerce website selling sustainable products", + "Create compelling product descriptions and blog content for eco-friendly products", + "Design a social media campaign to promote sustainable living and green products", + ] + + results = workflow.run(tasks) + + assert results is not None + assert isinstance(results, list) + assert len(results) == 1 + assert results[0] is not None + + +def test_batched_grid_workflow_max_loops_ten(): + """Test BatchedGridWorkflow with max_loops set to 10""" + agent = Agent( + agent_name="Ten-Loop-Agent", + agent_description="Agent for testing 10 execution loops", + model_name="gpt-4o", + max_loops=1, + ) + + workflow = BatchedGridWorkflow( + name="Ten-Loop-Workflow", + description="Testing workflow with 10 loops", + agents=[agent], + max_loops=10, + ) + + tasks = ["Iteratively refine a machine learning model concept"] + + results = workflow.run(tasks) + + assert results is not None + assert isinstance(results, list) + assert len(results) == 10 # Verify all 10 loops executed + + +def test_batched_grid_workflow_output_consistency(): + """Test that BatchedGridWorkflow produces consistent output structure""" + agents = [ + Agent( + agent_name=f"Consistency-Agent-{i+1}", + agent_description=f"Agent {i+1} for output consistency testing", + model_name="gpt-4o", + max_loops=1, + ) + for i in range(3) + ] + + workflow = BatchedGridWorkflow( + name="Output-Consistency-Workflow", + description="Testing output structure consistency", + agents=agents, + max_loops=2, + ) + + tasks = [ + "Generate idea A", + "Generate idea B", + "Generate idea C", + ] + + results = workflow.run(tasks) + + # Check output structure consistency + assert isinstance(results, list) + assert len(results) == 2 # max_loops=2 + + for loop_result in results: + assert loop_result is not None + assert isinstance(loop_result, list) + assert len(loop_result) == 3 # Three agents diff --git a/tests/structs/test_board_of_directors_swarm.py b/tests/structs/test_board_of_directors_swarm.py index 3001a296..f6099bb9 100644 --- a/tests/structs/test_board_of_directors_swarm.py +++ b/tests/structs/test_board_of_directors_swarm.py @@ -153,9 +153,7 @@ def test_board_of_directors_swarm_error_handling(): ) try: - BoardOfDirectorsSwarm( - agents=[analyst], max_loops=0 - ) + BoardOfDirectorsSwarm(agents=[analyst], max_loops=0) assert ( False ), "Should have raised ValueError for invalid max_loops" diff --git a/tests/structs/test_moa.py b/tests/structs/test_moa.py index 490e8f65..5552d1f9 100644 --- a/tests/structs/test_moa.py +++ b/tests/structs/test_moa.py @@ -186,9 +186,7 @@ def test_mixture_of_agents_error_handling(): ) try: - MixtureOfAgents( - agents=[analyst], aggregator_system_prompt="" - ) + MixtureOfAgents(agents=[analyst], aggregator_system_prompt="") assert ( False ), "Should have raised ValueError for empty system prompt"