README.md

Scripts

This directory contains scripts for building, testing, and running the atomic agents framework. These scripts provide a convenient way to interact with the framework.

Available Scripts

Build Script

The build script (build.py) provides commands for cleaning, linting, type checking, and building the package.

Features:

• Clean build artifacts
• Run linting with flake8
• Run type checking with mypy
• Build the package

Learn more about the Build Script

Run Script

The run script (run.py) provides a way to run workflows based on configuration files.

Features:

• Load configuration from YAML files
• Set up logging
• Instantiate and initialize agents
• Execute workflows

Learn more about the Run Script

Usage

Build Script

# Clean build artifacts
python scripts/build.py clean

# Run linting
python scripts/build.py lint

# Run type checking
python scripts/build.py typecheck

# Build the package
python scripts/build.py build

# Run all steps
python scripts/build.py all

Run Script

# Run a workflow with a specific configuration file
python scripts/run.py --config config/sample_config.yaml

# Run a specific workflow from the configuration
python scripts/run.py --config config/sample_config.yaml --workflow data_processing

# Enable debug logging
python scripts/run.py --config config/sample_config.yaml --debug

Configuration

The run script expects a configuration file with the following structure:

# Logging configuration
logging:
  level: "INFO"
  log_file: "logs/app.log"
  log_format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

# Agent configurations
agents:
  data_agent_1:
    type: "data"
    output_format: "dict"
    transformations:
      - type: "filter"
        column: "status"
        condition: "equals"
        value: "active"
  
  search_agent_1:
    type: "search"
    search_engine: "memory"
    apply_post_processing: true

# Workflow definitions
workflows:
  data_processing:
    description: "Process data and search for results"
    steps:
      - agent_id: "data_agent_1"
        type: "agent"
      - agent_id: "search_agent_1"
        type: "agent"

Extending the Scripts

To add new scripts to the framework:

1. Create a new Python file in the scripts directory
2. Implement your script functionality
3. Add command-line argument parsing if needed
4. Document the script usage

Example:

# scripts/test_workflow.py
import argparse
import asyncio
import sys
import os

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from utils.config_utils import load_config
from core.orchestration.orchestrator import Orchestrator
from core.messaging.message_broker import MessageBroker
from agents.data-agent.data_agent import DataAgent
from agents.search-agent.search_agent import SearchAgent


async def run_test_workflow(config_path, input_data):
    # Load configuration
    config = load_config(config_path)
    
    # Create message broker and orchestrator
    broker = MessageBroker()
    orchestrator = Orchestrator(broker)
    
    # Start the orchestrator
    await orchestrator.start()
    
    try:
        # Create and register agents
        data_agent = DataAgent("data_agent_1", config.get("data_agent", {}))
        search_agent = SearchAgent("search_agent_1", config.get("search_agent", {}))
        
        orchestrator.register_agent(data_agent)
        orchestrator.register_agent(search_agent)
        
        # Initialize agents
        await data_agent.initialize()
        await search_agent.initialize()
        
        # Create a test workflow
        workflow_def = {
            "steps": [
                {
                    "agent_id": "data_agent_1"
                },
                {
                    "agent_id": "search_agent_1"
                }
            ]
        }
        orchestrator.create_workflow("test_workflow", workflow_def)
        
        # Execute the workflow
        result = await orchestrator.execute_workflow("test_workflow", input_data)
        
        print(f"Workflow result: {result}")
        return result
    
    finally:
        # Stop the orchestrator
        await orchestrator.stop()


def main():
    parser = argparse.ArgumentParser(description="Run a test workflow")
    parser.add_argument("--config", required=True, help="Path to the configuration file")
    parser.add_argument("--input", default="{}", help="JSON input data for the workflow")
    
    args = parser.parse_args()
    
    # Parse input data
    import json
    input_data = json.loads(args.input)
    
    # Run the workflow
    asyncio.run(run_test_workflow(args.config, input_data))


if __name__ == "__main__":
    main()