README.md

Agents

This directory contains the individual agent implementations for the atomic agents framework. Each agent is specialized for a specific type of task and can be composed with other agents to create complex workflows.

Available Agents

Data Agent

The Data Agent is responsible for processing and transforming data. It supports various input and output formats and can apply transformations to the data.

Features:

• Convert between different data formats (CSV, JSON, DataFrame, etc.)
• Apply transformations (filter, sort, group, etc.)
• Configure output format

Learn more about the Data Agent

Search Agent

The Search Agent is responsible for performing search operations on various data sources. It supports different search engines and can apply post-processing to search results.

Features:

• Search in memory, databases, files, or external APIs
• Apply post-processing to search results (sorting, deduplication, etc.)
• Configure search parameters

Learn more about the Search Agent

Workflow Agent

The Workflow Agent is responsible for orchestrating workflows between other agents. It can define and execute complex workflows with conditional branching and data transformations.

Features:

• Define workflows with multiple steps
• Support conditional branching based on data
• Apply data transformations between steps
• Execute workflows with input data

Learn more about the Workflow Agent

Creating a New Agent

To create a new agent, follow these steps:

1. Create a new directory under agents/ with your agent name
2. Create an __init__.py file in the directory
3. Create a Python file for your agent implementation
4. Implement your agent by inheriting from BaseAgent in core/agent-base/base_agent.py
5. Override the process method to implement your agent's logic
6. Add tests in the tests/ directory
7. Register your agent in your configuration file

Example:

from core.agent-base.base_agent import BaseAgent

class MyCustomAgent(BaseAgent):
    async def initialize(self) -> None:
        await super().initialize()
        # Custom initialization logic

    async def process(self, input_data):
        # Process the input data
        result = self._process_data(input_data)
        return result
        
    async def shutdown(self) -> None:
        # Custom shutdown logic
        await super().shutdown()