From dd2b0c2a3c6fcc66e0d34426e399d08807a45306 Mon Sep 17 00:00:00 2001 From: harshalmore31 Date: Sun, 10 Aug 2025 20:51:09 +0530 Subject: [PATCH] added agent_loader --- agent_loader_research_team.py | 165 ++++++++ docs/swarms/utils/agent_loader.md | 353 ++++++++++++++++ examples/agents_loader_example.py | 127 ++++++ .../multi_agent/simple_agent_loader_demo.py | 140 +++++++ swarms/utils/agent_loader.py | 378 ++++++++++++++++++ test_agent_loader.py | 244 +++++++++++ 6 files changed, 1407 insertions(+) create mode 100644 agent_loader_research_team.py create mode 100644 docs/swarms/utils/agent_loader.md create mode 100644 examples/agents_loader_example.py create mode 100644 examples/multi_agent/simple_agent_loader_demo.py create mode 100644 swarms/utils/agent_loader.py create mode 100644 test_agent_loader.py diff --git a/agent_loader_research_team.py b/agent_loader_research_team.py new file mode 100644 index 00000000..1ae02301 --- /dev/null +++ b/agent_loader_research_team.py @@ -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) \ No newline at end of file diff --git a/docs/swarms/utils/agent_loader.md b/docs/swarms/utils/agent_loader.md new file mode 100644 index 00000000..c3e1fd32 --- /dev/null +++ b/docs/swarms/utils/agent_loader.md @@ -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 \ No newline at end of file diff --git a/examples/agents_loader_example.py b/examples/agents_loader_example.py new file mode 100644 index 00000000..12359af6 --- /dev/null +++ b/examples/agents_loader_example.py @@ -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() \ No newline at end of file diff --git a/examples/multi_agent/simple_agent_loader_demo.py b/examples/multi_agent/simple_agent_loader_demo.py new file mode 100644 index 00000000..e78eae52 --- /dev/null +++ b/examples/multi_agent/simple_agent_loader_demo.py @@ -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() \ No newline at end of file diff --git a/swarms/utils/agent_loader.py b/swarms/utils/agent_loader.py new file mode 100644 index 00000000..7b918421 --- /dev/null +++ b/swarms/utils/agent_loader.py @@ -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) \ No newline at end of file diff --git a/test_agent_loader.py b/test_agent_loader.py new file mode 100644 index 00000000..3a02693a --- /dev/null +++ b/test_agent_loader.py @@ -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) \ No newline at end of file