ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[NEW][Examples] [x402] [Fixed some examples]

Browse Source
pull/1172/head
Kye Gomez 1 week ago
parent a18717299f
commit 0e02c1226f
11 changed files with 3135 additions and 8 deletions
Show Stats Download Patch File Download Diff File

796
docs/swarms/structs/swarm_templates.md
Unescape View File

@ -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

102
examples/guides/x402_examples/README.md
Unescape View File

@ -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)

17
examples/guides/x402_examples/env.example
Unescape View File

@ -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

281
examples/guides/x402_examples/memecoin_agent_x402.py
Unescape View File

@ -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)

83
examples/guides/x402_examples/research_agent_x402_example.py
Unescape View File

@ -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)

4
examples/single_agent/tools/exa_search_agent.py
Unescape View File

@ -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],
)

3
swarms/structs/batched_grid_workflow.py
Unescape View File

@ -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.

1351
swarms/structs/swarm_templates.py
View File

File diff suppressed because it is too large Load Diff

498
tests/structs/test_batched_grid_workflow.py
Unescape View File

@ -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

4
tests/structs/test_board_of_directors_swarm.py
Unescape View File

@ -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"

4
tests/structs/test_moa.py
Unescape View File

@ -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"

Write Preview
Loading…
Cancel
Save