这段代码定义了一个名为RolePlaying的类，用于两个角色之间的角色扮演。
该类接受以下参数：
- assistant_role_name (str): 助手扮演的角色名称。
- user_role_name (str): 用户扮演的角色名称。
- critic_role_name (str): 评论家扮演的角色名称。带有”:obj:
"human"“的角色名称会将评论家设置为Human代理，否则将创建一个CriticAgent。 - task_prompt (str, optional): 需要执行的任务的提示信息。
- with_task_specify (bool, optional): 是否使用任务指定代理。
- with_task_planner (bool, optional): 是否使用任务规划器代理。
- with_critic_in_the_loop (bool, optional): 是否在循环中包括评论家。
- critic_criteria (str, optional): 评论家代理的评估标准。如果未指定，则设置评估标准以提高任务性能。
- model_type (ModelType, optional): 用于角色扮演的模型类型。如果指定，将覆盖所有代理中的模型。
- task_type (TaskType, optional): 执行的任务类型。
- assistant_agent_kwargs (Dict, optional): 传递给助手代理的附加参数。
- user_agent_kwargs (Dict, optional): 传递给用户代理的附加参数。
- task_specify_agent_kwargs (Dict, optional): 传递给任务指定代理的附加参数。
- task_planner_agent_kwargs (Dict, optional): 传递给任务规划器代理的附加参数。
- critic_kwargs (Dict, optional): 传递给评论家的附加参数。
- sys_msg_generator_kwargs (Dict, optional): 传递给系统消息生成器的附加参数。
- extend_sys_msg_meta_dicts (List[Dict], optional): 用于扩展系统消息元数据字典的字典列表。
- extend_task_specify_meta_dict (Dict, optional): 用于扩展任务指定元数据字典的字典。
- output_language (str, optional): 代理输出的语言。
初始化时可以接收很多参数，下面解析一下各个参数的含义和用途：
- assistant_role_name (str): 机器人角色的名称，通常是assistant
- user_role_name (str): 用户角色的名称，通常是user
- critic_role_name: 批评家角色的名称，默认为critic
- task_prompt (str): 任务描述或者指示
- with_task_specify (bool): 是否使用TaskSpecifyAgent来生成特定的任务提示
- with_task_planner (bool): 是否使用TaskPlannerAgent来规划任务
- with_critic_in_the_loop (bool): 是否让批评家参与对话过程（默认不参与）
- critic_criteria (Optional[str]): 批评家参与时需要满足的条件，如不为空，则批评家只会在满足条件时参与对话
- model_type (Optional[ModelType]): 语言模型类型，如GPT-3等，可以为空
- task_type (TaskType): 任务类型，默认为AI社交任务
- assistant_agent_kwargs (Optional[Dict]): 机器人agent的配置
- user_agent_kwargs (Optional[Dict]): 用户agent的配置
- task_specify_agent_kwargs (Optional[Dict]): 任务指示agent的配置
- task_planner_agent_kwargs (Optional[Dict]): 任务规划agent的配置
- critic_kwargs (Optional[Dict]): 批评家agent的配置
- sys_msg_generator_kwargs (Optional[Dict]): 系统消息生成器的配置
- extend_sys_msg_meta_dicts (Optional[List[Dict]]): 系统消息的元数据
- extend_task_specify_meta_dict (Optional[Dict]): 任务指示的元数据
- output_language (Optional[str]): 输出的语言，可以为空
代码主要功能是初始化Chatbot，其中有一些重要的方法，如init_specified_task_prompt用来生成特定的任务提示，init_planned_task_prompt用来规划任务提示，还有init_critic来初始化批评家。整个类的逻辑比较复杂，需要根据不同的需求进行配置和使用。
init_planned_task_prompt方法用于使用任务规划(agent)来生成一个计划好的任务提示(task prompt)，并将其添加到原始任务提示(task prompt)后面。如果没有任务规划(agent)，则不会生成计划好的任务提示。
参数说明：
task_planner_agent_kwargs：任务规划(agent)的额外参数，在初始化TaskPlannerAgent时传递。output_language：输出语言，agents输出语言的设置。
方法实现：
- 根据
self.with_task_planner属性判断是否启用了任务规划。 - 如果启用了任务规划(agent)且指定了语言模型类型(
model_type)，则将model_type添加到task_planner_agent_kwargs中。 - 使用
TaskPlannerAgent初始化一个任务规划(agent)。 - 通过调用
task_planner_agent.run方法传入任务提示(task prompt)获取计划好的任务提示，并将其附加到任务提示中。 - 如果没有启用任务规划(agent)，则将
self.planned_task_prompt置为None。
get_sys_message_info方法用于获取初始的机器人和用户系统消息，并返回包含系统消息和元数据的元组。
参数说明：
assistant_role_name：机器人角色的名称。user_role_name：用户角色的名称。sys_msg_generator：系统消息生成器。extend_sys_msg_meta_dicts：扩展的系统消息元数据字典的列表。
方法实现：
- 创建一个由两个字典组成的列表，每个字典都包含任务提示(task prompt)作为键。
- 如果
extend_sys_msg_meta_dicts为空并且任务类型是AI社交任务或MISALIGNMENT任务，则创建一个包含角色名称的字典列表。 - 如果
extend_sys_msg_meta_dicts不为空，则将原始的系统消息元数据字典和扩展的系统消息元数据字典进行合并。 - 使用
sys_msg_generator.from_dicts方法将元数据字典转换为机器人和用户的初始系统消息。 - 返回机器人和用户的初始系统消息以及系统消息元数据字典的列表。
init_agents方法用于初始化助手(agent)和用户(agent)，并设置它们的初始系统消息。
参数说明：
init_assistant_sys_msg：助手(agent)的初始系统消息。assistant_agent_kwargs：传递给助手(agent)的额外参数。init_user_sys_msg：用户(agent)的初始系统消息。user_agent_kwargs：传递给用户(agent)的额外参数。output_language：agents输出语言的设置。
方法实现：
- 如果指定了语言模型类型(
model_type)，则将其添加到助手(agent)和用户(agent)的额外参数中。 - 使用
ChatAgent初始化助手(agent)，并将助手的系统消息保存在self.assistant_sys_msg中。 - 使用
ChatAgent初始化用户(agent)，并将用户的系统消息保存在self.user_sys_msg中。
init_critic方法用于初始化评论者(agent)。如果评论者角色名称是”human”，则创建一个人类评论者，否则根据指定的评论者准则创建一个评论者(agent)。如果没有指定评论者准则，则默认为改善任务性能。
参数说明：
critic_role_name：评论者所扮演角色的名称。critic_criteria：评论者(agent)的评价准则。如果没有指定，则将评价准则设置为改善任务性能。critic_kwargs：传递给评论者(agent)的额外参数。sys_msg_generator：用于生成系统消息的生成器。sys_msg_meta_dicts：系统消息元数据字典的列表。
方法实现：
- 如果启用了评论者(agent)循环且评论者角色名称是”human”，则创建一个人类评论者。
- 否则，根据指定的评论者准则和角色名称创建一个
CriticAgent评论者(agent)。如果没有指定评论者准则，则默认为改善任务性能。 - 如果指定了语言模型类型(
model_type)，则将其添加到评论者(agent)的额外参数中。 - 使用评论者准则、角色名称和系统消息元数据字典创建评论者的系统消息。
- 将评论者的系统消息保存在
self.critic_sys_msg中。 - 使用
CriticAgent初始化评论者(agent)。
init_chat方法用于初始化对话，通过重置助手(agent)和用户(agent)，并向它们发送系统消息来重新开始对话。返回助手的介绍性消息和用户的回复消息列表。
方法实现：
- 重置助手(agent)和用户(agent)。
- 使用chat消息将系统消息再次发送给助手(agent)和用户(agent)。
- 创建助手的介绍性消息，其中包含用户系统消息的内容和一些指令。
- 创建用户的回复消息，其内容为助手系统消息的内容。
- 使用用户的回复消息调用助手(agent)的
step方法，返回助手的响应消息。 - 如果助手(agent)的响应消息已结束或没有消息，则抛出
ValueError异常。 - 返回助手的介绍性消息和助手的响应消息列表。
reduce_message_options方法用于处理一系列对话消息，并返回处理后的消息。如果提供了多个消息且with_critic_in_the_loop为False，则抛出ValueError异常。如果未提供任何消息，则抛出ValueError异常。
方法实现：
- 如果消息数量为0，则抛出
ValueError异常。 - 如果消息数量大于1且
with_critic_in_the_loop为False，则抛出ValueError异常。 - 否则，如果启用了评论者(agent)循环且评论者存在，则调用评论者的
reduce_step方法，返回评论者的响应消息，并将其作为处理后的消息。 - 否则，将第一个消息作为处理后的消息返回。
RolePlaying类中的step方法。该方法用于推进对话，接收助手发来的消息，使用用户(agent)处理该消息，然后使用助手(agent)处理生成的回复消息。返回一个元组，其中包含助手生成的消息、助手是否终止了对话以及其他助手相关的信息；还有一个元组，其中包含用户生成的消息、用户是否终止了对话以及其他用户相关的信息。
方法实现：
- 使用助手(agent)发送的消息调用用户(agent)的step方法，返回用户(agent)的响应。
- 如果用户(agent)的响应已经终止或没有消息，则创建一个空的ChatAgentResponse结构体表示助手的消息，并将用户(agent)的终止信息和额外信息作为ChatAgentResponse结构体的一部分返回。
- 否则，根据用户(agent)的响应消息使用reduce_message_options方法处理获取用户的消息。
- 将处理后的用户消息提交给用户(agent)。
- 使用用户消息调用助手(agent)的step方法，返回助手(agent)的响应。
- 如果助手(agent)的响应已终止或没有消息，则创建一个空的ChatAgentResponse结构体表示用户的消息，并将助手(agent)的终止信息和额外信息作为ChatAgentResponse结构体的一部分返回。
- 否则，根据助手(agent)的响应消息使用reduce_message_options方法处理获取助手的消息。
- 将处理后的助手消息提交给助手(agent)。
- 返回一个元组，第一个元素为助手的响应消息、助手是否终止了对话以及其他助手相关的信息，第二个元素为用户的响应消息、用户是否终止了对话以及其他用户相关的信息。
源码如下：
# =========== 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. ===========
from typing import Dict, List, Optional, Sequence, Tuple, Union
from camel.agents import (
ChatAgent,
CriticAgent,
TaskPlannerAgent,
TaskSpecifyAgent,
)
from camel.agents.chat_agent import ChatAgentResponse
from camel.generators import SystemMessageGenerator
from camel.human import Human
from camel.messages import BaseMessage
from camel.prompts import TextPrompt
from camel.typing import ModelType, RoleType, TaskType
class RolePlaying:
r"""Role playing between two agents.
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.
critic_role_name (str): The name of the role played by the critic.
Role name with :obj:`"human"` will set critic as a :obj:`Human`
agent, else will create a :obj:`CriticAgent`.
(default: :obj:`"critic"`)
task_prompt (str, optional): A prompt for the task to be performed.
(default: :obj:`""`)
with_task_specify (bool, optional): Whether to use a task specify
agent. (default: :obj:`True`)
with_task_planner (bool, optional): Whether to use a task planner
agent. (default: :obj:`False`)
with_critic_in_the_loop (bool, optional): Whether to include a critic
in the loop. (default: :obj:`False`)
critic_criteria (str, optional): Critic criteria for the critic agent.
If not specified, set the criteria to improve task performance.
model_type (ModelType, optional): Model type that will be used for
role playing. If specified, it will override the model in all
agents. (default: :obj:`None`)
task_type (TaskType, optional): The type of task to perform.
(default: :obj:`TaskType.AI_SOCIETY`)
assistant_agent_kwargs (Dict, optional): Additional arguments to pass
to the assistant agent. (default: :obj:`None`)
user_agent_kwargs (Dict, optional): Additional arguments to pass to
the user agent. (default: :obj:`None`)
task_specify_agent_kwargs (Dict, optional): Additional arguments to
pass to the task specify agent. (default: :obj:`None`)
task_planner_agent_kwargs (Dict, optional): Additional arguments to
pass to the task planner agent. (default: :obj:`None`)
critic_kwargs (Dict, optional): Additional arguments to pass to the
critic. (default: :obj:`None`)
sys_msg_generator_kwargs (Dict, optional): Additional arguments to
pass to the system message generator. (default: :obj:`None`)
extend_sys_msg_meta_dicts (List[Dict], optional): A list of dicts to
extend the system message meta dicts with. (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`)
"""
def __init__(
self,
assistant_role_name: str,
user_role_name: str,
*,
critic_role_name: str = "critic",
task_prompt: str = "",
with_task_specify: bool = True,
with_task_planner: bool = False,
with_critic_in_the_loop: bool = False,
critic_criteria: Optional[str] = None,
model_type: Optional[ModelType] = None,
task_type: TaskType = TaskType.AI_SOCIETY,
assistant_agent_kwargs: Optional[Dict] = None,
user_agent_kwargs: Optional[Dict] = None,
task_specify_agent_kwargs: Optional[Dict] = None,
task_planner_agent_kwargs: Optional[Dict] = None,
critic_kwargs: Optional[Dict] = None,
sys_msg_generator_kwargs: Optional[Dict] = None,
extend_sys_msg_meta_dicts: Optional[List[Dict]] = None,
extend_task_specify_meta_dict: Optional[Dict] = None,
output_language: Optional[str] = None,
) -> None:
self.with_task_specify = with_task_specify
self.with_task_planner = with_task_planner
self.with_critic_in_the_loop = with_critic_in_the_loop
self.model_type = model_type
self.task_type = task_type
self.task_prompt = task_prompt
self.specified_task_prompt: Optional[TextPrompt] = None
self.init_specified_task_prompt(assistant_role_name, user_role_name,
task_specify_agent_kwargs,
extend_task_specify_meta_dict,
output_language)
self.planned_task_prompt: Optional[TextPrompt] = None
self.init_planned_task_prompt(task_planner_agent_kwargs,
output_language)
sys_msg_generator = SystemMessageGenerator(
task_type=self.task_type, **(sys_msg_generator_kwargs or {}))
(init_assistant_sys_msg, init_user_sys_msg,
sys_msg_meta_dicts) = self.get_sys_message_info(
assistant_role_name, user_role_name, sys_msg_generator,
extend_sys_msg_meta_dicts)
self.assistant_agent: ChatAgent
self.user_agent: ChatAgent
self.assistant_sys_msg: BaseMessage
self.user_sys_msg: BaseMessage
self.init_agents(
init_assistant_sys_msg,
assistant_agent_kwargs,
init_user_sys_msg,
user_agent_kwargs,
output_language,
)
self.critic: Optional[Union[CriticAgent, Human]] = None
self.critic_sys_msg: Optional[BaseMessage] = None
self.init_critic(critic_role_name, critic_criteria, critic_kwargs,
sys_msg_generator, sys_msg_meta_dicts)
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.
"""
if self.with_task_specify:
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 {})
if self.model_type is not None:
if task_specify_agent_kwargs is None:
task_specify_agent_kwargs = {}
task_specify_agent_kwargs.update(dict(model=self.model_type))
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,
)
self.task_prompt = self.specified_task_prompt
def init_planned_task_prompt(self,
task_planner_agent_kwargs: Optional[Dict],
output_language: Optional[str]):
r"""Use a task plan agent to append a planned task prompt to task
prompt. The planned task prompt is generated based on the task
prompt, which can be original task prompt or specified task prompt
if available. If there is no task plan agent, planned task prompt
will not be generated.
Args:
task_planner_agent_kwargs (Dict, optional): Additional arguments
to pass to the task planner agent.
output_language (str, optional): The language to be output by the
agents.
"""
if self.with_task_planner:
if self.model_type is not None:
if task_planner_agent_kwargs is None:
task_planner_agent_kwargs = {}
task_planner_agent_kwargs.update(dict(model=self.model_type))
task_planner_agent = TaskPlannerAgent(
output_language=output_language,
**(task_planner_agent_kwargs or {}),
)
self.planned_task_prompt = task_planner_agent.run(self.task_prompt)
self.task_prompt = (f"{self.task_prompt}\n"
f"{self.planned_task_prompt}")
else:
self.planned_task_prompt = None
def get_sys_message_info(
self,
assistant_role_name: str,
user_role_name: str,
sys_msg_generator: SystemMessageGenerator,
extend_sys_msg_meta_dicts: Optional[List[Dict]] = None,
) -> Tuple[BaseMessage, BaseMessage, List[Dict]]:
r"""Get initial assistant and user system message with a list of
system message meta dicts.
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.
sys_msg_generator (SystemMessageGenerator): A system message
generator for agents.
extend_sys_msg_meta_dicts (List[Dict], optional): A list of dicts
to extend the system message meta dicts with.
Returns:
A tuple containing a `BaseMessage` representing the assistant's
initial system message, a `BaseMessage` representing the user's
initial system message, and a list of system message meta dicts.
"""
sys_msg_meta_dicts = [dict(task=self.task_prompt) for _ in range(2)]
if (extend_sys_msg_meta_dicts is None and self.task_type in [
TaskType.AI_SOCIETY,
TaskType.MISALIGNMENT,
]):
extend_sys_msg_meta_dicts = [
dict(assistant_role=assistant_role_name,
user_role=user_role_name) for _ in range(2)
]
if extend_sys_msg_meta_dicts is not None:
sys_msg_meta_dicts = [{
**sys_msg_meta_dict,
**extend_sys_msg_meta_dict
} for sys_msg_meta_dict, extend_sys_msg_meta_dict in zip(
sys_msg_meta_dicts, extend_sys_msg_meta_dicts)]
init_assistant_sys_msg, init_user_sys_msg = (
sys_msg_generator.from_dicts(
meta_dicts=sys_msg_meta_dicts,
role_tuples=[
(assistant_role_name, RoleType.ASSISTANT),
(user_role_name, RoleType.USER),
],
))
return init_assistant_sys_msg, init_user_sys_msg, sys_msg_meta_dicts
def init_agents(
self,
init_assistant_sys_msg: BaseMessage,
assistant_agent_kwargs: Optional[Dict],
init_user_sys_msg: BaseMessage,
user_agent_kwargs: Optional[Dict],
output_language: Optional[str],
):
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.
init_user_sys_msg (BaseMessage): User agent's initial system
message.
user_agent_kwargs (Dict, optional): Additional arguments to
pass to the user agent.
output_language (str, optional): The language to be output by the
agents.
"""
if self.model_type is not None:
if assistant_agent_kwargs is None:
assistant_agent_kwargs = {}
assistant_agent_kwargs.update(dict(model=self.model_type))
if user_agent_kwargs is None:
user_agent_kwargs = {}
user_agent_kwargs.update(dict(model=self.model_type))
self.assistant_agent = ChatAgent(
init_assistant_sys_msg,
output_language=output_language,
**(assistant_agent_kwargs or {}),
)
self.assistant_sys_msg = self.assistant_agent.system_message
self.user_agent = ChatAgent(
init_user_sys_msg,
output_language=output_language,
**(user_agent_kwargs or {}),
)
self.user_sys_msg = self.user_agent.system_message
def init_critic(self, critic_role_name: str,
critic_criteria: Optional[str],
critic_kwargs: Optional[Dict],
sys_msg_generator: SystemMessageGenerator,
sys_msg_meta_dicts: List[Dict]):
r"""Initialize critic agent. If critic role name is :obj:`"human"`,
create a :obj:`Human` critic agent. Else, create a :obj:`CriticAgent`
critic agent with specified critic criteria. If the critic criteria
is not specified, set it to improve task performance.
Args:
critic_role_name (str): The name of the role played by the critic.
critic_criteria (str, optional): Critic criteria for the
critic agent. If not specified, set the criteria to
improve task performance.
critic_kwargs (Dict, optional): Additional arguments to
pass to the critic.
sys_msg_generator (SystemMessageGenerator): A system message
generator for agents.
sys_msg_meta_dicts (list): A list of system message meta dicts.
"""
if self.with_critic_in_the_loop:
if critic_role_name.lower() == "human":
self.critic = Human(**(critic_kwargs or {}))
else:
critic_criteria = (critic_criteria
or "improving the task performance")
critic_msg_meta_dict = dict(critic_role=critic_role_name,
criteria=critic_criteria,
**sys_msg_meta_dicts[0])
self.critic_sys_msg = sys_msg_generator.from_dict(
critic_msg_meta_dict,
role_tuple=(critic_role_name, RoleType.CRITIC),
)
if self.model_type is not None:
if critic_kwargs is None:
critic_kwargs = {}
critic_kwargs.update(dict(model=self.model_type))
self.critic = CriticAgent(
self.critic_sys_msg,
**(critic_kwargs or {}),
)
def init_chat(self) -> Tuple[BaseMessage, List[BaseMessage]]:
r"""Initializes the chat by resetting both of the assistant and user
agents, and sending the system messages again to the agents using
chat messages. Returns the assistant's introductory message and the
user's response messages.
Returns:
A tuple containing a `BaseMessage` representing the assistant's
introductory message, and a list of `BaseMessage` representing
the user's response messages.
"""
self.assistant_agent.reset()
self.user_agent.reset()
# Send the system messages again to the agents using chat messages
assistant_msg = BaseMessage.make_assistant_message(
role_name=self.assistant_sys_msg.role_name,
content=(f"{self.user_sys_msg.content}. "
"Now start to give me instructions one by one. "
"Only reply with Instruction and Input."))
user_msg = BaseMessage.make_user_message(
role_name=self.user_sys_msg.role_name,
content=f"{self.assistant_sys_msg.content}")
assistant_response = self.assistant_agent.step(user_msg)
if assistant_response.terminated or assistant_response.msgs is None:
raise ValueError(f"Assistant agent terminated unexpectedly. "
f"Error info: {assistant_response.info}")
return assistant_msg, assistant_response.msgs
def reduce_message_options(
self,
messages: Sequence[BaseMessage],
) -> BaseMessage:
r"""Processes a sequence of chat messages, returning the processed
message. If multiple messages are provided and
`with_critic_in_the_loop` is `False`, raises a `ValueError`.
If no messages are provided, a `ValueError` will be raised.
Args:
messages: A sequence of `BaseMessage` objects to process.
Returns:
A single `BaseMessage` representing the processed message.
"""
if len(messages) == 0:
raise ValueError("No messages to process.")
if len(messages) > 1 and not self.with_critic_in_the_loop:
raise ValueError("Got than one message to process. "
f"Num of messages: {len(messages)}.")
elif self.with_critic_in_the_loop and self.critic is not None:
critic_response = self.critic.reduce_step(messages)
processed_msg = critic_response.msg
else:
processed_msg = messages[0]
return processed_msg
def step(
self,
assistant_msg: BaseMessage,
) -> Tuple[ChatAgentResponse, ChatAgentResponse]:
r"""Advances the conversation by taking a message from the assistant,
processing it using the user agent, and then processing the resulting
message using the assistant agent. Returns a tuple containing the
resulting assistant message, whether the assistant agent terminated
the conversation, and any additional assistant information, as well as
a tuple containing the resulting user message, whether the user agent
terminated the conversation, and any additional user information.
Args:
assistant_msg: A `BaseMessage` representing the message from the
assistant.
Returns:
A tuple containing two ChatAgentResponse: the first struct contains
the resulting assistant message, whether the assistant agent
terminated the conversation, and any additional assistant
information; the second struct contains the resulting user message,
whether the user agent terminated the conversation, and any
additional user information.
"""
user_response = self.user_agent.step(assistant_msg)
if user_response.terminated or user_response.msgs is None:
return (ChatAgentResponse([], False, {}),
ChatAgentResponse([], user_response.terminated,
user_response.info))
user_msg = self.reduce_message_options(user_response.msgs)
self.user_agent.submit_message(user_msg)
assistant_response = self.assistant_agent.step(user_msg)
if assistant_response.terminated or assistant_response.msgs is None:
return (ChatAgentResponse([], assistant_response.terminated,
assistant_response.info),
ChatAgentResponse([user_msg], False, user_response.info))
assistant_msg = self.reduce_message_options(assistant_response.msgs)
self.assistant_agent.submit_message(assistant_msg)
return (
ChatAgentResponse([assistant_msg], assistant_response.terminated,
assistant_response.info),
ChatAgentResponse([user_msg], user_response.terminated,
user_response.info),
)