# =========== Copyright 2023 @ 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 @ CAMEL-AI.org. All Rights Reserved. ===========
import random
import warnings
from typing import Any, Dict, Optional, Sequence
from colorama import Fore
from camel.agents.chat_agent import ChatAgent
from camel.memories import AgentMemory
from camel.messages import BaseMessage
from camel.models import BaseModelBackend
from camel.responses import ChatAgentResponse
from camel.utils import get_first_int, print_text_animated
# AgentOps decorator setting
try:
import os
if os.getenv("AGENTOPS_API_KEY") is not None:
from agentops import track_agent
else:
raise ImportError
except (ImportError, AttributeError):
from camel.utils import track_agent
[docs]
@track_agent(name="CriticAgent")
class CriticAgent(ChatAgent):
r"""A class for the critic agent that assists in selecting an option.
Args:
system_message (BaseMessage): The system message for the critic
agent.
model (BaseModelBackend, optional): The model backend to use for
generating responses. (default: :obj:`OpenAIModel` with
`GPT_4O_MINI`)
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:`6`)
retry_attempts (int, optional): The number of retry attempts if the
critic fails to return a valid option. (default: :obj:`2`)
verbose (bool, optional): Whether to print the critic's messages.
logger_color (Any): The color of the menu options displayed to the
user. (default: :obj:`Fore.MAGENTA`)
"""
def __init__(
self,
system_message: BaseMessage,
model: Optional[BaseModelBackend] = None,
memory: Optional[AgentMemory] = None,
message_window_size: int = 6,
retry_attempts: int = 2,
verbose: bool = False,
logger_color: Any = Fore.MAGENTA,
) -> None:
super().__init__(
system_message,
model=model,
memory=memory,
message_window_size=message_window_size,
)
self.options_dict: Dict[str, str] = dict()
self.retry_attempts = retry_attempts
self.verbose = verbose
self.logger_color = logger_color
[docs]
def flatten_options(self, messages: Sequence[BaseMessage]) -> str:
r"""Flattens the options to the critic.
Args:
messages (Sequence[BaseMessage]): A list of `BaseMessage` objects.
Returns:
str: A string containing the flattened options to the critic.
"""
options = [message.content for message in messages]
flatten_options = (
f"> Proposals from "
f"{messages[0].role_name} ({messages[0].role_type}). "
"Please choose an option:\n"
)
for index, option in enumerate(options):
flatten_options += f"Option {index + 1}:\n{option}\n\n"
self.options_dict[str(index + 1)] = option
format = (
f"Please first enter your choice ([1-{len(self.options_dict)}]) "
"and then your explanation and comparison: "
)
return flatten_options + format
[docs]
def get_option(self, input_message: BaseMessage) -> str:
r"""Gets the option selected by the critic.
Args:
input_message (BaseMessage): A `BaseMessage` object representing
the input message.
Returns:
str: The option selected by the critic.
"""
# TODO: Add support for editing options by the critic.
msg_content = input_message.content
i = 0
while i < self.retry_attempts:
critic_response = self.step(input_message)
if critic_response.msgs is None or len(critic_response.msgs) == 0:
raise RuntimeError("Got None critic messages.")
if critic_response.terminated:
raise RuntimeError("Critic step failed.")
critic_msg = critic_response.msg
if self.verbose:
print_text_animated(
self.logger_color + "\n> Critic response: "
f"\x1b[3m{critic_msg.content}\x1b[0m\n"
)
choice = self.parse_critic(critic_msg)
if choice in self.options_dict:
return self.options_dict[choice]
else:
input_message = BaseMessage(
role_name=input_message.role_name,
role_type=input_message.role_type,
meta_dict=input_message.meta_dict,
content="> Invalid choice. Please choose again.\n"
+ msg_content,
)
i += 1
warnings.warn(
"Critic failed to get a valid option. "
f"After {self.retry_attempts} attempts. "
"Returning a random option."
)
return random.choice(list(self.options_dict.values()))
[docs]
def parse_critic(self, critic_msg: BaseMessage) -> Optional[str]:
r"""Parses the critic's message and extracts the choice.
Args:
critic_msg (BaseMessage): A `BaseMessage` object representing the
critic's response.
Returns:
Optional[str]: The critic's choice as a string, or None if the
message could not be parsed.
"""
choice = str(get_first_int(critic_msg.content))
return choice
[docs]
def reduce_step(
self,
input_messages: Sequence[BaseMessage],
) -> ChatAgentResponse:
r"""Performs one step of the conversation by flattening options to the
critic, getting the option, and parsing the choice.
Args:
input_messages (Sequence[BaseMessage]): A list of BaseMessage
objects.
Returns:
ChatAgentResponse: A `ChatAgentResponse` object includes the
critic's choice.
"""
meta_chat_message = BaseMessage(
role_name=input_messages[0].role_name,
role_type=input_messages[0].role_type,
meta_dict=input_messages[0].meta_dict,
content="",
)
flatten_options = self.flatten_options(input_messages)
if self.verbose:
print_text_animated(
self.logger_color + f"\x1b[3m{flatten_options}\x1b[0m\n"
)
input_msg = meta_chat_message.create_new_instance(flatten_options)
option = self.get_option(input_msg)
output_msg = meta_chat_message.create_new_instance(option)
# TODO: The return `info` can be improved.
return ChatAgentResponse(
msgs=[output_msg],
terminated=False,
info={},
)
