CollabAgents 0.0.5

CollabAgents is a Python framework developed by Vishnu D. for developing AI agents equipped with specialized roles and tools to handle complex user requests efficiently. Users have 100 percent control over their prompts.

CollabAgents

Overview:

CollabAgents is a versatile Python framework developed by Vishnu Durairaj, designed to facilitate the creation of intelligent AI agents equipped with a diverse range of tools and functionalities. This open-source framework simplifies the development of sophisticated AI systems capable of handling various tasks through role-based agents and supports advanced multi-agent orchestration.

Features

• Create AI Agents with Tools: Easily build AI agents with a diverse set of tools, from Python execution to file operations and terminal commands.
• Role-Based Agents: Define agents with specific roles and responsibilities to handle different tasks and collaborate effectively.
• Interacting Entities: Simulate scenarios where multiple agents or companies work together to complete user requests.
• Flexible Integration: Supports both Anthropic and OpenAI models, providing flexibility in choosing the best AI model for your needs.
• Memory Management: Efficiently handle conversation history with various memory options:
  • ConversationBufferMemory: Retains the entire conversation history. Use this when you need to maintain a complete record of all interactions without any summarization or truncation.
  • ConversationBufferWindowMemory: Retains only the last K messages in the conversation. Ideal for scenarios where you want to limit memory usage and only keep the most recent interactions. (memory = ConversationBufferWindowMemory(last_k=3))
  • ConversationSummaryMemory: Automatically summarizes the conversation once it exceeds a specified number of messages. Use this to manage lengthy conversations by creating concise summaries while retaining overall context. (memory = ConversationSummaryMemory(number_of_messages=5))
  • ConversationSummaryBufferMemory: Combines summarization with selective message retention. Summarizes the conversation after a certain point and retains only the most recent N messages. Perfect for balancing context preservation with memory constraints by keeping key recent interactions while summarizing earlier ones. (memory = ConversationSummaryBufferMemory(buffer_size=5))

Colab Notebook

You can try out the notebook directly in Google Colab using the following link:

Open in Colab

Installation

You can install the CollabAgents framework using pip:

pip install CollabAgents

Example Use Case

1. Agents With Tools

Here’s an example of how to create a agent with tools using CollabAgents:

import os,nest_asyncio
from CollabAgents.helper import print_colored
from CollabAgents.agent import StructuredAgent
from CollabAgents.models import OpenaiChatModel
from CollabAgents.tools.FileOperationsTool import SaveFile, CreateFolder
from CollabAgents.memory import ConversationSummaryBufferMemory,ConversationSummaryMemory,ConversationBufferWindowMemory,ConversationBufferMemory

# If you are running this script in a notebook
# nest_asyncio.apply()

# Step 1: This concise description helps in understanding the agent's responsibilities and guides the system 
# in determining what types of tasks should be assigned to this agent. 
description = "Responsible for writing story."

# Step 2: Provide instructions for the agent (System Prompt)
instruction = "You are a creative storyteller who crafts imaginative narratives with vivid details, rich characters, and unexpected plot twists."

# Step 3: Load pre-defined tools that the agent will use
# These tools enable the agent to create folders and save files
tools = [CreateFolder, SaveFile]

# Step 4: Set your OpenAI API key
openai_api_key = "You API Key"
# openai_api_key = os.getenv('OPENAI_API_KEY')

# Step 5: Initialize the language model for the agent
model = OpenaiChatModel(model_name="gpt-4o-mini", api_key=openai_api_key, temperature=0)

# Step 6: Initialize memory - 4 different techniques are available.

# This option retains the entire conversation history. 
# Use this when you need to maintain a complete record of all interactions without any summarization or truncation.
memory = ConversationBufferMemory()        

# This option retains only the last K messages in the conversation.
# Ideal for scenarios where you want to limit memory usage and only keep the most recent interactions.
# memory = ConversationBufferWindowMemory(last_k=3)    

# This option automatically summarizes the conversation once it exceeds a specified number of messages.
# Use this to manage lengthy conversations by creating concise summaries while retaining overall context.
# memory = ConversationSummaryMemory(number_of_messages=5)   


# This option combines summarization with selective message retention. It summarizes the conversation after a certain point and retains only the most recent N messages.
# Perfect for balancing context preservation with memory constraints by keeping key recent interactions while summarizing earlier ones.
# memory = ConversationSummaryBufferMemory(buffer_size=5)    

# Step 7: Initialize the agent with the model, description, instructions, and tools. Set verbose to True to see the steps by step actions.
agent = StructuredAgent(model=model, agent_name="AI Assistant", agent_description=description, agent_instructions=instruction, tools=tools, assistant_agents=[],max_allowed_attempts=50, verbose=True,memory=memory)

print_colored("Starting the application...........", "green")

# Example user input
# user_input = "Create a story about AI agents and save it in a new folder. The story should have two chapters, and each chapter should be saved separately inside the folder"

user_input = input("User : ")

# Initialize the messages list to store conversation history
messages = []

# Step 8: Process user input and interact with the agent
while user_input != "bye":
    # The agent processes the user input and generates a response
    output = agent.run(user_input, messages)

    # Update the messages list with the agent's response
    messages = agent.messages

    # If verbose=False is set during agent initialization, uncomment the following line to see the agent's responses
    # print_colored(f"Assistant : {output}", "purple")

    # Prompt the user for the next input
    user_input = input("User Input : ")

2. Creating Custom Tools

Here’s an example of how to create a custom tool using pydantic base model:

import os
from typing import List
from pydantic import BaseModel,Field

class AppendToFile(BaseModel):
    # Based on this docstring, the model will determine when to use this tool. Ensure it clearly describes the tool's purpose.
    """
    Use this tool to append content to an existing file.
    """
    # Provides justification for selecting this tool, helping to ensure it is chosen appropriately and not at random. You can ignore this.
    reasoning :List[str] = Field(description="Why you are using this tool")

    # Thses are the required argument with its data types clearly declared.
    file_name: str = Field(..., description="File name to append to.") 
    content: str = Field(..., description="Content to append.")

    # Every tool must include a `run` method. This method will be called dynamically during interactions to perform the tool's primary function.
    def run(self):
        try:
            with open(self.file_name, "a") as file:
                file.write(self.content)
            return f"Content appended to file: {self.file_name}"
        except Exception as e:
            return f"An error occurred while appending to the file: {str(e)}"
    
AppendToFile(reasoning=["Thoughts"],file_name="path to the file",content="content to append").run()

3. Multi-Agents

Here’s an example of how to create multiple agents with tools using CollabAgents:

import os,nest_asyncio
from CollabAgents.models import AnthropicModel
from CollabAgents.agent import StructuredAgent
from CollabAgents.tools.FileOperationsTool import CreateFolder,SaveFile,ListFilesInDirectory,ReadTextFile
from CollabAgents.memory import ConversationSummaryBufferMemory,ConversationSummaryMemory,ConversationBufferWindowMemory,ConversationBufferMemory

# If you are running this script in a notebook
# nest_asyncio.apply()

api_key = "API KEY"

model = AnthropicModel(model_name="claude-3-sonnet-20240229",api_key=api_key,temperature=0)

memory = ConversationBufferMemory() 

# Let's Build a story writing company with 3 agents (CEO, Story Planner, Story Writer)

# ----------------------------------------------Create Story Planner Agent --------------------------------------

# Description for the Story Planner agent
Story_Planner_agent_description = (
    "Responsible for creating the story outline and structure. "
    "Works with the CEO to understand client requirements and plan the story accordingly."
)

# Instructions for the Story Planner agent
Story_Planner_instructions = (
    "As the Story Planner, you will receive the client requirements from the CEO. "
    "Create a detailed outline and structure for the story based on these requirements. "
    "Save the story outline in the project folder created by the CEO. "
    "Ensure that your plan is comprehensive and aligns with the client's vision. "
    "Collaborate with the CEO and Story Writer as needed to ensure the story's structure is clear and achievable."
)

# Tools for Story Planner

story_planner_tools = [SaveFile]

story_planner = StructuredAgent(model,"Story Planner",Story_Planner_agent_description,Story_Planner_instructions,tools=story_planner_tools,max_allowed_attempts=20,memory=memory)

# ----------------------------------------------- Create Story Writer Agent ------------------------------------

# Description for the Story Writer agent
Story_Writer_agent_description = (
    "Responsible for writing the story based on the outline provided by the Story Planner. "
    "Ensures the final story meets client expectations and follows the planned structure."
)

# Instructions for the Story Writer agent
Story_Writer_instructions = (
    "As the Story Writer, you will receive the story outline from the Story Planner, which is stored in the project folder created by the CEO. "
    "Write the story based on this outline, ensuring it meets the client's requirements and follows the planned structure. "
    "Save the completed story as a Markdown (.md) file in the same project folder. "
    "Communicate with the CEO for feedback and any necessary revisions to ensure the final story aligns with client expectations."
)

# Tools for Story Writer
story_writter_tools = [ReadTextFile,SaveFile]

story_writter = StructuredAgent(model,"Story Writer",Story_Writer_agent_description,Story_Writer_instructions,tools=story_writter_tools,max_allowed_attempts=20,memory=memory)

# ------------------------------------------------ Create CEO Agent --------------------------------------------

# Description for the CEO agent
CEO_agent_description = (
    "Responsible for client communication, task assignments, and coordination with other agents. "
    "Creates a new folder for each project to organize story planning and writing, and ensures the final story is stored as a Markdown file."
)

# Instructions for the CEO agent
CEO_agent_instructions = (
    "As the CEO of the Story Writing Company, you will manage client interactions and gather detailed requirements. "
    "Create a new folder in the current directory for each project based on the client's requirements to keep all related files organized. "
    "Delegate tasks to the Story Planner and Story Writer, instructing them to store their respective outputs (story plan and final story content) in the designated folder by providing the exact path. "
    "Ensure that the final story is saved as a Markdown (.md) file in the project folder and coordinate with both agents to guarantee smooth progress and timely delivery."
)

# Tools for CEO
CEO_tools = [CreateFolder,ListFilesInDirectory]

CEO_agent = StructuredAgent(model,"CEO",CEO_agent_description,CEO_agent_instructions,tools=CEO_tools,assistant_agents=[story_planner,story_writter],max_allowed_attempts=10,memory=memory)

# Let's ask CEO agent to create a story about AI

if __name__ =="__main__":

    # Example user input
    # user_input = "Create a story about AI agents. keep 2 chapters and the tone should be horror"

    user_input = input("User : ")

    # Initialize the messages list to store conversation history
    messages = []

    # Step 7: Process user input and interact with the agent
    while user_input != "bye":
        # The agent processes the user input and generates a response
        output = CEO_agent.run(user_input, messages)

        # Update the messages list with the agent's response
        messages = CEO_agent.messages

        # If verbose=False is set during agent initialization, uncomment the following line to see the agent's responses
        # print_colored(f"Assistant : {output}", "purple")

        # Prompt the user for the next input
        user_input = input("User Input : ")

4. Async Implementation

Here’s an example of async implementation:

import os,nest_asyncio,asyncio
from CollabAgents.helper import print_colored
from CollabAgents.agent import StructuredAgent
from CollabAgents.models import OpenaiChatModel
from CollabAgents.tools.FileOperationsTool import SaveFile, CreateFolder
from CollabAgents.memory import ConversationSummaryBufferMemory,ConversationSummaryMemory,ConversationBufferWindowMemory,ConversationBufferMemory

# If you are running this script in a notebook
# nest_asyncio.apply()

# Step 1: This concise description helps in understanding the agent's responsibilities and guides the system 
# in determining what types of tasks should be assigned to this agent. 
description = "Responsible for writing story."

# Step 2: Provide instructions for the agent (System Prompt)
instruction = "You are a creative storyteller who crafts imaginative narratives with vivid details, rich characters, and unexpected plot twists."

# Step 3: Load pre-defined tools that the agent will use
# These tools enable the agent to create folders and save files
tools = [CreateFolder, SaveFile]

# Step 4: Set your OpenAI API key
openai_api_key = "You API Key"
# openai_api_key = os.getenv('OPENAI_API_KEY')

# Step 5: Initialize the language model for the agent
model = OpenaiChatModel(model_name="gpt-4o-mini", api_key=openai_api_key, temperature=0)

# Step 6: Initialize memory - 4 different techniques are available.

# This option retains the entire conversation history. 
# Use this when you need to maintain a complete record of all interactions without any summarization or truncation.
memory = ConversationBufferMemory()        

# This option retains only the last K messages in the conversation.
# Ideal for scenarios where you want to limit memory usage and only keep the most recent interactions.
# memory = ConversationBufferWindowMemory(last_k=3)    

# This option automatically summarizes the conversation once it exceeds a specified number of messages.
# Use this to manage lengthy conversations by creating concise summaries while retaining overall context.
# memory = ConversationSummaryMemory(number_of_messages=5)   


# This option combines summarization with selective message retention. It summarizes the conversation after a certain point and retains only the most recent N messages.
# Perfect for balancing context preservation with memory constraints by keeping key recent interactions while summarizing earlier ones.
# memory = ConversationSummaryBufferMemory(buffer_size=5)    

# Step 7: Initialize the agent with the model, description, instructions, and tools. Set verbose to True to see the steps by step actions.
agent = StructuredAgent(model=model, agent_name="AI Assistant", agent_description=description, agent_instructions=instruction, tools=tools, assistant_agents=[],max_allowed_attempts=50, verbose=True,memory=memory)

if __name__ =="__main__":

    async def main():

        print_colored("Starting the application...........", "green")

        # Example user input
        # user_input = "Create a story about AI agents and save it in a new folder. The story should have two chapters, and each chapter should be saved separately inside the folder"

        user_input = input("User : ")

        # Initialize the messages list to store conversation history
        messages = []

        # Step 8: Process user input and interact with the agent
        while user_input != "bye":

            # The agent processes the user input and generates a response
            output = await agent.arun(user_input, messages)

            # Update the messages list with the agent's response
            messages = agent.messages

            # If verbose=False is set during agent initialization, uncomment the following line to see the agent's responses
            # print_colored(f"Assistant : {output}", "purple")

            # Prompt the user for the next input
            user_input = input("User Input : ")

    asyncio.run(main())

For More Tutorials:

Visit my YouTube Channel

Linkedin

Check out my LinkedIn

License

This project is licensed under the MIT License. - see the LICENSE file for details.