Towards AI

The leading AI community and content platform focused on making AI accessible to all. Check out our new course platform: https://academy.towardsai.net/courses/beginner-to-advanced-llm-dev

Follow publication

Building a Multi-Agent System to Accomplish Complex Tasks

Najib Sharifi, Ph.D.
Towards AI
Published in
8 min readMay 24, 2024

Image generated with copilate.
import os
import openai
from openai import OpenAI
from crewai_tools import SerperDevTool

os.environ["SERPER_API_KEY"] = "your serper api key goes here"
os.environ["OPENAI_API_KEY"] = "your open AI key goes here"

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
search_tool = SerperDevTool(api_key=os.getenv("SERPER_API_KEY"))
class Agent:
def __init__(self, name, role, backstory, goal, tools=None):
self.name = name
self.backstory = backstory
self.goal = goal
self.role = role
self.memory = []
self.tools = tools if tools else []
self.message_box = []
# adding memory for the agent to store recent tasks and outputs
def add_to_memory(self, entry):
self.memory.append(entry)

# sending messages to other agents
def send_message(self, recipient, message):
recipient.message_box.append((self.name, message))

# reading the messages sent from other agents before performing task
# this is done by removing messages from message box and added to memory
def read_messages(self):
while self.message_box:
sender, message = self.message_box.pop(0)
self.add_to_memory(f"message from the {sender}: {message}")

# we now define the function that will do the task assigned
# reading messages and adding task to the memory first
# the agent will take up the specialised role assigned and querry gpt3.5

def do_task(self, task, inputs):
self.read_messages()
task_info = task.info
self.add_to_memory(f"doing task: {task_info}")

'''for the research agent, the search_tool will be assigned to the agent
which it will be able to use to do a google search online'''


if 'search_tool' in self.tools:
search_query = task_info
search_results = search_tool.run(query=search_query)
inputs['search_results'] = search_results
task_info += f"\n\nsearch results:\n{search_results}"

llm_response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": f"you are a {self.role}. {self.backstory} Your goal is {self.goal}."},
{"role": "user", "content": task_info}
]
)
output = llm_response.choices[0].message.content
self.add_to_memory(f"task output: {output}")

return output
class TheArchitect:
def __init__(self, agents, tasks):
# dictionary of all agents based on name
self.agents = {agent.name: agent for agent in agents}
self.tasks = tasks

def process(self, inputs):
results = {}
current_result = None

for task in self.tasks:
task_agent = self.agents[task.agent.name]
'''to help with debugging and also checking flow of info
we can check/print which tasks are assigned to which agent'''

print(f"assignin task {task.name} to agent {task_agent.name}: {task.info}")

if current_result:
inputs['previous_result'] = current_result

if 'search' in inputs:
search_query = inputs['search']
search_results = search_tool.run(query=search_query)
inputs['search_results'] = search_results

agent_output = task_agent.do_task(task, inputs)
current_result = agent_output

# send the agent's output as a message to all other agents
for agent_name, agent in self.agents.items():
if agent_name != task_agent.name:
task_agent.send_message(agent, agent_output)

results[task.agent.name] = agent_output

return results
data_collector = Agent(
name="researcher",
role="Climate Data Collector",
goal="Collect comprehensive climate data from multiple sources.",
backstory="You gather climate data on temperature, carbon dioxide levels and other variables relevant to climate change, from reliable sources.",
tools=['search_tool']
)

data_analyst = Agent(
name="Data Scientist",
role="Climate Data Scientist",
goal="Analyse the collected climate data to identify significant trends.",
backstory="You analyse climate data to find significant trends and understand the impact of various factors on climate change.",
tools=[]
)

report_writer = Agent(
name="Report Writer",
role="Senior Scientific Report Writer",
goal="Generate a comprehensive report on climate change findings.",
backstory="You write detailed scientific reports based on the analysed climate data, highlighting key findings and implications.",
tools=[]
)

peer_reviewer = Agent(
name="Peer Reviewer",
role="Scientific Peer Reviewer",
goal="Review the scientific report for accuracy, clarity, and completeness.",
backstory="You review scientific reports to ensure they are accurate, clear, and meet the standards for scientific publication.",
tools=[]
)

final_report_writer = Agent(
name="Final Report Writer",
role="Final Report Writer",
goal="Incorporate peer review feedback and finalize the scientific report.",
backstory="You finalize the scientific report by incorporating feedback from peer reviewer and ensure it is publication ready.",
tools=[]
)
analyse_data = Task(
info=(
"Using the following climate data, analyze for trends and patterns:\n{previous_result}\n"
"1. Identify significant trends in temperature, CO2 levels, and precipitation.\n"
"2. Determine potential causes of observed trends.\n"
"3. Summarize key findings in a detailed analysis report."
),
expected_output="Detailed analysis report on climate data trends and potential causes.",
agent=data_analyst,
name="Data Analysis"
)

write_report = Task(
info=(
"Using the following analysis report, write a comprehensive scientific report on climate change findings:\n{previous_result}\n"
"1. Include an introduction, methodology, results, discussion, and conclusion.\n"
"2. Use clear and precise language suitable for a scientific audience.\n"
"3. Ensure all findings are supported by data and analysis."
),
expected_output="Comprehensive scientific report on climate change findings.",
agent=report_writer,
name="Report Writing"
)

review_report = Task(
info=(
"Using the following scientific report, review for accuracy, clarity, and completeness:\n{previous_result}\n"
"1. Ensure the report adheres to scientific standards.\n"
"2. Check for any errors or inaccuracies in data and analysis.\n"
"3. Provide feedback and suggestions for improvement."
),
expected_output="Reviewed and revised scientific report, ready for publication.",
agent=peer_reviewer,
name="Peer Review"
)

finalize_report = Task(
info=(
"Using the following peer-reviewed report, incorporate feedback and finalize the scientific report:\n{previous_result}\n"
"1. Address all feedback and suggestions provided by the peer reviewer.\n"
"2. Ensure the report is polished and ready for publication.\n"
"3. Provide the final version of the scientific report."
),
expected_output="Finalized scientific report, ready for publication.",
agent=final_report_writer,
name="Finalize Report"
)
ClimateResearchSystem = TheArchitect(
agents=[data_collector, data_analyst, report_writer, peer_reviewer, final_report_writer],
tasks=[collect_data, analyse_data, write_report, review_report, finalize_report]
)


result = ClimateResearchSystem.process(inputs={
"topic": "Climate Change",
"search": "latest climate data trends"
})
from IPython.display import Markdown
Markdown(result['Final Report Writer'])

Published in Towards AI

The leading AI community and content platform focused on making AI accessible to all. Check out our new course platform: https://academy.towardsai.net/courses/beginner-to-advanced-llm-dev

Written by Najib Sharifi, Ph.D.

PhD in Molecular Chemistry | Machine Learning Engineer.

No responses yet

Write a response