harshalmore31
3 weeks ago
parent
dd2b0c2a3c
commit
223d398a20
@ -1,165 +0,0 @@
|
||||
"""
|
||||
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)
|
||||
@ -1,244 +0,0 @@
|
||||
"""
|
||||
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