Agentic Task Simulator

What It Does

Installation

pip install matplotlib==3.7.1

Usage

python agentic_task_simulator.py --agents agent1.py agent2.py --scenario scenario.json

Source Code

import argparse
import json
import tempfile
import os
import matplotlib.pyplot as plt

def simulate_agents(agent_scripts, scenario_file, visualize):
    """
    Simulates the execution of agents in a mock environment based on a scenario file.

    Args:
        agent_scripts (list): List of paths to agent scripts.
        scenario_file (str): Path to the scenario JSON file.
        visualize (bool): Whether to visualize the workflow.

    Returns:
        dict: Simulation results including logs and debug metrics.
    """
    if not agent_scripts:
        raise ValueError("No agent scripts provided.")

    if not os.path.exists(scenario_file):
        raise FileNotFoundError(f"Scenario file '{scenario_file}' not found.")

    try:
        with open(scenario_file, 'r') as f:
            scenario = json.load(f)
    except json.JSONDecodeError:
        raise ValueError("Invalid JSON in scenario file.")

    logs = []
    metrics = {}

    for agent_script in agent_scripts:
        if not os.path.exists(agent_script):
            raise FileNotFoundError(f"Agent script '{agent_script}' not found.")

        logs.append(f"Simulating agent: {agent_script}")

    # Mock simulation logic
    for step in scenario.get("steps", []):
        logs.append(f"Executing step: {step}")

    metrics["total_steps"] = len(scenario.get("steps", []))
    metrics["agents_used"] = len(agent_scripts)

    if visualize:
        visualize_workflow(metrics)

    return {"logs": logs, "metrics": metrics}

def visualize_workflow(metrics):
    """Visualizes the workflow metrics using a bar chart."""
    labels = list(metrics.keys())
    values = list(metrics.values())

    plt.bar(labels, values)
    plt.title("Simulation Metrics")
    plt.xlabel("Metric")
    plt.ylabel("Value")
    plt.show()

def main():
    parser = argparse.ArgumentParser(
        description="Agentic Task Simulator: Simulate and debug AI agent workflows."
    )
    parser.add_argument(
        "--agents",
        nargs='+',
        required=True,
        help="Paths to agent scripts to simulate."
    )
    parser.add_argument(
        "--scenario",
        required=True,
        help="Path to the JSON scenario file."
    )
    parser.add_argument(
        "--visualize",
        action='store_true',
        help="Visualize the simulation metrics."
    )

    args = parser.parse_args()

    try:
        results = simulate_agents(args.agents, args.scenario, args.visualize)
        print("Simulation Logs:")
        for log in results["logs"]:
            print(log)
        print("\nDebug Metrics:")
        print(json.dumps(results["metrics"], indent=4))
    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    main()