Spaces:
Configuration error
Configuration error
| # ========= 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. ========= | |
| 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 | |
| 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 | |
| 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 | |
| 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())) | |
| 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 | |
| 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={}, | |
| ) | |