Multi AI Agent Systems with crewAI

이 글은 Multi AI Agent Systems with crewAI 강의를 보고 정리한 글입니다.

---

Outline:

• CrewAI 는 Multi AI Agents System 을 개발하기 위한 프레임워크임. 여기서는 Multi Agent Workflow System 을 만드는데 필요한 Building block 들에 대해서 배워볼거임.
  • Role-playing
  • tool use
  • memory
  • Guardrails
  • Collaboration
  • Workflow Style (e.g Hierachical, Parallel, Series, Manager Agent delegating to a number of worker agents)

 

Moura 가 말하는 현재의 좋은 Agent System 구조:

• Manager Agent 와 다수의 역할을 가진 Wokrer Agent가 협업을 하는 구조.
• 역할을 가진 Worker Agent 는 명확한 목표를 가지고 일함.

 

1. Overview

Agent 시스템이 어떤 목표를 가지고 움지직이고, 어떻게 동작하고, 예시들도 한번 살펴보자.

Multi-Agent Collaboration:

• Agent 가 협업하는 방식은 Task 마다 다를 수 있음.

 

 

Multi-Agent System 을 만들 때:

• 목표에 따라서 어떤 Agent 가 필요한지, 어떤 도구가 필요한지 결정해야한다.
• 취업을 도와주는 Agent System 를 만든다고 했을 때는 다음과 같은 것들이 필요할 것:
  • Agents:
    • Tech Job Researcher
    • Personal Profile for Engineer
    • Resume Strategist for Engineers
    • Engineering Interview Prepare
  • Tools:
    • Search the internet
    • Read websites
    • Read resume
    • Perform RAG (Retrieval Augmented Generation) on resume

 

 

2. AI Agents

ChatGPT 와 같은 LLM 과의 소통에서 더 나은 답변을 얻기 위한 과정은 반복적인 상호작용이 필요하다:

• 한번에 원하는 형식의 답을 얻는 건 어렵기 때문에, 여러번의 상호작용과 피드백을 거쳐서 올바른 답을 얻을 수 있는거임.
• 이걸 자동화 할 수 있는게 AI Agents 임. Agents 를 통해서 피드백을 줘서 돌려깎아서 더 나은 답을 만들도록 하는거다. 역할만 나눠서 일을 분배해서 나누는 것 뿐 아니라

 

CrewAI:

• AI 에이전트 시스템 쉽게 사용
• 에이전트 시스템을 함께 조합하여 작동할 수 있도록 하는 패턴 제공
• 에이전트가 즉시 사용할 수 있는 다양한 도구와 기술들을 제공
• 개발자들이 에이전트를 위해 맞춤형 도구를 만들 수 있는 모델 제공
• 개발된 에이전트를 실제 운영 환경에 배포할 수 있는 플랫폼을 제공

 

 

2.1 CrewAI Core Concepts

2.1.1 Agents

Agents:

• Perform tasks
• Make decisions
• Communicate with other agents
• Think of an agent as a member of a team, with specific skills and a particular job to do.

 

Agent Attributes:

• Role: Defines the agent's function within the crew. It determines the kind of tasks the agent is best suited for.
• Goal: The individual objective that the agent aims to achieve. It guides the agent's decision-making process.
• Backstory: Provides context to the agent's role and goal, enriching the interaction and collaboration dynamics.
• LLM (optional): Represents the language model that will run the agent. It dynamically fetches the model name from the OPENAI_MODEL_NAME environment variable, defaulting to "gpt-4" if not specified.
• Tools (optional): Set of capabilities or functions that the agent can use to perform tasks. Expected to be instances of custom classes compatible with the agent's execution environment. Tools are initialized with a default value of an empty list.
• Function Calling LLM (optional): Specifies the language model that will handle the tool calling for this agent, overriding the crew function calling LLM if passed. Default is None.
• Max Iter (Optional): max_iter is the maximum number of iterations the agent can perform before being forced to give its best answer. Default is 25.
• Max RPM (optional): max_rpm is Tte maximum number of requests per minute the agent can perform to avoid rate limits. It's optional and can be left unspecified, with a default value of None.
• Max Execution Time (optional): max_execution_time is the Maximum execution time for an agent to execute a task. It's optional and can be left unspecified, with a default value of None, meaning no max execution time.
• Verbose (optional): Setting this to True configures the internal logger to provide detailed execution logs, aiding in debugging and monitoring. Default is False.
• Allow Delegation (optional): Agents can delegate tasks or questions to one another, ensuring that each task is handled by the most suitable agent. Default is True.
• Step Callback (optional): A function that is called after each step of the agent. This can be used to log the agent's actions or to perform other operations. It will overwrite the crew step_callback.
• Cache (optional): Indicates if the agent should use a cache for tool usage. Default is True.
• System Template (optional): Specifies the system format for the agent. Default is None.
• Prompt Template (optional): Specifies the prompt format for the agent. Default is None.
• Response Template (optional): Specifies the response format for the agent. Default is None.

# Example: Creating an agent with all attributes
from crewai import Agent

agent = Agent(
  role='Data Analyst',
  goal='Extract actionable insights',
  backstory="""You're a data analyst at a large company.
  You're responsible for analyzing data and providing insights
  to the business.
  You're currently working on a project to analyze the
  performance of our marketing campaigns.""",
  tools=[my_tool1, my_tool2],  # Optional, defaults to an empty list
  llm=my_llm,  # Optional
  function_calling_llm=my_llm,  # Optional
  max_iter=15,  # Optional
  max_rpm=None, # Optional
  max_execution_time=None, # Optional
  verbose=True,  # Optional
  allow_delegation=True,  # Optional
  step_callback=my_intermediate_step_callback,  # Optional
  cache=True,  # Optional
  system_template=my_system_template,  # Optional
  prompt_template=my_prompt_template,  # Optional
  response_template=my_response_template,  # Optional
  config=my_config,  # Optional
  crew=my_crew,  # Optional
  tools_handler=my_tools_handler,  # Optional
  cache_handler=my_cache_handler,  # Optional
  callbacks=[callback1, callback2],  # Optional
  agent_executor=my_agent_executor  # Optional
)

 

 

2.1.2 Tasks

Task:

• tasks are specific assignments completed by agents.
• They provide all necessary details for execution, such as a description, the agent responsible, required tools, and more, facilitating a wide range of action complexities
• Tasks within crewAI can be collaborative, requiring multiple agents to work together

 

Task Attributes:

• Description: A clear, concise statement of what the task entails.
• Agent: The agent responsible for the task, assigned either directly or by the crew's process.
• Expected Output: A detailed description of what the task's completion looks like.
• Tools (optional): The functions or capabilities the agent can utilize to perform the task.
• Async Execution (optional): If set, the task executes asynchronously, allowing progression without waiting for completion.
• Context (optional): Specifies tasks whose outputs are used as context for this task.
• Config (optional): Additional configuration details for the agent executing the task, allowing further customization.
• Output JSON (optional): Outputs a JSON object, requiring an OpenAI client. Only one output format can be set.
• Output Pydantic (optional): Outputs a Pydantic model object, requiring an OpenAI client. Only one output format can be set
• Output File (optional): Saves the task output to a file. If used with Output JSON or Output Pydantic, specifies how the output is saved
• Callback (optional): A Python callable that is executed with the task's output upon completion.
• Human Input (optional): Indicates if the task requires human feedback at the end, useful for tasks needing human oversight.

 

Creating a Task:

from crewai import Task

task = Task(
    description='Find and summarize the latest and most relevant news on AI',
    agent=sales_agent
)

 

 

Creating a Task with Tools:

import os
os.environ["OPENAI_API_KEY"] = "Your Key"
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API key

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool

research_agent = Agent(
  role='Researcher',
  goal='Find and summarize the latest AI news',
  backstory="""You're a researcher at a large company.
  You're responsible for analyzing data and providing insights
  to the business.""",
  verbose=True
)

search_tool = SerperDevTool()

task = Task(
  description='Find and summarize the latest AI news',
  expected_output='A bullet list summary of the top 5 most important AI news',
  agent=research_agent,
  tools=[search_tool]
)

crew = Crew(
    agents=[research_agent],
    tasks=[task],
    verbose=2
)

result = crew.kickoff()
print(result)

 

 

Referring to Other Tasks:

• In crewAI, the output of one task is automatically relayed into the next one.
• but you can specifically define what tasks' output, including multiple, should be used as context for another task.
• This is useful when you have a task that depends on the output of another task that is not performed immediately after it. This is done through the context attribute of the task:

research_ai_task = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    async_execution=True,
    agent=research_agent,
    tools=[search_tool]
)

research_ops_task = Task(
    description='Find and summarize the latest AI Ops news',
    expected_output='A bullet list summary of the top 5 most important AI Ops news',
    async_execution=True,
    agent=research_agent,
    tools=[search_tool]
)

write_blog_task = Task(
    description="Write a full blog post about the importance of AI and its latest news",
    expected_output='Full blog post that is 4 paragraphs long',
    agent=writer_agent,
    context=[research_ai_task, research_ops_task]
)

 

 

Asynchronous Execution:

• You can define a task to be executed asynchronously
• This is useful for tasks that take a long time to be completed, that are not crucial for the next tasks to be performed.
• You can then use the context attribute to define in a future task that it should wait for the output of the asynchronous task to be completed.

list_ideas = Task(
    description="List of 5 interesting ideas to explore for an article about AI.",
    expected_output="Bullet point list of 5 ideas for an article.",
    agent=researcher,
    async_execution=True # Will be executed asynchronously
)

list_important_history = Task(
    description="Research the history of AI and give me the 5 most important events.",
    expected_output="Bullet point list of 5 important events.",
    agent=researcher,
    async_execution=True # Will be executed asynchronously
)

write_article = Task(
    description="Write an article about AI, its history, and interesting ideas.",
    expected_output="A 4 paragraph article about AI.",
    agent=writer,
    context=[list_ideas, list_important_history] # Will wait for the output of the two tasks to be completed
)

 

 

Accessing a Specific Task Output:

• Once a crew finishes running, you can access the output of a specific task by using the output attribute of the task object

# ...
task1 = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    agent=research_agent,
    tools=[search_tool]
)

#...

crew = Crew(
    agents=[research_agent],
    tasks=[task1, task2, task3],
    verbose=2
)

result = crew.kickoff()

# Returns a TaskOutput object with the description and results of the task
print(f"""
    Task completed!
    Task: {task1.output.description}
    Output: {task1.output.raw_output}
""")

 

 

2.1.3 Crews

Crew:

• A crew in crewAI represents a collaborative group of agents working together to achieve a set of tasks
• Each crew defines the strategy for task execution, agent collaboration, and the overall workflow

 

Crew Attributes:

• Tasks: A list of tasks assigned to the crew.
• Agents: A list of agents that are part of the crew
• Process (optional): The process flow (e.g., sequential, hierarchical) the crew follows.
• Verbose (optional): The verbosity level for logging during execution.
• Manager LLM (optional): The language model used by the manager agent in a hierarchical process. Required when using a hierarchical process
• Function Calling LLM (optional): If passed, the crew will use this LLM to do function calling for tools for all agents in the crew. Each agent can have its own LLM, which overrides the crew's LLM for function calling.
• Config (optional): Optional configuration settings for the crew, in Json or Dict[str, Any] format.
• Max RPM (optional): Maximum requests per minute the crew adheres to during execution.
• Language (optional): Language used for the crew, defaults to English
• Language File (optional): Path to the language file to be used for the crew.
• Memory (optional): Utilized for storing execution memories (short-term, long-term, entity memory).
• Cache (optional): Specifies whether to use a cache for storing the results of tools' execution
• Embedder (optional): Configuration for the embedder to be used by the crew. Mostly used by memory for now.
• Full Output (optional): Whether the crew should return the full output with all tasks outputs or just the final output.
• Step Callback (optional): A function that is called after each step of every agent. This can be used to log the agent's actions or to perform other operations; it won't override the agent-specific step_callback.
• Task Callback (optional): A function that is called after the completion of each task. Useful for monitoring or additional operations post-task execution
• Share Crew (optional): Whether you want to share the complete crew information and execution with the crewAI team to make the library better, and allow us to train models.
• Output Log File (optional): Whether you want to have a file with the complete crew output and execution. You can set it using True and it will default to the folder you are currently in and it will be called logs.txt or passing a string with the full path and name of the file.
• Manager Agent (optional): manager sets a custom agent that will be used as a manager.
• Manager Callbacks (optional): manager_callbacks takes a list of callback handlers to be executed by the manager agent when a hierarchical process is used.
• Prompt File (optional): Path to the prompt JSON file to be used for the crew.

 

 

Creating a Crew:

• When assembling a crew, you combine agents with complementary roles and tools, assign tasks, and select a process that dictates their execution order and interaction.

from crewai import Crew, Agent, Task, Process
from langchain_community.tools import DuckDuckGoSearchRun

# Define agents with specific roles and tools
researcher = Agent(
    role='Senior Research Analyst',
    goal='Discover innovative AI technologies',
    backstory="""You're a senior research analyst at a large company.
        You're responsible for analyzing data and providing insights
        to the business.
        You're currently working on a project to analyze the
        trends and innovations in the space of artificial intelligence.""",
    tools=[DuckDuckGoSearchRun()]
)

writer = Agent(
    role='Content Writer',
    goal='Write engaging articles on AI discoveries',
    backstory="""You're a senior writer at a large company.
        You're responsible for creating content to the business.
        You're currently working on a project to write about trends
        and innovations in the space of AI for your next meeting.""",
    verbose=True
)

# Create tasks for the agents
research_task = Task(
    description='Identify breakthrough AI technologies',
    agent=researcher,
    expected_output='A bullet list summary of the top 5 most important AI news'
)
write_article_task = Task(
    description='Draft an article on the latest AI technologies',
    agent=writer,
    expected_output='3 paragraph blog post on the latest AI technologies'
)

# Assemble the crew with a sequential process
my_crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_article_task],
    process=Process.sequential,
    full_output=True,
    verbose=True,
)

 

 

Memory Utilization:

• Crews can utilize memory (short-term, long-term, and entity memory) to enhance their execution and learning over time
• This feature allows crews to store and recall execution memories, aiding in decision-making and task execution strategies.

 

 

Cache Utilization:

• Caches can be employed to store the results of tools' execution, making the process more efficient by reducing the need to re-execute identical tasks.

 

 

Crew Usage Metrics:

• After the crew execution, you can access the usage_metrics attribute to view the language model (LLM) usage metrics for all tasks executed by the crew. This provides insights into operational efficiency and areas for improvement.

# Access the crew's usage metrics
crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)

 

 

Crew Execution Process:

• Sequential Process: Tasks are executed one after another, allowing for a linear flow of work
• Hierarchical Process: A manager agent coordinates the crew, delegating tasks and validating outcomes before proceeding
• Note: A manager_llm or manager_agent is required for this process and it's essential for validating the process flow

 

 

Kicking Off a Crew:

• Once your crew is assembled, initiate the workflow with the kickoff() method. This starts the execution process according to the defined process flow.
• CrewAI provides several methods for better control over the kickoff process: kickoff(), kickoff_for_each(), kickoff_async(), and kickoff_for_each_async().
  • kickoff(): Starts the execution process according to the defined process flow
  • kickoff_for_each(): Executes tasks for each agent individually
  • kickoff_async(): Initiates the workflow asynchronously
  • kickoff_for_each_async(): Executes tasks for each agent individually in an asynchronous manner.

# Start the crew's task execution
result = my_crew.kickoff()
print(result)

# Example of using kickoff_for_each
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
results = my_crew.kickoff_for_each(inputs=inputs_array)
for result in results:
    print(result)

# Example of using kickoff_async
inputs = {'topic': 'AI in healthcare'}
async_result = my_crew.kickoff_async(inputs=inputs)
print(async_result)

# Example of using kickoff_for_each_async
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
async_results = my_crew.kickoff_for_each_async(inputs=inputs_array)
for async_result in async_results:
    print(async_result)

 

 

2.1.4 Processes

Process:

• processes orchestrate the execution of tasks by agents
• These processes ensure tasks are distributed and executed efficiently, in alignment with a predefined strategy.

 

 

Process Implementations:

• Sequential: Executes tasks sequentially, ensuring tasks are completed in an orderly progression
• Hierarchical: Organizes tasks in a managerial hierarchy, where tasks are delegated and executed based on a structured chain of command. A manager language model (manager_llm) or a custom manager agent (manager_agent) must be specified in the crew to enable the hierarchical process, facilitating the creation and management of tasks by the manager
• Consensual Process (Planned): Aiming for collaborative decision-making among agents on task execution, this process type introduces a democratic approach to task management within CrewAI. It is planned for future development and is not currently implemented in the codebase.

 

 

Sequential Process:

• This method mirrors dynamic team workflows, progressing through tasks in a thoughtful and systematic manner. Task execution follows the predefined order in the task list, with the output of one task serving as context for the next.
• To customize task context, utilize the context parameter in the Task class to specify outputs that should be used as context for subsequent tasks.

 

 

Hierarchical Process:

• Emulates a corporate hierarchy, CrewAI allows specifying a custom manager agent or automatically creates one, requiring the specification of a manager language model (manager_llm). This agent oversees task execution, including planning, delegation, and validation. Tasks are not pre-assigned; the manager allocates tasks to agents based on their capabilities, reviews outputs, and assesses task completion.

 

 

Additional Task Features:

• Asynchronous Execution
• Human Input Review: An optional feature that enables the review of task outputs by humans to ensure quality and accuracy before finalization. This additional step introduces a layer of oversight, providing an opportunity for human intervention and validation.
• Output Customization: Tasks support various output formats, including JSON (output_json), Pydantic models (output_pydantic), and file outputs (output_file), providing flexibility in how task results are captured and utilized

 

 

2.1.5 Tools

Tools:

• tools empower agents with capabilities ranging from web searching and data analysis to collaboration and delegating tasks among coworkers
• A tool in CrewAI is a skill or function that agents can utilize to perform various actions.
• This includes tools from the crewAI Toolkit and LangChain Tools, enabling everything from simple searches to complex interactions and effective teamwork among agents.

 

 

Using crewAI Tools:

• To enhance your agents' capabilities with crewAI tools, begin by installing our extra tools package

pip install 'crewai[tools]'

 

 

Here's an example demonstrating their use:

import os
from crewai import Agent, Task, Crew
# Importing crewAI tools
from crewai_tools import (
    DirectoryReadTool,
    FileReadTool,
    SerperDevTool,
    WebsiteSearchTool
)

# Set up API keys
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API key
os.environ["OPENAI_API_KEY"] = "Your Key"

# Instantiate tools
docs_tool = DirectoryReadTool(directory='./blog-posts')
file_tool = FileReadTool()
search_tool = SerperDevTool()
web_rag_tool = WebsiteSearchTool()

# Create agents
researcher = Agent(
    role='Market Research Analyst',
    goal='Provide up-to-date market analysis of the AI industry',
    backstory='An expert analyst with a keen eye for market trends.',
    tools=[search_tool, web_rag_tool],
    verbose=True
)

writer = Agent(
    role='Content Writer',
    goal='Craft engaging blog posts about the AI industry',
    backstory='A skilled writer with a passion for technology.',
    tools=[docs_tool, file_tool],
    verbose=True
)

# Define tasks
research = Task(
    description='Research the latest trends in the AI industry and provide a summary.',
    expected_output='A summary of the top 3 trending developments in the AI industry with a unique perspective on their significance.',
    agent=researcher
)

write = Task(
    description='Write an engaging blog post about the AI industry, based on the research analyst’s summary. Draw inspiration from the latest blog posts in the directory.',
    expected_output='A 4-paragraph blog post formatted in markdown with engaging, informative, and accessible content, avoiding complex jargon.',
    agent=writer,
    output_file='blog-posts/new_post.md'  # The final blog post will be saved here
)

# Assemble a crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research, write],
    verbose=2
)

# Execute tasks
crew.kickoff()

 

 

Available crewAI Tools:

• Error Handling: All tools are built with error handling capabilities, allowing agents to gracefully manage exceptions and continue their tasks.
• Caching Mechanism:
  • All tools support caching, enabling agents to efficiently reuse previously obtained results, reducing the load on external resources and speeding up the execution time
  • You can also define finer control over the caching mechanism using the cache_function attribute on the tool.

 

 

Here is a list of the available tools and their descriptions:

• CodeDocsSearchTool: A RAG tool optimized for searching through code documentation and related technical documents
• CSVSearchTool: A RAG tool designed for searching within CSV files, tailored to handle structured data.
• DirectorySearchTool: A RAG tool for searching within directories, useful for navigating through file systems
• DOCXSearchTool: A RAG tool aimed at searching within DOCX documents, ideal for processing Word files
• DirectoryReadTool: Facilitates reading and processing of directory structures and their contents
• FileReadTool: Enables reading and extracting data from files, supporting various file formats.
• GithubSearchTool: A RAG tool for searching within GitHub repositories, useful for code and documentation search.
• SerperDevTool: A specialized tool for development purposes, with specific functionalities under development
• TXTSearchTool: A RAG tool focused on searching within text (.txt) files, suitable for unstructured data.
• JSONSearchTool: A RAG tool designed for searching within JSON files, catering to structured data handling.
• MDXSearchTool: A RAG tool tailored for searching within Markdown (MDX) files, useful for documentation.
• PDFSearchTool: A RAG tool aimed at searching within PDF documents, ideal for processing scanned documents.
• PGSearchTool: A RAG tool optimized for searching within PostgreSQL databases, suitable for database queries
• RagTool: A general-purpose RAG tool capable of handling various data sources and types.
• ScrapeElementFromWebsiteTool: Enables scraping specific elements from websites, useful for targeted data extraction.
• ScrapeWebsiteTool: Facilitates scraping entire websites, ideal for comprehensive data collection
• WebsiteSearchTool: A RAG tool for searching website content, optimized for web data extraction.
• XMLSearchTool: A RAG tool designed for searching within XML files, suitable for structured data formats.
• YoutubeChannelSearchTool: A RAG tool for searching within YouTube channels, useful for video content analysis
• YoutubeVideoSearchTool: A RAG tool aimed at searching within YouTube videos, ideal for video data extraction
• BrowserbaseTool: A tool for interacting with and extracting data from web browsers.
• ExaSearchTool: A tool designed for performing exhaustive searches across various data sources.

 

 

Creating your own Tools:

• Subclassing BaseTool
• Utilizing the tool Decorator

# Subclassing BaseTool 
from crewai_tools import BaseTool

class MyCustomTool(BaseTool):
    name: str = "Name of my tool"
    description: str = "Clear description for what this tool is useful for, your agent will need this information to use it."

    def _run(self, argument: str) -> str:
        # Implementation goes here
        return "Result from custom tool"

# Utilizing the tool Decorator

from crewai_tools import tool
@tool("Name of my tool")
def my_tool(question: str) -> str:
    """Clear description for what this tool is useful for, your agent will need this information to use it."""
    # Function logic here
    return "Result from your custom tool"

 

 

Custom Caching Mechanism:

from crewai_tools import tool

@tool
def multiplication_tool(first_number: int, second_number: int) -> str:
    """Useful for when you need to multiply two numbers together."""
    return first_number * second_number

def cache_func(args, result):
    # In this case, we only cache the result if it's a multiple of 2
    cache = result % 2 == 0
    return cache

multiplication_tool.cache_function = cache_func

writer1 = Agent(
        role="Writer",
        goal="You write lessons of math for kids.",
        backstory="You're an expert in writing and you love to teach kids but you know nothing of math.",
        tools=[multiplication_tool],
        allow_delegation=False,
    )

 

 

2.1.6 Memory

Memory:

• This system comprises short-term memory, long-term memory, entity memory, and contextual memory, each serving a unique purpose in aiding agents to remember, reason, and learn from past interactions.

 

 

Memory System Components:

• Short-Term Memory
  • Temporarily stores recent interactions and outcomes enabling agents to recall and utilize information relevant to their current context during the current executions.
• Long-Term Memory:
  • Preserves valuable insights and learnings from past executions, allowing agents to build and refine their knowledge over time.
  • So Agents can remeber what they did right and wrong across multiple executions
• Entity Memory:
  • Captures and organizes information about entities (people, places, concepts) encountered during tasks facilitating deeper understanding and relationship mapping
• Contextual Memory:
  • Maintains the context of interactions by combining ShortTermMemory, LongTermMemory, and EntityMemory, aiding in the coherence and relevance of agent responses over a sequence of tasks or a conversation.

 

 

Example: Configuring Memory for a Crew

from crewai import Crew, Agent, Task, Process

# Assemble your crew with memory capabilities
my_crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,
    memory=True,
    verbose=True
)

 

 

Additional Embedding Providers - Using OpenAI embeddings (already default)

from crewai import Crew, Agent, Task, Process

my_crew = Crew(
        agents=[...],
        tasks=[...],
        process=Process.sequential,
        memory=True,
        verbose=True,
        embedder={
                "provider": "openai",
                "config":{
                        "model": 'text-embedding-3-small'
                }
        }
)

 

 

2.1.7 Collaboration

Collaboration:

• Collaboration in CrewAI is fundamental, enabling agents to combine their skills, share information, and assist each other in task execution, embodying a truly cooperative ecosystem.
• Information Sharing: Ensures all agents are well-informed and can contribute effectively by sharing data and findings
• Task Assistance: Allows agents to seek help from peers with the required expertise for specific tasks.
• Resource Allocation: Optimizes task execution through the efficient distribution and sharing of resources among agents

 

 

Enhanced Attributes for Improved Collaboration:

• The Crew class has been enriched with several attributes to support advanced functionalities:
• Language Model Management (manager_llm, function_calling_llm):
  • Manages language models for executing tasks and tools, facilitating sophisticated agent-tool interactions
  • Note that while manager_llm is mandatory for hierarchical processes to ensure proper execution flow, function_calling_llm is optional, with a default value provided for streamlined tool interaction.
• Custom Manager Agent (manager_agent):
  • Allows specifying a custom agent as the manager instead of using the default manager provided by CrewAI.
• Process Flow (process):
  • Defines the execution logic (e.g., sequential, hierarchical) to streamline task distribution and execution

 

 

3. Create agents to research and write an article

Content-Planner, Content-Writer, Content-Editor 이 협업을 하는 Multiple Agent System 을 간단하게 구축해보자.

from crewai import Agent, Task, Crew

import os
from utils import get_openai_api_key

openai_api_key = get_openai_api_key()
os.environ["OPENAI_MODEL_NAME"] = 'gpt-3.5-turbo'

planner = Agent(
    role="Content Planner",
    goal="Plan engaging and factually accurate content on {topic}",
    backstory="You're working on planning a blog article "
              "about the topic: {topic}."
              "You collect information that helps the "
              "audience learn something "
              "and make informed decisions. "
              "Your work is the basis for "
              "the Content Writer to write an article on this topic.",
    allow_delegation=False,
    verbose=True
)

writer = Agent(
    role="Content Writer",
    goal="Write insightful and factually accurate "
         "opinion piece about the topic: {topic}",
    backstory="You're working on a writing "
              "a new opinion piece about the topic: {topic}. "
              "You base your writing on the work of "
              "the Content Planner, who provides an outline "
              "and relevant context about the topic. "
              "You follow the main objectives and "
              "direction of the outline, "
              "as provide by the Content Planner. "
              "You also provide objective and impartial insights "
              "and back them up with information "
              "provide by the Content Planner. "
              "You acknowledge in your opinion piece "
              "when your statements are opinions "
              "as opposed to objective statements.",
    allow_delegation=False,
    verbose=True
)

editor = Agent(
    role="Editor",
    goal="Edit a given blog post to align with "
         "the writing style of the organization. ",
    backstory="You are an editor who receives a blog post "
              "from the Content Writer. "
              "Your goal is to review the blog post "
              "to ensure that it follows journalistic best practices,"
              "provides balanced viewpoints "
              "when providing opinions or assertions, "
              "and also avoids major controversial topics "
              "or opinions when possible.",
    allow_delegation=False,
    verbose=True
)

plan = Task(
    description=(
        "1. Prioritize the latest trends, key players, "
            "and noteworthy news on {topic}.\n"
        "2. Identify the target audience, considering "
            "their interests and pain points.\n"
        "3. Develop a detailed content outline including "
            "an introduction, key points, and a call to action.\n"
        "4. Include SEO keywords and relevant data or sources."
    ),
    expected_output="A comprehensive content plan document "
        "with an outline, audience analysis, "
        "SEO keywords, and resources.",
    agent=planner,
)

write = Task(
    description=(
        "1. Use the content plan to craft a compelling "
            "blog post on {topic}.\n"
        "2. Incorporate SEO keywords naturally.\n"
        "3. Sections/Subtitles are properly named "
            "in an engaging manner.\n"
        "4. Ensure the post is structured with an "
            "engaging introduction, insightful body, "
            "and a summarizing conclusion.\n"
        "5. Proofread for grammatical errors and "
            "alignment with the brand's voice.\n"
    ),
    expected_output="A well-written blog post "
        "in markdown format, ready for publication, "
        "each section should have 2 or 3 paragraphs.",
    agent=writer,
)

edit = Task(
    description=("Proofread the given blog post for "
                 "grammatical errors and "
                 "alignment with the brand's voice."),
    expected_output="A well-written blog post in markdown format, "
                    "ready for publication, "
                    "each section should have 2 or 3 paragraphs.",
    agent=editor
)

crew = Crew(
    agents=[planner, writer, editor],
    tasks=[plan, write, edit],
    verbose=2
)

result = crew.kickoff(inputs={"topic": "Artificial Intelligence"})

from IPython.display import Markdown
Markdown(result)

 

 

4. Key elements of AI Agents

핵심 요소로는 Role-playing, Focus, Tools, Cooperation, Guararails, Memory 가 있음.

 

여기서는 Guardrails 에 대해서만 정리해보자.

 

Guardrails:

• 우리의 Agent 시스템이 할루시네이션을 일으키지 않고, 탈선하지 않고, random loop 에 빠지지 않도록 하고, 계속해서 같은 도구만을 사용하는 등의 행동을 하지 않도록 하고, 답변을 만드는데 너무 오래걸리는 등의 행동을 하지 않도록 하는 즉 정상 궤도에서 동작할 수 있도록 해주는 시스템 구성 요소를 말함.

 

5. Multi agent customer support automation

CrewAI 가 Multi Agent System 을 만드는데 중요한 요소라고 생각하는 다음의 것들을 조합해서 한번 시스템을 만들어볼거임.

• Role Playing
• Focus
• Tools
• Cooperation
• Guardrails
• Memory

 

 

CrewAI 에서는 Delegation 매커니즘이 있어서, Task 를 자신이 하기에 적절하지 않다라고 한다면 다른 Agent 에게 넘겨주는 매커니즘이 있음.

 

 

Agent 에게 Tool 을 줄 떄는 Agent Level 로 주거나, Task Level 로 줄 수 있음:

• Task Level 에서 준 Tool 은 언제나 Agent Level 의 Tool 을 override 한다.

 

 

여기서는 Customer Support Agent System 을 만들어봄:

# Warning control
import warnings
warnings.filterwarnings('ignore')

from crewai import Agent, Task, Crew

import os
from utils import get_openai_api_key

openai_api_key = get_openai_api_key()
os.environ["OPENAI_MODEL_NAME"] = 'gpt-3.5-turbo'

support_agent = Agent(
    role="Senior Support Representative",
    goal="Be the most friendly and helpful "
        "support representative in your team",
    backstory=(
        "You work at crewAI (https://crewai.com) and "
        " are now working on providing "
        "support to {customer}, a super important customer "
        " for your company."
        "You need to make sure that you provide the best support!"
        "Make sure to provide full complete answers, "
        " and make no assumptions."
    ),
    allow_delegation=False,
    verbose=True
)

support_quality_assurance_agent = Agent(
    role="Support Quality Assurance Specialist",
    goal="Get recognition for providing the "
    "best support quality assurance in your team",
    backstory=(
        "You work at crewAI (https://crewai.com) and "
        "are now working with your team "
        "on a request from {customer} ensuring that "
        "the support representative is "
        "providing the best support possible.\n"
        "You need to make sure that the support representative "
        "is providing full"
        "complete answers, and make no assumptions."
    ),
    verbose=True
)

from crewai_tools import SerperDevTool, \
                         ScrapeWebsiteTool, \
                         WebsiteSearchTool

docs_scrape_tool = ScrapeWebsiteTool(
    website_url="https://docs.crewai.com/how-to/Creating-a-Crew-and-kick-it-off/"
)

inquiry_resolution = Task(
    description=(
        "{customer} just reached out with a super important ask:\n"
        "{inquiry}\n\n"
        "{person} from {customer} is the one that reached out. "
        "Make sure to use everything you know "
        "to provide the best support possible."
        "You must strive to provide a complete "
        "and accurate response to the customer's inquiry."
    ),
    expected_output=(
        "A detailed, informative response to the "
        "customer's inquiry that addresses "
        "all aspects of their question.\n"
        "The response should include references "
        "to everything you used to find the answer, "
        "including external data or solutions. "
        "Ensure the answer is complete, "
        "leaving no questions unanswered, and maintain a helpful and friendly "
        "tone throughout."
    ),
    tools=[docs_scrape_tool],
    agent=support_agent,
)

quality_assurance_review = Task(
    description=(
        "Review the response drafted by the Senior Support Representative for {customer}'s inquiry. "
        "Ensure that the answer is comprehensive, accurate, and adheres to the "
        "high-quality standards expected for customer support.\n"
        "Verify that all parts of the customer's inquiry "
        "have been addressed "
        "thoroughly, with a helpful and friendly tone.\n"
        "Check for references and sources used to "
        " find the information, "
        "ensuring the response is well-supported and "
        "leaves no questions unanswered."
    ),
    expected_output=(
        "A final, detailed, and informative response "
        "ready to be sent to the customer.\n"
        "This response should fully address the "
        "customer's inquiry, incorporating all "
        "relevant feedback and improvements.\n"
        "Don't be too formal, we are a chill and cool company "
        "but maintain a professional and friendly tone throughout."
    ),
    agent=support_quality_assurance_agent,
)

crew = Crew(
  agents=[support_agent, support_quality_assurance_agent],
  tasks=[inquiry_resolution, quality_assurance_review],
  verbose=2,
  memory=True
)

inputs = {
    "customer": "DeepLearningAI",
    "person": "Andrew Ng",
    "inquiry": "I need help with setting up a Crew "
               "and kicking it off, specifically "
               "how can I add memory to my crew? "
               "Can you provide guidance?"
}

result = crew.kickoff(inputs=inputs)

from IPython.display import Markdown
Markdown(result)

 

 

6. Mental framework for agent creation

잘 설계된 Agent System 을 만들 때 필요한 도구로 Mental Framework 가 있음:

• 이건 자기가 작업의 매니저가 되보라고 하는 것. 좋은 매니저는 프로젝트를 성공적으로 완료할 수 있ㅇ르거니까. 여기서부터 시작하는 것.
• Step 1: "What is the goal", "What is the process"
• Step 2: "What kind of peoeple would i need to hire to get this done?

 

 

7. Key elements of agent tools

Great tool 의 3가지 조건:

• Versatile: 다양항 상황에 대해서 처리할 수 있는 도구여야함. AI 어플리케이션 특성상 Fuzzy input 이 들어올 수 있으니.
• Fault-tolerant: 단순히 툴 사용에 실패했을 때 Agent 가 멈추게 하는게 아니라 우아하게 대처할 수 있어야함. (다른 툴을 사용하거나, 에러 메시지를 본거나, 재시도를 하거나 등)
• Caching: Latency 를 줄이거나, 비용을 줄이거나, Rate Limits 에 대응하기 위해서 캐싱을 사용할 수 있을 것.

 

 

다음과 같이 Agent 의 workflow 에서 에러를 만났을 때 이를 인식하고, 사용할 수 없는 도구임을 알고, Fallback 적인 행동을 취하도록 해야함.

 

 

8. Key elements of well defined tasks.

Task 정의에서 중요한 것:

• Clear description of the task
• Set a clear and concise expectation
• Set a context
• Set a callback (Optional)
• Output 을 어떻게 할 건지 (e.g File, Json, Pydantic)
• Override Agent tool with specfic task tools
• Force human input before end of task
• Execute asynchronously
• Run in parallel

 

 

9. Multi agent Collaboration

Finalcial Analysis 작업을 위한 Multi agent Collaboration 을 구현해보자. 여기서는 Hierachical 방식으로 구현해볼거임.

# Warning control
import warnings
warnings.filterwarnings('ignore')

from crewai import Agent, Task, Crew

import os
from utils import get_openai_api_key, get_serper_api_key

openai_api_key = get_openai_api_key()
os.environ["OPENAI_MODEL_NAME"] = 'gpt-3.5-turbo'
os.environ["SERPER_API_KEY"] = get_serper_api_key()

from crewai_tools import ScrapeWebsiteTool, SerperDevTool

search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

data_analyst_agent = Agent(
    role="Data Analyst",
    goal="Monitor and analyze market data in real-time "
         "to identify trends and predict market movements.",
    backstory="Specializing in financial markets, this agent "
              "uses statistical modeling and machine learning "
              "to provide crucial insights. With a knack for data, "
              "the Data Analyst Agent is the cornerstone for "
              "informing trading decisions.",
    verbose=True,
    allow_delegation=True,
    tools = [scrape_tool, search_tool]
)

trading_strategy_agent = Agent(
    role="Trading Strategy Developer",
    goal="Develop and test various trading strategies based "
         "on insights from the Data Analyst Agent.",
    backstory="Equipped with a deep understanding of financial "
              "markets and quantitative analysis, this agent "
              "devises and refines trading strategies. It evaluates "
              "the performance of different approaches to determine "
              "the most profitable and risk-averse options.",
    verbose=True,
    allow_delegation=True,
    tools = [scrape_tool, search_tool]
)

execution_agent = Agent(
    role="Trade Advisor",
    goal="Suggest optimal trade execution strategies "
         "based on approved trading strategies.",
    backstory="This agent specializes in analyzing the timing, price, "
              "and logistical details of potential trades. By evaluating "
              "these factors, it provides well-founded suggestions for "
              "when and how trades should be executed to maximize "
              "efficiency and adherence to strategy.",
    verbose=True,
    allow_delegation=True,
    tools = [scrape_tool, search_tool]
)

risk_management_agent = Agent(
    role="Risk Advisor",
    goal="Evaluate and provide insights on the risks "
         "associated with potential trading activities.",
    backstory="Armed with a deep understanding of risk assessment models "
              "and market dynamics, this agent scrutinizes the potential "
              "risks of proposed trades. It offers a detailed analysis of "
              "risk exposure and suggests safeguards to ensure that "
              "trading activities align with the firm’s risk tolerance.",
    verbose=True,
    allow_delegation=True,
    tools = [scrape_tool, search_tool]
)

# Task for Data Analyst Agent: Analyze Market Data
data_analysis_task = Task(
    description=(
        "Continuously monitor and analyze market data for "
        "the selected stock ({stock_selection}). "
        "Use statistical modeling and machine learning to "
        "identify trends and predict market movements."
    ),
    expected_output=(
        "Insights and alerts about significant market "
        "opportunities or threats for {stock_selection}."
    ),
    agent=data_analyst_agent,
)

# Task for Trading Strategy Agent: Develop Trading Strategies
strategy_development_task = Task(
    description=(
        "Develop and refine trading strategies based on "
        "the insights from the Data Analyst and "
        "user-defined risk tolerance ({risk_tolerance}). "
        "Consider trading preferences ({trading_strategy_preference})."
    ),
    expected_output=(
        "A set of potential trading strategies for {stock_selection} "
        "that align with the user's risk tolerance."
    ),
    agent=trading_strategy_agent,
)

# Task for Trade Advisor Agent: Plan Trade Execution
execution_planning_task = Task(
    description=(
        "Analyze approved trading strategies to determine the "
        "best execution methods for {stock_selection}, "
        "considering current market conditions and optimal pricing."
    ),
    expected_output=(
        "Detailed execution plans suggesting how and when to "
        "execute trades for {stock_selection}."
    ),
    agent=execution_agent,
)

# Task for Risk Advisor Agent: Assess Trading Risks
risk_assessment_task = Task(
    description=(
        "Evaluate the risks associated with the proposed trading "
        "strategies and execution plans for {stock_selection}. "
        "Provide a detailed analysis of potential risks "
        "and suggest mitigation strategies."
    ),
    expected_output=(
        "A comprehensive risk analysis report detailing potential "
        "risks and mitigation recommendations for {stock_selection}."
    ),
    agent=risk_management_agent,
)

from crewai import Crew, Process
from langchain_openai import ChatOpenAI

# Define the crew with agents and tasks
financial_trading_crew = Crew(
    agents=[data_analyst_agent, 
            trading_strategy_agent, 
            execution_agent, 
            risk_management_agent],

    tasks=[data_analysis_task, 
           strategy_development_task, 
           execution_planning_task, 
           risk_assessment_task],

    manager_llm=ChatOpenAI(model="gpt-3.5-turbo", 
                           temperature=0.7),
    process=Process.hierarchical,
    verbose=True
)

# Example data for kicking off the process
financial_trading_inputs = {
    'stock_selection': 'AAPL',
    'initial_capital': '100000',
    'risk_tolerance': 'Medium',
    'trading_strategy_preference': 'Day Trading',
    'news_impact_consideration': True
}

### this execution will take some time to run
result = financial_trading_crew.kickoff(inputs=financial_trading_inputs)

from IPython.display import Markdown
Markdown(result)

 

 

Conclusion

Agent System 설계 과정:

• Mental Framework 
• 목적 Task
• 목적을 위해 필요한 Agents, 필요한 Tools
• Workflow Process

 

Tool 사용에 있어서 중요한 것:

• Fault-tolerant
• Versatile
• Caching

 

Task 정의에서 중요한 것:

• Clear description of the task
• Set a clear and concise expectation
• Set a context
• Set a callback (Optional)
• Output 을 어떻게 할 건지 (e.g File, Json, Pydantic)
• Override Agent tool with specfic task tools
• Force human input before end of task
• Execute asynchronously
• Run in parallel

 

Effective Agent System 을 만들기 위한 Tips:

• 명확한 Role 
• Mental Framework
• 피드백 매커니즘.
• Quality Assurance 를 위한 Agent
• 단계별로 처리하는 플로우 매커니즘