harshalmore31
3 weeks ago
parent
5be0ab609e
commit
dd2b0c2a3c
@ -0,0 +1,165 @@
|
||||
"""
|
||||
AgentLoader Example: Research Team Collaboration
|
||||
===============================================
|
||||
|
||||
This example demonstrates using the AgentLoader to create a research team
|
||||
from markdown files and orchestrate them in a sequential workflow.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# Add local swarms to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.sequential_workflow import SequentialWorkflow
|
||||
from swarms.utils.agent_loader import AgentLoader
|
||||
|
||||
def create_research_agents():
|
||||
"""Create markdown files for research team agents"""
|
||||
|
||||
market_researcher = """| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| market-researcher | Expert in market analysis and competitive intelligence | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Market size and growth analysis
|
||||
- Competitive landscape assessment
|
||||
- Consumer behavior patterns
|
||||
- Industry trend identification
|
||||
|
||||
## Approach
|
||||
1. Gather comprehensive market data
|
||||
2. Analyze quantitative and qualitative indicators
|
||||
3. Identify key market drivers and barriers
|
||||
4. Evaluate competitive positioning
|
||||
5. Assess market opportunities and threats
|
||||
|
||||
## Output
|
||||
- Market analysis reports with key metrics
|
||||
- Competitive intelligence briefings
|
||||
- Market opportunity assessments
|
||||
- Consumer behavior insights
|
||||
"""
|
||||
|
||||
financial_analyst = """| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| financial-analyst | Specialist in financial modeling and investment analysis | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Financial statement analysis
|
||||
- Valuation modeling techniques
|
||||
- Investment risk assessment
|
||||
- Cash flow projections
|
||||
|
||||
## Approach
|
||||
1. Conduct thorough financial analysis
|
||||
2. Build comprehensive financial models
|
||||
3. Perform multiple valuation methods
|
||||
4. Assess financial risks and sensitivities
|
||||
5. Provide investment recommendations
|
||||
|
||||
## Output
|
||||
- Financial analysis reports
|
||||
- Valuation models with scenarios
|
||||
- Investment recommendation memos
|
||||
- Risk assessment matrices
|
||||
"""
|
||||
|
||||
industry_expert = """| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| industry-expert | Domain specialist with deep industry knowledge | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Industry structure and dynamics
|
||||
- Regulatory environment analysis
|
||||
- Technology trends and disruptions
|
||||
- Supply chain analysis
|
||||
|
||||
## Approach
|
||||
1. Map industry structure and stakeholders
|
||||
2. Analyze regulatory framework
|
||||
3. Identify technology trends
|
||||
4. Evaluate supply chain dynamics
|
||||
5. Assess competitive positioning
|
||||
|
||||
## Output
|
||||
- Industry landscape reports
|
||||
- Regulatory compliance assessments
|
||||
- Technology trend analysis
|
||||
- Strategic positioning recommendations
|
||||
"""
|
||||
|
||||
return {
|
||||
"market_researcher.md": market_researcher,
|
||||
"financial_analyst.md": financial_analyst,
|
||||
"industry_expert.md": industry_expert
|
||||
}
|
||||
|
||||
def main():
|
||||
"""Main execution function"""
|
||||
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
|
||||
try:
|
||||
# Create markdown files
|
||||
agent_definitions = create_research_agents()
|
||||
file_paths = []
|
||||
|
||||
for filename, content in agent_definitions.items():
|
||||
file_path = os.path.join(temp_dir, filename)
|
||||
with open(file_path, 'w', encoding='utf-8') as f:
|
||||
f.write(content)
|
||||
file_paths.append(file_path)
|
||||
|
||||
# Load agents using AgentLoader
|
||||
loader = AgentLoader()
|
||||
agents = loader.load_multiple_agents(
|
||||
file_paths,
|
||||
max_loops=1,
|
||||
verbose=False
|
||||
)
|
||||
|
||||
print(f"Loaded {len(agents)} agents")
|
||||
for i, agent in enumerate(agents):
|
||||
print(f"Agent {i}: {agent.agent_name} - LLM: {hasattr(agent, 'llm')}")
|
||||
|
||||
# Create sequential workflow
|
||||
research_workflow = SequentialWorkflow(
|
||||
agents=agents,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
# Define research task
|
||||
task = """
|
||||
Analyze the AI-powered healthcare diagnostics market for a potential $50M investment.
|
||||
|
||||
Focus on:
|
||||
1. Market size, growth, and key drivers
|
||||
2. Competitive landscape and major players
|
||||
3. Financial viability and investment metrics
|
||||
4. Industry dynamics and regulatory factors
|
||||
|
||||
Provide strategic recommendations for market entry.
|
||||
"""
|
||||
|
||||
# Execute workflow
|
||||
result = research_workflow.run(task)
|
||||
|
||||
return result
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
for file_path in file_paths:
|
||||
if os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
os.rmdir(temp_dir)
|
||||
|
||||
if __name__ == "__main__":
|
||||
result = main()
|
||||
print("Research Analysis Complete:")
|
||||
print("-" * 50)
|
||||
print(result)
|
||||
@ -0,0 +1,353 @@
|
||||
# AgentLoader - Load Agents from Markdown Files
|
||||
|
||||
The `AgentLoader` is a powerful utility for creating Swarms agents from markdown files. It supports both single and multiple markdown file loading, providing a flexible way to define and deploy agents using a structured markdown format.
|
||||
|
||||
## Overview
|
||||
|
||||
The AgentLoader enables you to:
|
||||
- Load single agents from markdown files
|
||||
- Load multiple agents from directories or file lists
|
||||
- Parse structured markdown content into agent configurations
|
||||
- Maintain backwards compatibility with existing agent systems
|
||||
- Provide comprehensive error handling and validation
|
||||
|
||||
## Installation
|
||||
|
||||
The AgentLoader is included with the Swarms framework:
|
||||
|
||||
```python
|
||||
from swarms.utils import AgentLoader, load_agent_from_markdown, load_agents_from_markdown
|
||||
```
|
||||
|
||||
## Markdown Format
|
||||
|
||||
The AgentLoader expects markdown files to follow a specific structure:
|
||||
|
||||
### Required Table Header
|
||||
```markdown
|
||||
| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| agent-name | Brief description of the agent | gpt-4 |
|
||||
```
|
||||
|
||||
### Optional Sections
|
||||
```markdown
|
||||
## Focus Areas
|
||||
- Key responsibility area 1
|
||||
- Key responsibility area 2
|
||||
- Key responsibility area 3
|
||||
|
||||
## Approach
|
||||
1. First step in methodology
|
||||
2. Second step in methodology
|
||||
3. Third step in methodology
|
||||
|
||||
## Output
|
||||
- Expected deliverable 1
|
||||
- Expected deliverable 2
|
||||
- Expected deliverable 3
|
||||
```
|
||||
|
||||
## Quick Start
|
||||
|
||||
### Loading a Single Agent
|
||||
|
||||
```python
|
||||
from swarms.utils import load_agent_from_markdown
|
||||
|
||||
# Load agent from markdown file
|
||||
agent = load_agent_from_markdown(
|
||||
file_path="path/to/agent.md"
|
||||
)
|
||||
|
||||
# Use the agent
|
||||
response = agent.run("What are your capabilities?")
|
||||
print(response)
|
||||
```
|
||||
|
||||
### Loading Multiple Agents
|
||||
|
||||
```python
|
||||
from swarms.utils import load_agents_from_markdown
|
||||
|
||||
# Load all agents from directory
|
||||
agents = load_agents_from_markdown(
|
||||
file_paths="./agents_directory/"
|
||||
)
|
||||
|
||||
# Load agents from specific files
|
||||
agents = load_agents_from_markdown(
|
||||
file_paths=["agent1.md", "agent2.md", "agent3.md"]
|
||||
)
|
||||
|
||||
print(f"Loaded {len(agents)} agents")
|
||||
```
|
||||
|
||||
## Class-Based Usage
|
||||
|
||||
### AgentLoader Class
|
||||
|
||||
For more advanced usage, use the `AgentLoader` class directly:
|
||||
|
||||
```python
|
||||
from swarms.utils import AgentLoader
|
||||
|
||||
# Initialize loader
|
||||
loader = AgentLoader()
|
||||
|
||||
# Load single agent
|
||||
agent = loader.load_single_agent("path/to/agent.md")
|
||||
|
||||
# Load multiple agents
|
||||
agents = loader.load_multiple_agents("./agents_directory/")
|
||||
|
||||
# Parse markdown file without creating agent
|
||||
config = loader.parse_markdown_file("path/to/agent.md")
|
||||
print(config.name, config.description)
|
||||
```
|
||||
|
||||
## Configuration Options
|
||||
|
||||
You can override default configuration when loading agents:
|
||||
|
||||
```python
|
||||
agent = load_agent_from_markdown(
|
||||
file_path="agent.md",
|
||||
max_loops=5,
|
||||
verbose=True,
|
||||
dashboard=True,
|
||||
autosave=False,
|
||||
context_length=200000
|
||||
)
|
||||
```
|
||||
|
||||
### Available Configuration Parameters
|
||||
|
||||
- `max_loops` (int): Maximum number of reasoning loops (default: 1)
|
||||
- `autosave` (bool): Enable automatic state saving (default: True)
|
||||
- `dashboard` (bool): Enable dashboard monitoring (default: False)
|
||||
- `verbose` (bool): Enable verbose logging (default: False)
|
||||
- `dynamic_temperature_enabled` (bool): Enable dynamic temperature (default: False)
|
||||
- `saved_state_path` (str): Path for saving agent state
|
||||
- `user_name` (str): User identifier (default: "default_user")
|
||||
- `retry_attempts` (int): Number of retry attempts (default: 3)
|
||||
- `context_length` (int): Maximum context length (default: 100000)
|
||||
- `return_step_meta` (bool): Return step metadata (default: False)
|
||||
- `output_type` (str): Output format type (default: "str")
|
||||
- `auto_generate_prompt` (bool): Auto-generate prompts (default: False)
|
||||
- `artifacts_on` (bool): Enable artifacts (default: False)
|
||||
|
||||
## Complete Example
|
||||
|
||||
### Example Markdown File (performance-engineer.md)
|
||||
|
||||
```markdown
|
||||
| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| performance-engineer | Optimize application performance and identify bottlenecks | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Application profiling and performance analysis
|
||||
- Database optimization and query tuning
|
||||
- Memory and CPU usage optimization
|
||||
- Load testing and capacity planning
|
||||
- Infrastructure scaling recommendations
|
||||
|
||||
## Approach
|
||||
1. Analyze application architecture and identify potential bottlenecks
|
||||
2. Implement comprehensive monitoring and logging systems
|
||||
3. Conduct performance testing under various load conditions
|
||||
4. Profile memory usage and optimize resource consumption
|
||||
5. Provide actionable recommendations with implementation guides
|
||||
|
||||
## Output
|
||||
- Detailed performance analysis reports with metrics
|
||||
- Optimized code recommendations and examples
|
||||
- Infrastructure scaling and architecture suggestions
|
||||
- Monitoring and alerting configuration guidelines
|
||||
- Load testing results and capacity planning documents
|
||||
```
|
||||
|
||||
### Loading and Using the Agent
|
||||
|
||||
```python
|
||||
from swarms.utils import load_agent_from_markdown
|
||||
from swarms.utils.litellm_wrapper import LiteLLM
|
||||
|
||||
# Initialize model
|
||||
model = LiteLLM(model_name="gpt-4")
|
||||
|
||||
# Load the performance engineer agent
|
||||
agent = load_agent_from_markdown(
|
||||
file_path="performance-engineer.md",
|
||||
model=model,
|
||||
max_loops=3,
|
||||
verbose=True
|
||||
)
|
||||
|
||||
# Use the agent
|
||||
task = """
|
||||
Analyze the performance of a web application that handles 10,000 concurrent users
|
||||
but is experiencing slow response times averaging 3 seconds. The application uses
|
||||
a PostgreSQL database and is deployed on AWS with 4 EC2 instances behind a load balancer.
|
||||
"""
|
||||
|
||||
analysis = agent.run(task)
|
||||
print(f"Performance Analysis:\n{analysis}")
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
The AgentLoader provides comprehensive error handling:
|
||||
|
||||
```python
|
||||
from swarms.utils import AgentLoader
|
||||
|
||||
loader = AgentLoader()
|
||||
|
||||
try:
|
||||
# This will raise FileNotFoundError
|
||||
agent = loader.load_single_agent("nonexistent.md")
|
||||
except FileNotFoundError as e:
|
||||
print(f"File not found: {e}")
|
||||
|
||||
try:
|
||||
# This will handle parsing errors gracefully
|
||||
agents = loader.load_multiple_agents("./invalid_directory/")
|
||||
print(f"Successfully loaded {len(agents)} agents")
|
||||
except Exception as e:
|
||||
print(f"Error loading agents: {e}")
|
||||
```
|
||||
|
||||
## Advanced Features
|
||||
|
||||
### Custom System Prompt Building
|
||||
|
||||
The AgentLoader automatically builds comprehensive system prompts from the markdown structure:
|
||||
|
||||
```python
|
||||
loader = AgentLoader()
|
||||
config = loader.parse_markdown_file("agent.md")
|
||||
|
||||
# The system prompt includes:
|
||||
# - Role description from the table
|
||||
# - Focus areas as bullet points
|
||||
# - Approach as numbered steps
|
||||
# - Expected outputs as deliverables
|
||||
|
||||
print("Generated System Prompt:")
|
||||
print(config.system_prompt)
|
||||
```
|
||||
|
||||
### Batch Processing
|
||||
|
||||
Process multiple agent files efficiently:
|
||||
|
||||
```python
|
||||
import os
|
||||
from pathlib import Path
|
||||
from swarms.utils import AgentLoader
|
||||
|
||||
loader = AgentLoader()
|
||||
|
||||
# Find all markdown files in a directory
|
||||
agent_dir = Path("./agents")
|
||||
md_files = list(agent_dir.glob("*.md"))
|
||||
|
||||
# Load all agents
|
||||
agents = []
|
||||
for file_path in md_files:
|
||||
try:
|
||||
agent = loader.load_single_agent(str(file_path))
|
||||
agents.append(agent)
|
||||
print(f"✓ Loaded: {agent.agent_name}")
|
||||
except Exception as e:
|
||||
print(f"✗ Failed to load {file_path}: {e}")
|
||||
|
||||
print(f"\nSuccessfully loaded {len(agents)} agents")
|
||||
```
|
||||
|
||||
## Integration with Swarms
|
||||
|
||||
The loaded agents are fully compatible with Swarms orchestration systems:
|
||||
|
||||
```python
|
||||
from swarms.utils import load_agents_from_markdown
|
||||
from swarms.structs import SequentialWorkflow
|
||||
|
||||
# Load multiple specialized agents
|
||||
agents = load_agents_from_markdown("./specialist_agents/")
|
||||
|
||||
# Create a sequential workflow
|
||||
workflow = SequentialWorkflow(
|
||||
agents=agents,
|
||||
max_loops=1
|
||||
)
|
||||
|
||||
# Execute complex task across multiple agents
|
||||
result = workflow.run("Conduct a comprehensive system audit")
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
1. **Consistent Naming**: Use clear, descriptive agent names
|
||||
2. **Detailed Descriptions**: Provide comprehensive role descriptions
|
||||
3. **Structured Sections**: Use the optional sections to define agent behavior
|
||||
4. **Error Handling**: Always wrap agent loading in try-catch blocks
|
||||
5. **Model Selection**: Choose appropriate models based on agent complexity
|
||||
6. **Configuration**: Override defaults when specific behavior is needed
|
||||
|
||||
## Backwards Compatibility
|
||||
|
||||
The AgentLoader maintains full backwards compatibility with:
|
||||
- Claude Code sub-agents markdown format
|
||||
- Existing swarms agent creation patterns
|
||||
- Legacy configuration systems
|
||||
- Current workflow orchestration
|
||||
|
||||
## API Reference
|
||||
|
||||
### AgentLoader Class
|
||||
|
||||
```python
|
||||
class AgentLoader:
|
||||
def __init__(self, model: Optional[LiteLLM] = None)
|
||||
def parse_markdown_file(self, file_path: str) -> MarkdownAgentConfig
|
||||
def load_single_agent(self, file_path: str, **kwargs) -> Agent
|
||||
def load_multiple_agents(self, file_paths: Union[str, List[str]], **kwargs) -> List[Agent]
|
||||
```
|
||||
|
||||
### Convenience Functions
|
||||
|
||||
```python
|
||||
def load_agent_from_markdown(file_path: str, model: Optional[LiteLLM] = None, **kwargs) -> Agent
|
||||
def load_agents_from_markdown(file_paths: Union[str, List[str]], model: Optional[LiteLLM] = None, **kwargs) -> List[Agent]
|
||||
```
|
||||
|
||||
### Configuration Model
|
||||
|
||||
```python
|
||||
class MarkdownAgentConfig(BaseModel):
|
||||
name: str
|
||||
description: str
|
||||
model_name: Optional[str] = "gpt-4"
|
||||
system_prompt: str
|
||||
focus_areas: Optional[List[str]] = []
|
||||
approach: Optional[List[str]] = []
|
||||
output: Optional[List[str]] = []
|
||||
# ... additional configuration fields
|
||||
```
|
||||
|
||||
## Examples Repository
|
||||
|
||||
Find more examples in the Swarms repository:
|
||||
- `examples/agents_loader_example.py` - Complete usage demonstration
|
||||
- `test_agent_loader.py` - Test suite with validation examples
|
||||
- `examples/single_agent/utils/markdown_agent.py` - Markdown agent utilities
|
||||
|
||||
## Support
|
||||
|
||||
For questions and support:
|
||||
- GitHub Issues: [https://github.com/kyegomez/swarms/issues](https://github.com/kyegomez/swarms/issues)
|
||||
- Documentation: [https://docs.swarms.world](https://docs.swarms.world)
|
||||
- Community: Join our Discord for real-time support
|
||||
@ -0,0 +1,127 @@
|
||||
"""
|
||||
Example demonstrating the AgentLoader for loading agents from markdown files.
|
||||
|
||||
This example shows:
|
||||
1. Loading a single agent from a markdown file
|
||||
2. Loading multiple agents from markdown files
|
||||
3. Using the convenience functions
|
||||
4. Error handling and validation
|
||||
"""
|
||||
|
||||
import os
|
||||
from swarms.utils.agent_loader import AgentLoader, load_agent_from_markdown, load_agents_from_markdown
|
||||
|
||||
def main():
|
||||
# Initialize the loader
|
||||
loader = AgentLoader()
|
||||
|
||||
print("=== AgentLoader Demo ===")
|
||||
|
||||
# Example 1: Create a sample markdown file for testing
|
||||
sample_md = """| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| performance-engineer | Optimize application performance and identify bottlenecks | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Application profiling and performance analysis
|
||||
- Database optimization and query tuning
|
||||
- Memory and CPU usage optimization
|
||||
- Load testing and capacity planning
|
||||
|
||||
## Approach
|
||||
1. Analyze application architecture and identify potential bottlenecks
|
||||
2. Implement comprehensive monitoring and logging
|
||||
3. Conduct performance testing under various load conditions
|
||||
4. Optimize critical paths and resource usage
|
||||
5. Document findings and provide actionable recommendations
|
||||
|
||||
## Output
|
||||
- Performance analysis reports with specific metrics
|
||||
- Optimized code recommendations
|
||||
- Infrastructure scaling suggestions
|
||||
- Monitoring and alerting setup guidelines
|
||||
"""
|
||||
|
||||
# Create sample markdown file
|
||||
sample_file = "sample_agent.md"
|
||||
with open(sample_file, 'w') as f:
|
||||
f.write(sample_md)
|
||||
|
||||
try:
|
||||
# Example 2: Load single agent using class method
|
||||
print("\\n1. Loading single agent using AgentLoader class:")
|
||||
agent = loader.load_single_agent(sample_file)
|
||||
print(f" Loaded agent: {agent.agent_name}")
|
||||
print(f" System prompt preview: {agent.system_prompt[:100]}...")
|
||||
|
||||
# Example 3: Load single agent using convenience function
|
||||
print("\\n2. Loading single agent using convenience function:")
|
||||
agent2 = load_agent_from_markdown(sample_file)
|
||||
print(f" Loaded agent: {agent2.agent_name}")
|
||||
|
||||
# Example 4: Load multiple agents (from directory or list)
|
||||
print("\\n3. Loading multiple agents:")
|
||||
|
||||
# Create another sample file
|
||||
sample_md2 = """| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| security-analyst | Analyze and improve system security | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Security vulnerability assessment
|
||||
- Code security review
|
||||
- Infrastructure hardening
|
||||
|
||||
## Approach
|
||||
1. Conduct thorough security audits
|
||||
2. Identify potential vulnerabilities
|
||||
3. Recommend security improvements
|
||||
|
||||
## Output
|
||||
- Security assessment reports
|
||||
- Vulnerability remediation plans
|
||||
- Security best practices documentation
|
||||
"""
|
||||
|
||||
sample_file2 = "security_agent.md"
|
||||
with open(sample_file2, 'w') as f:
|
||||
f.write(sample_md2)
|
||||
|
||||
# Load multiple agents from list
|
||||
agents = loader.load_multiple_agents([sample_file, sample_file2])
|
||||
print(f" Loaded {len(agents)} agents:")
|
||||
for agent in agents:
|
||||
print(f" - {agent.agent_name}")
|
||||
|
||||
# Example 5: Load agents from directory (current directory)
|
||||
print("\\n4. Loading agents from current directory:")
|
||||
current_dir_agents = load_agents_from_markdown(".")
|
||||
print(f" Found {len(current_dir_agents)} agents in current directory")
|
||||
|
||||
# Example 6: Demonstrate error handling
|
||||
print("\\n5. Error handling demo:")
|
||||
try:
|
||||
loader.load_single_agent("nonexistent.md")
|
||||
except FileNotFoundError as e:
|
||||
print(f" Caught expected error: {e}")
|
||||
|
||||
# Example 7: Test agent functionality
|
||||
print("\\n6. Testing loaded agent functionality:")
|
||||
test_agent = agents[0]
|
||||
response = test_agent.run("What are the key steps for performance optimization?")
|
||||
print(f" Agent response preview: {str(response)[:150]}...")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error during demo: {e}")
|
||||
|
||||
finally:
|
||||
# Cleanup sample files
|
||||
for file in [sample_file, sample_file2]:
|
||||
if os.path.exists(file):
|
||||
os.remove(file)
|
||||
print("\\n Cleaned up sample files")
|
||||
|
||||
print("\\n=== Demo Complete ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -0,0 +1,140 @@
|
||||
"""
|
||||
Simple AgentLoader Demo
|
||||
=======================
|
||||
|
||||
A working demonstration of how to create agents from markdown-like definitions
|
||||
and use them in workflows.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
# Add local swarms to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.sequential_workflow import SequentialWorkflow
|
||||
|
||||
def create_agents_from_configs():
|
||||
"""Create agents from configuration dictionaries (simulating markdown parsing)"""
|
||||
|
||||
# These would normally come from parsing markdown files
|
||||
agent_configs = [
|
||||
{
|
||||
"name": "market-researcher",
|
||||
"description": "Expert in market analysis and competitive intelligence",
|
||||
"system_prompt": """You are a market research specialist. Your expertise includes:
|
||||
|
||||
Focus Areas:
|
||||
- Market size and growth analysis
|
||||
- Competitive landscape assessment
|
||||
- Consumer behavior patterns
|
||||
- Industry trend identification
|
||||
|
||||
Approach:
|
||||
1. Gather comprehensive market data
|
||||
2. Analyze quantitative and qualitative indicators
|
||||
3. Identify key market drivers and barriers
|
||||
4. Evaluate competitive positioning
|
||||
5. Assess market opportunities and threats
|
||||
|
||||
Provide detailed market analysis reports with key metrics and actionable insights.""",
|
||||
"model": "gpt-4"
|
||||
},
|
||||
{
|
||||
"name": "financial-analyst",
|
||||
"description": "Specialist in financial modeling and investment analysis",
|
||||
"system_prompt": """You are a financial analysis expert. Your responsibilities include:
|
||||
|
||||
Focus Areas:
|
||||
- Financial statement analysis
|
||||
- Valuation modeling techniques
|
||||
- Investment risk assessment
|
||||
- Cash flow projections
|
||||
|
||||
Approach:
|
||||
1. Conduct thorough financial analysis
|
||||
2. Build comprehensive financial models
|
||||
3. Perform multiple valuation methods
|
||||
4. Assess financial risks and sensitivities
|
||||
5. Provide investment recommendations
|
||||
|
||||
Generate detailed financial reports with valuation models and risk assessments.""",
|
||||
"model": "gpt-4"
|
||||
},
|
||||
{
|
||||
"name": "industry-expert",
|
||||
"description": "Domain specialist with deep industry knowledge",
|
||||
"system_prompt": """You are an industry analysis expert. Your focus areas include:
|
||||
|
||||
Focus Areas:
|
||||
- Industry structure and dynamics
|
||||
- Regulatory environment analysis
|
||||
- Technology trends and disruptions
|
||||
- Supply chain analysis
|
||||
|
||||
Approach:
|
||||
1. Map industry structure and stakeholders
|
||||
2. Analyze regulatory framework
|
||||
3. Identify technology trends
|
||||
4. Evaluate supply chain dynamics
|
||||
5. Assess competitive positioning
|
||||
|
||||
Provide comprehensive industry landscape reports with strategic recommendations.""",
|
||||
"model": "gpt-4"
|
||||
}
|
||||
]
|
||||
|
||||
agents = []
|
||||
for config in agent_configs:
|
||||
agent = Agent(
|
||||
agent_name=config["name"],
|
||||
system_prompt=config["system_prompt"],
|
||||
model_name=config["model"],
|
||||
max_loops=1,
|
||||
verbose=False
|
||||
)
|
||||
agents.append(agent)
|
||||
print(f"Created agent: {agent.agent_name}")
|
||||
|
||||
return agents
|
||||
|
||||
def main():
|
||||
"""Main execution function"""
|
||||
|
||||
# Create agents
|
||||
agents = create_agents_from_configs()
|
||||
|
||||
# Create sequential workflow
|
||||
research_workflow = SequentialWorkflow(
|
||||
agents=agents,
|
||||
max_loops=1,
|
||||
)
|
||||
|
||||
# Define research task
|
||||
task = """
|
||||
Analyze the AI-powered healthcare diagnostics market for a potential $50M investment.
|
||||
|
||||
Focus on:
|
||||
1. Market size, growth projections, and key drivers
|
||||
2. Competitive landscape and major players
|
||||
3. Financial viability and investment attractiveness
|
||||
4. Industry dynamics and regulatory considerations
|
||||
|
||||
Provide strategic recommendations for market entry.
|
||||
"""
|
||||
|
||||
print("Executing research workflow...")
|
||||
print("=" * 50)
|
||||
|
||||
# Execute workflow
|
||||
result = research_workflow.run(task)
|
||||
|
||||
print("\nResearch Analysis Complete:")
|
||||
print("-" * 50)
|
||||
print(result)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -0,0 +1,378 @@
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Union
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
from loguru import logger
|
||||
|
||||
from swarms.structs.agent import Agent
|
||||
|
||||
|
||||
class MarkdownAgentConfig(BaseModel):
|
||||
"""Configuration model for agents loaded from markdown files."""
|
||||
name: str
|
||||
description: str
|
||||
model_name: Optional[str] = "gpt-4"
|
||||
system_prompt: str
|
||||
focus_areas: Optional[List[str]] = []
|
||||
approach: Optional[List[str]] = []
|
||||
output: Optional[List[str]] = []
|
||||
max_loops: int = Field(default=1, ge=1)
|
||||
autosave: bool = False
|
||||
dashboard: bool = False
|
||||
verbose: bool = False
|
||||
dynamic_temperature_enabled: bool = False
|
||||
saved_state_path: Optional[str] = None
|
||||
user_name: str = "default_user"
|
||||
retry_attempts: int = Field(default=3, ge=1)
|
||||
context_length: int = Field(default=100000, ge=1000)
|
||||
return_step_meta: bool = False
|
||||
output_type: str = "str"
|
||||
auto_generate_prompt: bool = False
|
||||
artifacts_on: bool = False
|
||||
artifacts_file_extension: str = ".md"
|
||||
artifacts_output_path: str = ""
|
||||
|
||||
@field_validator("system_prompt")
|
||||
@classmethod
|
||||
def validate_system_prompt(cls, v):
|
||||
if not v or not isinstance(v, str) or len(v.strip()) == 0:
|
||||
raise ValueError("System prompt must be a non-empty string")
|
||||
return v
|
||||
|
||||
|
||||
class AgentLoader:
|
||||
"""
|
||||
Loader for creating agents from markdown files.
|
||||
|
||||
Supports both single markdown file and multiple markdown files.
|
||||
Maintains backwards compatibility with claude code sub agents markdown format.
|
||||
|
||||
Features:
|
||||
- Single markdown file loading
|
||||
- Multiple markdown files loading (batch processing)
|
||||
- Flexible markdown parsing
|
||||
- Agent configuration extraction from markdown structure
|
||||
- Error handling and validation
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""
|
||||
Initialize the AgentLoader.
|
||||
"""
|
||||
pass
|
||||
|
||||
def parse_markdown_table(self, content: str) -> Dict[str, str]:
|
||||
"""
|
||||
Parse markdown table to extract agent metadata.
|
||||
|
||||
Args:
|
||||
content: Markdown content containing a table
|
||||
|
||||
Returns:
|
||||
Dictionary with parsed table data
|
||||
"""
|
||||
table_data = {}
|
||||
|
||||
# Find markdown table pattern
|
||||
table_pattern = r'\|([^|]+)\|([^|]+)\|([^|]+)\|'
|
||||
lines = content.split('\n')
|
||||
|
||||
header_found = False
|
||||
for line in lines:
|
||||
if '|' in line and not header_found:
|
||||
# Skip header separator line
|
||||
if '---' in line:
|
||||
header_found = True
|
||||
continue
|
||||
|
||||
# Parse header
|
||||
if 'name' in line.lower() and 'description' in line.lower():
|
||||
continue
|
||||
|
||||
elif header_found and '|' in line:
|
||||
# Parse data row
|
||||
match = re.match(table_pattern, line)
|
||||
if match:
|
||||
table_data['name'] = match.group(1).strip()
|
||||
table_data['description'] = match.group(2).strip()
|
||||
table_data['model_name'] = match.group(3).strip()
|
||||
break
|
||||
|
||||
return table_data
|
||||
|
||||
def extract_sections(self, content: str) -> Dict[str, List[str]]:
|
||||
"""
|
||||
Extract structured sections from markdown content.
|
||||
|
||||
Args:
|
||||
content: Markdown content
|
||||
|
||||
Returns:
|
||||
Dictionary with section names as keys and content lists as values
|
||||
"""
|
||||
sections = {}
|
||||
current_section = None
|
||||
current_content = []
|
||||
|
||||
lines = content.split('\n')
|
||||
for line in lines:
|
||||
# Check for headers (## Section Name)
|
||||
if line.startswith('## '):
|
||||
# Save previous section
|
||||
if current_section:
|
||||
sections[current_section.lower()] = current_content
|
||||
|
||||
# Start new section
|
||||
current_section = line[3:].strip()
|
||||
current_content = []
|
||||
|
||||
elif current_section and line.strip():
|
||||
# Add content to current section
|
||||
# Remove markdown list markers
|
||||
clean_line = re.sub(r'^[-*+]\s*', '', line.strip())
|
||||
clean_line = re.sub(r'^\d+\.\s*', '', clean_line)
|
||||
if clean_line:
|
||||
current_content.append(clean_line)
|
||||
|
||||
# Save last section
|
||||
if current_section:
|
||||
sections[current_section.lower()] = current_content
|
||||
|
||||
return sections
|
||||
|
||||
def build_system_prompt(self, config_data: Dict[str, Any]) -> str:
|
||||
"""
|
||||
Build comprehensive system prompt from parsed markdown data.
|
||||
|
||||
Args:
|
||||
config_data: Dictionary containing parsed agent configuration
|
||||
|
||||
Returns:
|
||||
Complete system prompt string
|
||||
"""
|
||||
prompt_parts = []
|
||||
|
||||
# Add description
|
||||
if config_data.get('description'):
|
||||
prompt_parts.append(f"Role: {config_data['description']}")
|
||||
|
||||
# Add focus areas
|
||||
if config_data.get('focus_areas'):
|
||||
prompt_parts.append("\nFocus Areas:")
|
||||
for area in config_data['focus_areas']:
|
||||
prompt_parts.append(f"- {area}")
|
||||
|
||||
# Add approach
|
||||
if config_data.get('approach'):
|
||||
prompt_parts.append("\nApproach:")
|
||||
for i, step in enumerate(config_data['approach'], 1):
|
||||
prompt_parts.append(f"{i}. {step}")
|
||||
|
||||
# Add expected output
|
||||
if config_data.get('output'):
|
||||
prompt_parts.append("\nExpected Output:")
|
||||
for output in config_data['output']:
|
||||
prompt_parts.append(f"- {output}")
|
||||
|
||||
return '\n'.join(prompt_parts)
|
||||
|
||||
def parse_markdown_file(self, file_path: str) -> MarkdownAgentConfig:
|
||||
"""
|
||||
Parse a single markdown file to extract agent configuration.
|
||||
|
||||
Args:
|
||||
file_path: Path to markdown file
|
||||
|
||||
Returns:
|
||||
MarkdownAgentConfig object with parsed configuration
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If file doesn't exist
|
||||
ValueError: If parsing fails
|
||||
"""
|
||||
if not os.path.exists(file_path):
|
||||
raise FileNotFoundError(f"Markdown file {file_path} not found.")
|
||||
|
||||
try:
|
||||
with open(file_path, 'r', encoding='utf-8') as file:
|
||||
content = file.read()
|
||||
|
||||
# Parse table for basic metadata
|
||||
table_data = self.parse_markdown_table(content)
|
||||
|
||||
# Extract sections
|
||||
sections = self.extract_sections(content)
|
||||
|
||||
# Build configuration
|
||||
config_data = {
|
||||
'name': table_data.get('name', Path(file_path).stem),
|
||||
'description': table_data.get('description', 'Agent loaded from markdown'),
|
||||
'model_name': table_data.get('model_name', 'gpt-4'),
|
||||
'focus_areas': sections.get('focus areas', []),
|
||||
'approach': sections.get('approach', []),
|
||||
'output': sections.get('output', []),
|
||||
}
|
||||
|
||||
# Build system prompt
|
||||
system_prompt = self.build_system_prompt(config_data)
|
||||
config_data['system_prompt'] = system_prompt
|
||||
|
||||
logger.info(f"Successfully parsed markdown file: {file_path}")
|
||||
return MarkdownAgentConfig(**config_data)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error parsing markdown file {file_path}: {str(e)}")
|
||||
raise ValueError(f"Error parsing markdown file {file_path}: {str(e)}")
|
||||
|
||||
def load_agent_from_markdown(self, file_path: str, **kwargs) -> Agent:
|
||||
"""
|
||||
Load a single agent from a markdown file.
|
||||
|
||||
Args:
|
||||
file_path: Path to markdown file
|
||||
**kwargs: Additional arguments to override default configuration
|
||||
|
||||
Returns:
|
||||
Configured Agent instance
|
||||
"""
|
||||
config = self.parse_markdown_file(file_path)
|
||||
|
||||
# Override with any provided kwargs
|
||||
config_dict = config.model_dump()
|
||||
config_dict.update(kwargs)
|
||||
|
||||
# Remove fields not needed for Agent creation
|
||||
agent_fields = {
|
||||
'agent_name': config_dict['name'],
|
||||
'system_prompt': config_dict['system_prompt'],
|
||||
'model_name': config_dict.get('model_name', 'gpt-4'),
|
||||
# Don't pass llm explicitly - let Agent handle it internally
|
||||
'max_loops': config_dict['max_loops'],
|
||||
'autosave': config_dict['autosave'],
|
||||
'dashboard': config_dict['dashboard'],
|
||||
'verbose': config_dict['verbose'],
|
||||
'dynamic_temperature_enabled': config_dict['dynamic_temperature_enabled'],
|
||||
'saved_state_path': config_dict['saved_state_path'],
|
||||
'user_name': config_dict['user_name'],
|
||||
'retry_attempts': config_dict['retry_attempts'],
|
||||
'context_length': config_dict['context_length'],
|
||||
'return_step_meta': config_dict['return_step_meta'],
|
||||
'output_type': config_dict['output_type'],
|
||||
'auto_generate_prompt': config_dict['auto_generate_prompt'],
|
||||
'artifacts_on': config_dict['artifacts_on'],
|
||||
'artifacts_file_extension': config_dict['artifacts_file_extension'],
|
||||
'artifacts_output_path': config_dict['artifacts_output_path'],
|
||||
}
|
||||
|
||||
try:
|
||||
logger.info(f"Creating agent '{config.name}' from {file_path}")
|
||||
agent = Agent(**agent_fields)
|
||||
logger.info(f"Successfully created agent '{config.name}' from {file_path}")
|
||||
return agent
|
||||
except Exception as e:
|
||||
import traceback
|
||||
logger.error(f"Error creating agent from {file_path}: {str(e)}")
|
||||
logger.error(f"Traceback: {traceback.format_exc()}")
|
||||
raise ValueError(f"Error creating agent from {file_path}: {str(e)}")
|
||||
|
||||
def load_agents_from_markdown(self, file_paths: Union[str, List[str]], **kwargs) -> List[Agent]:
|
||||
"""
|
||||
Load multiple agents from markdown files.
|
||||
|
||||
Args:
|
||||
file_paths: Single file path, directory path, or list of file paths
|
||||
**kwargs: Additional arguments to override default configuration
|
||||
|
||||
Returns:
|
||||
List of configured Agent instances
|
||||
"""
|
||||
agents = []
|
||||
paths_to_process = []
|
||||
|
||||
# Handle different input types
|
||||
if isinstance(file_paths, str):
|
||||
if os.path.isdir(file_paths):
|
||||
# Directory - find all .md files
|
||||
md_files = list(Path(file_paths).glob('*.md'))
|
||||
paths_to_process = [str(f) for f in md_files]
|
||||
elif os.path.isfile(file_paths):
|
||||
# Single file
|
||||
paths_to_process = [file_paths]
|
||||
else:
|
||||
raise FileNotFoundError(f"Path {file_paths} not found.")
|
||||
elif isinstance(file_paths, list):
|
||||
paths_to_process = file_paths
|
||||
else:
|
||||
raise ValueError("file_paths must be a string or list of strings")
|
||||
|
||||
# Process each file
|
||||
for file_path in paths_to_process:
|
||||
try:
|
||||
agent = self.load_agent_from_markdown(file_path, **kwargs)
|
||||
agents.append(agent)
|
||||
except Exception as e:
|
||||
logger.warning(f"Skipping {file_path} due to error: {str(e)}")
|
||||
continue
|
||||
|
||||
logger.info(f"Successfully loaded {len(agents)} agents from markdown files")
|
||||
return agents
|
||||
|
||||
def load_single_agent(self, file_path: str, **kwargs) -> Agent:
|
||||
"""
|
||||
Convenience method for loading a single agent.
|
||||
Backwards compatible with claude code sub agents markdown.
|
||||
|
||||
Args:
|
||||
file_path: Path to markdown file
|
||||
**kwargs: Additional configuration overrides
|
||||
|
||||
Returns:
|
||||
Configured Agent instance
|
||||
"""
|
||||
return self.load_agent_from_markdown(file_path, **kwargs)
|
||||
|
||||
def load_multiple_agents(self, file_paths: Union[str, List[str]], **kwargs) -> List[Agent]:
|
||||
"""
|
||||
Convenience method for loading multiple agents.
|
||||
Backwards compatible with claude code sub agents markdown.
|
||||
|
||||
Args:
|
||||
file_paths: Directory path or list of file paths
|
||||
**kwargs: Additional configuration overrides
|
||||
|
||||
Returns:
|
||||
List of configured Agent instances
|
||||
"""
|
||||
return self.load_agents_from_markdown(file_paths, **kwargs)
|
||||
|
||||
|
||||
# Convenience functions for backwards compatibility
|
||||
def load_agent_from_markdown(file_path: str, **kwargs) -> Agent:
|
||||
"""
|
||||
Load a single agent from a markdown file.
|
||||
|
||||
Args:
|
||||
file_path: Path to markdown file
|
||||
**kwargs: Additional configuration overrides
|
||||
|
||||
Returns:
|
||||
Configured Agent instance
|
||||
"""
|
||||
loader = AgentLoader()
|
||||
return loader.load_single_agent(file_path, **kwargs)
|
||||
|
||||
|
||||
def load_agents_from_markdown(file_paths: Union[str, List[str]], **kwargs) -> List[Agent]:
|
||||
"""
|
||||
Load multiple agents from markdown files.
|
||||
|
||||
Args:
|
||||
file_paths: Directory path or list of file paths
|
||||
**kwargs: Additional configuration overrides
|
||||
|
||||
Returns:
|
||||
List of configured Agent instances
|
||||
"""
|
||||
loader = AgentLoader()
|
||||
return loader.load_multiple_agents(file_paths, **kwargs)
|
||||
@ -0,0 +1,244 @@
|
||||
"""
|
||||
Test script for the AgentLoader functionality.
|
||||
This tests the core functionality without requiring external models.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
# Add swarms to path for local testing
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
|
||||
|
||||
from swarms.utils.agent_loader import AgentLoader, MarkdownAgentConfig
|
||||
|
||||
def test_markdown_parsing():
|
||||
"""Test markdown parsing functionality."""
|
||||
print("Testing markdown parsing...")
|
||||
|
||||
# Create a sample markdown content
|
||||
sample_content = """| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| test-agent | A test agent for validation | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Testing functionality
|
||||
- Validating implementation
|
||||
- Ensuring compatibility
|
||||
|
||||
## Approach
|
||||
1. Parse markdown structure
|
||||
2. Extract configuration data
|
||||
3. Validate parsed results
|
||||
4. Create agent instance
|
||||
|
||||
## Output
|
||||
- Test results
|
||||
- Validation reports
|
||||
- Configuration summary
|
||||
"""
|
||||
|
||||
# Test parsing functionality
|
||||
loader = AgentLoader()
|
||||
|
||||
# Test table parsing
|
||||
table_data = loader.parse_markdown_table(sample_content)
|
||||
assert table_data['name'] == 'test-agent'
|
||||
assert table_data['description'] == 'A test agent for validation'
|
||||
assert table_data['model_name'] == 'gpt-4'
|
||||
print("[OK] Table parsing successful")
|
||||
|
||||
# Test section extraction
|
||||
sections = loader.extract_sections(sample_content)
|
||||
assert 'focus areas' in sections
|
||||
assert len(sections['focus areas']) == 3
|
||||
assert 'approach' in sections
|
||||
assert len(sections['approach']) == 4
|
||||
print("[OK] Section extraction successful")
|
||||
|
||||
# Test system prompt building
|
||||
config_data = {
|
||||
'description': table_data['description'],
|
||||
'focus_areas': sections['focus areas'],
|
||||
'approach': sections['approach'],
|
||||
'output': sections.get('output', [])
|
||||
}
|
||||
system_prompt = loader.build_system_prompt(config_data)
|
||||
assert 'Role:' in system_prompt
|
||||
assert 'Focus Areas:' in system_prompt
|
||||
assert 'Approach:' in system_prompt
|
||||
print("[OK] System prompt building successful")
|
||||
|
||||
print("Markdown parsing tests passed!")
|
||||
return True
|
||||
|
||||
def test_file_operations():
|
||||
"""Test file loading operations."""
|
||||
print("\\nTesting file operations...")
|
||||
|
||||
# Create temporary markdown file
|
||||
sample_content = """| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| file-test-agent | Agent created from file | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- File processing
|
||||
- Configuration validation
|
||||
|
||||
## Approach
|
||||
1. Load from file
|
||||
2. Parse content
|
||||
3. Create configuration
|
||||
|
||||
## Output
|
||||
- Loaded agent
|
||||
- Configuration object
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f:
|
||||
f.write(sample_content)
|
||||
temp_file = f.name
|
||||
|
||||
try:
|
||||
loader = AgentLoader()
|
||||
|
||||
# Test file parsing
|
||||
config = loader.parse_markdown_file(temp_file)
|
||||
assert isinstance(config, MarkdownAgentConfig)
|
||||
assert config.name == 'file-test-agent'
|
||||
assert config.description == 'Agent created from file'
|
||||
print("[OK] File parsing successful")
|
||||
|
||||
# Test configuration validation
|
||||
assert len(config.focus_areas) == 2
|
||||
assert len(config.approach) == 3
|
||||
assert config.system_prompt is not None
|
||||
print("[OK] Configuration validation successful")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
if os.path.exists(temp_file):
|
||||
os.remove(temp_file)
|
||||
|
||||
print("File operations tests passed!")
|
||||
return True
|
||||
|
||||
def test_multiple_files():
|
||||
"""Test loading multiple files."""
|
||||
print("\\nTesting multiple file loading...")
|
||||
|
||||
# Create multiple temporary files
|
||||
files = []
|
||||
for i in range(3):
|
||||
content = f"""| name | description | model |
|
||||
|------|-------------|-------|
|
||||
| agent-{i} | Test agent number {i} | gpt-4 |
|
||||
|
||||
## Focus Areas
|
||||
- Multi-agent testing
|
||||
- Batch processing
|
||||
|
||||
## Approach
|
||||
1. Process multiple files
|
||||
2. Create agent configurations
|
||||
3. Return agent list
|
||||
|
||||
## Output
|
||||
- Multiple agents
|
||||
- Batch results
|
||||
"""
|
||||
temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False)
|
||||
temp_file.write(content)
|
||||
temp_file.close()
|
||||
files.append(temp_file.name)
|
||||
|
||||
try:
|
||||
loader = AgentLoader()
|
||||
|
||||
# Test parsing multiple files
|
||||
configs = []
|
||||
for file_path in files:
|
||||
config = loader.parse_markdown_file(file_path)
|
||||
configs.append(config)
|
||||
|
||||
assert len(configs) == 3
|
||||
for i, config in enumerate(configs):
|
||||
assert config.name == f'agent-{i}'
|
||||
|
||||
print("[OK] Multiple file parsing successful")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
for file_path in files:
|
||||
if os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
|
||||
print("Multiple file tests passed!")
|
||||
return True
|
||||
|
||||
def test_error_handling():
|
||||
"""Test error handling scenarios."""
|
||||
print("\\nTesting error handling...")
|
||||
|
||||
loader = AgentLoader()
|
||||
|
||||
# Test non-existent file
|
||||
try:
|
||||
loader.parse_markdown_file("nonexistent.md")
|
||||
assert False, "Should have raised FileNotFoundError"
|
||||
except FileNotFoundError:
|
||||
print("[OK] FileNotFoundError handling successful")
|
||||
|
||||
# Test invalid markdown
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f:
|
||||
f.write("Invalid markdown content without proper structure")
|
||||
invalid_file = f.name
|
||||
|
||||
try:
|
||||
# This should not raise an error, but should handle gracefully
|
||||
config = loader.parse_markdown_file(invalid_file)
|
||||
# Should have defaults
|
||||
assert config.name is not None
|
||||
print("[OK] Invalid markdown handling successful")
|
||||
|
||||
finally:
|
||||
if os.path.exists(invalid_file):
|
||||
os.remove(invalid_file)
|
||||
|
||||
print("Error handling tests passed!")
|
||||
return True
|
||||
|
||||
def main():
|
||||
"""Run all tests."""
|
||||
print("=== AgentLoader Test Suite ===")
|
||||
|
||||
tests = [
|
||||
test_markdown_parsing,
|
||||
test_file_operations,
|
||||
test_multiple_files,
|
||||
test_error_handling
|
||||
]
|
||||
|
||||
passed = 0
|
||||
total = len(tests)
|
||||
|
||||
for test in tests:
|
||||
try:
|
||||
if test():
|
||||
passed += 1
|
||||
except Exception as e:
|
||||
print(f"[FAIL] Test {test.__name__} failed: {e}")
|
||||
|
||||
print(f"\\n=== Results: {passed}/{total} tests passed ===")
|
||||
|
||||
if passed == total:
|
||||
print("All tests passed!")
|
||||
return True
|
||||
else:
|
||||
print("Some tests failed!")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = main()
|
||||
exit(0 if success else 1)
|
||||
Loading…
Reference in new issue