# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
from collections import deque
from typing import Dict, List, Optional
from camel.agents import (
ChatAgent,
TaskCreationAgent,
TaskPrioritizationAgent,
TaskSpecifyAgent,
)
from camel.agents.chat_agent import ChatAgentResponse
from camel.generators import SystemMessageGenerator
from camel.logger import get_logger
from camel.messages import BaseMessage
from camel.prompts import TextPrompt
from camel.types import RoleType, TaskType
logger = get_logger(__name__)
[docs]
class BabyAGI:
r"""The BabyAGI Agent adapted from `"Task-driven Autonomous Agent"
<https://github.com/yoheinakajima/babyagi>`_.
Args:
assistant_role_name (str): The name of the role played by the
assistant.
user_role_name (str): The name of the role played by the user.
task_prompt (str, optional): A prompt for the task to be performed.
(default: :obj:`""`)
task_type (TaskType, optional): The type of task to perform.
(default: :obj:`TaskType.AI_SOCIETY`)
max_task_history (int): The maximum number of previous tasks
information to include in the task agent.
(default: :obj:10)
assistant_agent_kwargs (Dict, optional): Additional arguments to pass
to the assistant agent. (default: :obj:`None`)
task_specify_agent_kwargs (Dict, optional): Additional arguments to
pass to the task specify agent. (default: :obj:`None`)
task_creation_agent_kwargs (Dict, optional): Additional arguments to
pass to the task creation agent. (default: :obj:`None`)
task_prioritization_agent_kwargs (Dict, optional): Additional arguments
to pass to the task prioritization agent. (default: :obj:`None`)
sys_msg_generator_kwargs (Dict, optional): Additional arguments to
pass to the system message generator. (default: :obj:`None`)
extend_task_specify_meta_dict (Dict, optional): A dict to extend the
task specify meta dict with. (default: :obj:`None`)
output_language (str, optional): The language to be output by the
agents. (default: :obj:`None`)
message_window_size (int, optional): The maximum number of previous
messages to include in the context window. If `None`, no windowing
is performed. (default: :obj:`None`)
"""
def __init__(
self,
assistant_role_name: str,
user_role_name: str,
task_prompt: str = "",
task_type: TaskType = TaskType.AI_SOCIETY,
max_task_history: int = 10,
assistant_agent_kwargs: Optional[Dict] = None,
task_specify_agent_kwargs: Optional[Dict] = None,
task_creation_agent_kwargs: Optional[Dict] = None,
task_prioritization_agent_kwargs: Optional[Dict] = None,
sys_msg_generator_kwargs: Optional[Dict] = None,
extend_task_specify_meta_dict: Optional[Dict] = None,
output_language: Optional[str] = None,
message_window_size: Optional[int] = None,
) -> None:
self.task_type = task_type
self.task_prompt = task_prompt
self.specified_task_prompt: TextPrompt
self.init_specified_task_prompt(
assistant_role_name,
user_role_name,
task_specify_agent_kwargs,
extend_task_specify_meta_dict,
output_language,
)
sys_msg_generator = SystemMessageGenerator(
task_type=self.task_type, **(sys_msg_generator_kwargs or {})
)
init_assistant_sys_msg = sys_msg_generator.from_dicts(
meta_dicts=[
dict(
assistant_role=assistant_role_name,
user_role=user_role_name,
task=self.specified_task_prompt,
)
],
role_tuples=[
(assistant_role_name, RoleType.ASSISTANT),
],
)
self.assistant_agent: ChatAgent
self.assistant_sys_msg: Optional[BaseMessage]
self.task_creation_agent: TaskCreationAgent
self.task_prioritization_agent: TaskPrioritizationAgent
self.init_agents(
init_assistant_sys_msg[0],
assistant_agent_kwargs,
task_creation_agent_kwargs,
task_prioritization_agent_kwargs,
output_language,
message_window_size,
)
self.subtasks: deque = deque([])
self.solved_subtasks: List[str] = []
self.MAX_TASK_HISTORY = max_task_history
[docs]
def init_specified_task_prompt(
self,
assistant_role_name: str,
user_role_name: str,
task_specify_agent_kwargs: Optional[Dict],
extend_task_specify_meta_dict: Optional[Dict],
output_language: Optional[str],
):
r"""Use a task specify agent to generate a specified task prompt.
Generated specified task prompt will be used to replace original
task prompt. If there is no task specify agent, specified task
prompt will not be generated.
Args:
assistant_role_name (str): The name of the role played by the
assistant.
user_role_name (str): The name of the role played by the user.
task_specify_agent_kwargs (Dict, optional): Additional arguments
to pass to the task specify agent.
extend_task_specify_meta_dict (Dict, optional): A dict to extend
the task specify meta dict with.
output_language (str, optional): The language to be output by the
agents.
"""
task_specify_meta_dict = dict()
if self.task_type in [TaskType.AI_SOCIETY, TaskType.MISALIGNMENT]:
task_specify_meta_dict.update(
dict(
assistant_role=assistant_role_name,
user_role=user_role_name,
)
)
task_specify_meta_dict.update(extend_task_specify_meta_dict or {})
task_specify_agent = TaskSpecifyAgent(
task_type=self.task_type,
output_language=output_language,
**(task_specify_agent_kwargs or {}),
)
self.specified_task_prompt = task_specify_agent.run(
self.task_prompt,
meta_dict=task_specify_meta_dict,
)
[docs]
def init_agents(
self,
init_assistant_sys_msg: BaseMessage,
assistant_agent_kwargs: Optional[Dict],
task_creation_agent_kwargs: Optional[Dict],
task_prioritization_agent_kwargs: Optional[Dict],
output_language: Optional[str],
message_window_size: Optional[int] = None,
):
r"""Initialize assistant and user agents with their system messages.
Args:
init_assistant_sys_msg (BaseMessage): Assistant agent's initial
system message.
assistant_agent_kwargs (Dict, optional): Additional arguments to
pass to the assistant agent.
task_creation_agent_kwargs (Dict, optional): Additional arguments
to pass to the task creation agent.
task_prioritization_agent_kwargs (Dict, optional): Additional
arguments to pass to the task prioritization agent.
output_language (str, optional): The language to be output by the
agents.
message_window_size (int, optional): The maximum number of previous
messages to include in the context window. If `None`, no
windowing is performed. (default: :obj:`None`)
"""
self.assistant_agent = ChatAgent(
init_assistant_sys_msg,
output_language=output_language,
message_window_size=message_window_size,
**(assistant_agent_kwargs or {}),
)
self.assistant_sys_msg = self.assistant_agent.system_message
self.assistant_agent.reset()
self.task_creation_agent = TaskCreationAgent(
objective=self.specified_task_prompt,
role_name=getattr(self.assistant_sys_msg, 'role_name', None)
or "assistant",
output_language=output_language,
message_window_size=message_window_size,
**(task_creation_agent_kwargs or {}),
)
self.task_creation_agent.reset()
self.task_prioritization_agent = TaskPrioritizationAgent(
objective=self.specified_task_prompt,
output_language=output_language,
message_window_size=message_window_size,
**(task_prioritization_agent_kwargs or {}),
)
self.task_prioritization_agent.reset()
[docs]
def step(self) -> ChatAgentResponse:
r"""BabyAGI agent would pull the first task from the task list,
complete the task based on the context, then creates new tasks and
re-prioritizes the task list based on the objective and the result of
the previous task. It returns assistant message.
Returns:
ChatAgentResponse: it contains the resulting assistant message,
whether the assistant agent terminated the conversation,
and any additional assistant information.
"""
if not self.subtasks:
new_subtask_list = self.task_creation_agent.run(task_list=[])
prioritized_subtask_list = self.task_prioritization_agent.run(
new_subtask_list
)
self.subtasks = deque(prioritized_subtask_list)
task_name = self.subtasks.popleft()
assistant_msg_msg = BaseMessage.make_user_message(
role_name=getattr(self.assistant_sys_msg, 'role_name', None)
or "assistant",
content=f"{task_name}",
)
assistant_response = self.assistant_agent.step(assistant_msg_msg)
assistant_msg = assistant_response.msgs[0]
self.solved_subtasks.append(task_name)
past_tasks = self.solved_subtasks + list(self.subtasks)
new_subtask_list = self.task_creation_agent.run(
task_list=past_tasks[-self.MAX_TASK_HISTORY :]
)
if new_subtask_list:
self.subtasks.extend(new_subtask_list)
prioritized_subtask_list = self.task_prioritization_agent.run(
task_list=list(self.subtasks)[-self.MAX_TASK_HISTORY :]
)
self.subtasks = deque(prioritized_subtask_list)
else:
logger.info("no new tasks")
assistant_response.info['task_name'] = task_name
assistant_response.info['subtasks'] = list(self.subtasks)
if not self.subtasks:
terminated = True
assistant_response.info['termination_reasons'] = (
"All tasks are solved"
)
return ChatAgentResponse(
msgs=[assistant_msg],
terminated=terminated,
info=assistant_response.info,
)
return ChatAgentResponse(
msgs=[assistant_msg],
terminated=assistant_response.terminated,
info=assistant_response.info,
)
