Agent Workflow Simulator

What It Does

Installation

pip install click

Usage

python agent_workflow_simulator.py --workflow <path_to_workflow_file> [--log-file <path_to_log_file>]

{
  "steps": [
    {"name": "Step 1", "action": "Action 1", "execution_time": 1},
    {"name": "Step 2", "action": "Action 2", "execution_time": 2}
  ]
}

from agent_workflow_simulator import simulate_workflow

workflow = {
    "steps": [
        {"name": "Step 1", "action": "Action 1", "execution_time": 1},
        {"name": "Step 2", "action": "Action 2", "execution_time": 2}
    ]
}

simulate_workflow(workflow, log_file="simulation.log")

Source Code

import json
import logging
import time
import click

def simulate_workflow(workflow, log_file=None):
    """
    Simulates the agent workflow step-by-step.

    Args:
        workflow (dict): The workflow definition.
        log_file (str): Optional log file to write simulation logs.

    Returns:
        None
    """
    logger = logging.getLogger("AgentWorkflowSimulator")
    logger.setLevel(logging.INFO)

    handler = logging.FileHandler(log_file) if log_file else logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger.handlers = [handler]  # Ensure no duplicate handlers

    logger.info("Starting workflow simulation...")

    steps = workflow.get("steps", [])
    if not steps:
        logger.error("No steps found in the workflow.")
        return

    for step in steps:
        step_name = step.get("name", "Unnamed Step")
        action = step.get("action", "No action defined")

        logger.info(f"Executing step: {step_name}")
        logger.info(f"Action: {action}")

        try:
            # Simulate execution time
            execution_time = float(step.get("execution_time", 1))
            time.sleep(execution_time)
            logger.info(f"Step '{step_name}' completed in {execution_time} seconds.")
        except Exception as e:
            logger.error(f"Error during step '{step_name}': {e}")

    logger.info("Workflow simulation completed.")

@click.command()
@click.option('--workflow', type=click.Path(exists=True), required=True, help='Path to the workflow file (JSON or Python script).')
@click.option('--log-file', type=click.Path(), default=None, help='Optional log file to write simulation logs.')
def main(workflow, log_file):
    """
    CLI entry point for the Agent Workflow Simulator.
    """
    try:
        with open(workflow, 'r') as f:
            workflow_data = json.load(f)
        simulate_workflow(workflow_data, log_file)
    except json.JSONDecodeError:
        click.echo("Error: Invalid JSON format in workflow file.", err=True)
    except FileNotFoundError:
        click.echo("Error: Workflow file not found.", err=True)
    except Exception as e:
        click.echo(f"An error occurred: {e}", err=True)

if __name__ == "__main__":
    main()