Generating Agents and Tasks files dynamically

Crews
, , ,
1

I was looking for a way to dynamically generate the agents.yaml and tasks.yaml files including the task and agent decorators in crew.py. I thought I might have the start of a script built to run “ollama list” pulling the list of local agents, but the whole system changed on me the past couple of days. This would have allowed me to generate a permanent agents and crew.py file with the correct decorators.

Here is the crew.py file:

import os
import subprocess
import yaml
from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools.tools.rag.rag_tool import RagTool
from crewai_tools import SerperDevTool

# Tools
rag_tool = RagTool()
serper_tool = SerperDevTool()

@CrewBase
class GenerateAgentInfoCrew:
    """Crew to dynamically generate ideal agent information from models."""
    output_file = 'outputs/ollama_agents.md'  # Single output file

    def __init__(self):
        # Ensure output directory exists
        os.makedirs(os.path.dirname(self.output_file), exist_ok=True)
        self.models = self.load_ollama_models()  # Load models dynamically

    def load_ollama_models(self):
        """Loads model names from the 'ollama list' command output."""
        try:
            # Run the ollama list command and capture the output
            result = subprocess.run(['ollama', 'list'], capture_output=True, text=True)

            # Split the output into lines, skipping the header
            lines = result.stdout.splitlines()[1:]

            # Extract the model names from the first column
            models = []
            for line in lines:
                parts = line.split()
                if len(parts) > 0:
                    model_name = parts[0].split(':')[0]  # Remove the :latest tag
                    models.append({"llm": model_name})
            return models

        except FileNotFoundError:
            print("Error: 'ollama' command not found. Make sure Ollama is installed and in your PATH.")
            return []


    @agent
    def dynamic_agent(self, model) -> Agent:
        """Creates a dynamic agent for the given model."""
        agent_name = f"agent_{model['llm']}"
        return Agent(
            name=agent_name,
            role="Model Ideal Agent Creator",
            goal="Generate the ideal CrewAI agent information.",
            backstory=(
                "You are the model loaded from the Ollama repository. "
                "Your task is to describe your ideal CrewAI agent, including role, goal, and backstory."
            ),
            llm=f"ollama/{model['llm']}",  # Add the "ollama/" prefix here
            tools=[rag_tool, serper_tool],
            verbose=True,
            memory=True,
            allow_delegation=False,
            human_input=False
        )

    @task
    def generate_agent_info_task(self, model) -> Task:
        """Task for generating agent information."""
        description = (
            "Describe your ideal CrewAI agent with the following details:\n"
            "- Role: What would be your primary role as an agent?\n"
            "- Goal: What would you aim to achieve?\n"
            "- Backstory: Provide a creative backstory for your agent.\n"
            "- LLM: Mention the model you represent."
        )

        # Modify the task to append results to a single file
        def write_output(output):
            with open(self.output_file, mode='a', encoding='utf-8') as f:
                f.write(f"## Agent Information for {model['llm']}\n\n")
                f.write(output + "\n\n")

        agent_name = f"agent_{model['llm']}"  # Use the same name here
        return Task(
            description=description,
            expected_output="Role, Goal, Backstory, and LLM description.",
            tools=[rag_tool, serper_tool],
            agent=agent_name,  # Refer to the agent by its name
            post_process=write_output  # Append output to the single file
        )

    @crew
    def crew(self) -> Crew:
        """Creates a dynamic crew for generating agent information."""
        agents = []
        tasks = []
        for model in self.models:
            agent_instance = self.dynamic_agent(model)
            agents.append(agent_instance)
            tasks.append(self.generate_agent_info_task(model))

        def generate_yaml_files(agents, tasks):
            agents_data = {agent.name: {"llm": agent.llm} for agent in agents}
            with open("config/agents.yaml", "w") as f:  # Write to config/agents.yaml
                yaml.dump(agents_data, f)

            tasks_data = {
                task.name: {"agent": task.agent, "description": task.description}
                for task in tasks
            }
            with open("config/tasks.yaml", "w") as f:  # Write to config/tasks.yaml
                yaml.dump(tasks_data, f)

        # Generate the YAML files
        generate_yaml_files(agents, tasks)

        # Rename existing YAML files if they exist
        if os.path.exists("config/agents.yaml"):
            os.rename("config/agents.yaml", "config/agents.yaml.old")
        if os.path.exists("config/tasks.yaml"):
            os.rename("config/tasks.yaml", "config/tasks.yaml.old")

        # Load the Crew from the generated YAML files
        return Crew.from_yaml()  # This will load from config/agents.yaml and config/tasks.yaml


# Example of execution:
if __name__ == "__main__":
    crew = GenerateAgentInfoCrew().crew()
    crew.kickoff(inputs={})
2

Although it might not perfectly achieve your goal, I believe it’s worth exploring the Planning feature. As stated in the docs:

The planning feature in CrewAI allows you to add planning capability to your crew. When enabled, before each Crew iteration, all Crew information is sent to an AgentPlanner that will plan the tasks step by step, and this plan will be added to each task description.

To enable planning, you need to set the following parameters:

  • planning: Set to True
  • planning_llm: Assign a specific LLM you would like to use.

Here’s a code example:

from crewai import Crew, Agent, Task, Process, LLM

my_crew = Crew(
    agents=self.agents,
    tasks=self.tasks,
    process=Process.sequential,
    planning=True,
    planning_llm=LLM(model="anthropic/claude-3-5-sonnet-20241022"),
)
3

I thought about using planning, but I wanted a comprehensive list of agents generated for the purposes of planning with hierarchical process. Planning requires the agents list to be already populated. I wanted to make a file of the agents automatically with each agent’s output including their specializations. This will not run with the new UV addition. Can anyone give me a clue as to how to update the code so that I can dynamically create single member crews with outputs in the format needed for the agents.yaml? I have used CrewAI Assistant, SIFT, Local: Ollama/ALIENINTELLIGENCE/crewaiconsultant:latest, and Gemini Advanced to assist in creating the current code.

4

I’m actually playing around writing a Python script that needs to do similar, but not using CrewAI at the moment. The idea is to use it as a blueprint to build a CrewAI crew.

5

I was just able to complete the agents file and tried to run planning, but it didn’t work due to the change to UV from poetry. I used Gemini to output the agent information in the agents.yaml file. I then used a python script to output the updated information to put into the crew.py file. Here is the code for the python script I used:

def generate_crew_py(names):
  """Generates the content for crew.py with agent names from a file."""
  output = []
  for name in names:
    output.append(f"""\
\t@agent
\tdef {name}(self) -> Agent:
\t\treturn Agent(
\t\t\tconfig=self.agents_config['{name}'],
\t\t\tverbose=True,
\t\t\ttools=[rag_tool, serper_tool],
\t\t\tmemory=True,
\t\t\tallow_delegation=False,
\t\t\thuman_input=False
\t\t)
""")
  return "\n".join(output)

def read_names_from_file(filename):
  """Reads agent names from a .txt file."""
  with open(filename, 'r') as file:
    return [line.strip() for line in file]

def write_to_file(filename, content):
  """Writes the content to the specified file."""
  with open(filename, 'w') as file:
    file.write(content)

# Get the agent names from the file
names = read_names_from_file(r'D:\AIDataReference\agents\src\agents\AgentNames.txt') 

# Generate the crew.py content
crew_content = generate_crew_py(names)

# Write the output to crew_output.py
write_to_file('crew_output.py', crew_content)

Maybe you can adapt this to assist with development. I do also want to have an auto generated hierarchical planned crew that can complete tasks and connect to a database by pulling tasks with context.