Advanced Agent Configuration and Customization

In addition to the basic setup and agent creation, Cleo offers a rich set of advanced features that allow you to configure and customize agents in highly flexible ways. This section explores how to fine-tune agent behavior, create custom execution strategies, and adjust agent configurations dynamically using configuration files.

---

Custom Task Execution Strategies

While Cleo provides default task execution strategies, you may need to customize the way tasks are executed by your agents. This can be useful in scenarios where you want agents to prioritize certain tasks, execute them asynchronously, or handle specific edge cases.

1. Task Execution Strategy Customization

You can define custom strategies for task execution by subclassing the base Task class and overriding its execution methods. This gives you complete control over how tasks are handled by the agent.

Example: Custom Task Execution Strategy

pythonCopyEditfrom cleo import Task

class CustomTask(Task):
    def __init__(self, priority: int):
        self.priority = priority

    def execute(self):
        self.log(f"Executing task with priority {self.priority}")
        # Custom task logic here
        return "Task completed"

# Example usage
task = CustomTask(priority=10)
task.execute()

In this example, a custom task execution strategy is created by subclassing the Task class. The priority attribute influences how the task is executed within the agent's queue or workflow.

2. Asynchronous Task Execution

You can also implement asynchronous task execution to allow agents to perform multiple tasks concurrently. This is particularly useful for tasks that involve waiting for external responses or involve long-running operations.

Example: Asynchronous Task Execution with asyncio

pythonCopyEditimport asyncio

class AsyncTask(Task):
    async def execute(self):
        self.log("Starting asynchronous task...")
        await asyncio.sleep(2)
        self.log("Asynchronous task completed.")
        return "Task finished"

# Example usage
async def run_async_task():
    task = AsyncTask()
    await task.execute()

# Running the asynchronous task
asyncio.run(run_async_task())

This example demonstrates how to use asyncio for asynchronous task execution, allowing agents to handle tasks without blocking their operation while waiting for responses.

---

Dynamic Configuration via Files

For complex agent setups, it’s often helpful to use configuration files that can be read and adjusted dynamically. Cleo allows you to define agents and tasks with parameters specified in configuration files, making it easier to modify behavior without changing the underlying code.

1. Configuration File Setup

Cleo supports various configuration formats such as JSON, YAML, and TOML for dynamic configuration. You can define settings like agent parameters, task properties, and external service configurations in these files.

Example: Reading Configuration from a JSON File

jsonCopyEdit{
  "agent_name": "TaskMaster",
  "tasks": [
    {"name": "task1", "priority": 1},
    {"name": "task2", "priority": 2}
  ]
}

In Python, you can read and parse the configuration file as follows:

pythonCopyEditimport json

def load_config(file_path):
    with open(file_path, 'r') as file:
        return json.load(file)

# Load the agent's configuration
config = load_config("agent_config.json")
print(config['agent_name'])  # Output: TaskMaster

This approach allows you to adjust the behavior of agents by simply modifying the configuration file without needing to change the code itself.

2. Using YAML for Configuration

YAML is a popular configuration format due to its readability and simplicity. You can use the PyYAML library to load YAML files into Python objects.

Example: Configuration in YAML

yamlCopyEditagent_name: TaskMaster
tasks:
  - name: task1
    priority: 1
  - name: task2
    priority: 2

In Python:

pythonCopyEditimport yaml

def load_yaml_config(file_path):
    with open(file_path, 'r') as file:
        return yaml.safe_load(file)

# Load the agent's configuration
config = load_yaml_config("agent_config.yaml")
print(config['agent_name'])  # Output: TaskMaster

This allows you to modify configurations and update agent behavior by editing a simple YAML file.

---

Customizing Agent Behavior with Middleware

Cleo supports the use of middleware to enhance or modify the behavior of agents during task execution. Middleware functions allow you to insert logic before or after task execution, or even change the task’s input or output.

1. Using Middleware for Task Execution

Middleware is a powerful tool for customizing task workflows. You can define middleware functions that execute before or after tasks, providing the flexibility to modify task parameters, log information, or even perform validations.

Example: Middleware for Logging Task Execution

pythonCopyEditclass TaskMiddleware:
    def __init__(self, agent):
        self.agent = agent

    def before_task_execution(self, task):
        self.agent.log(f"About to execute task: {task.name}")

    def after_task_execution(self, task, result):
        self.agent.log(f"Task {task.name} executed. Result: {result}")

# Attach middleware to agent
middleware = TaskMiddleware(agent)
middleware.before_task_execution(task)
task.execute()
middleware.after_task_execution(task, "Task completed")

In this example, middleware functions are defined to log messages before and after task execution, giving you the ability to add custom behavior during the agent’s lifecycle.

---

Agent Initialization and Dependency Injection

For complex agents that rely on multiple components, Cleo supports dependency injection to help manage the initialization of various system components, such as databases, external APIs, or other services.

1. Dependency Injection Example

You can inject dependencies into agents to ensure they have access to the resources they need.

pythonCopyEditclass Database:
    def connect(self):
        # Simulate a database connection
        return "Database connection established."

class AgentWithDatabase:
    def __init__(self, db: Database):
        self.db = db

    def execute(self):
        connection = self.db.connect()
        self.log(connection)
        return "Agent completed task."

# Example usage
db = Database()
agent = AgentWithDatabase(db)
agent.execute()

This setup ensures that the agent has access to its dependencies (like a database) and allows for easier testing and maintenance of your code.