From ec81f87c89de86bc9f7590188257ea9527364dff Mon Sep 17 00:00:00 2001 From: Mark Sze <66362098+marklysze@users.noreply.github.com> Date: Tue, 30 Apr 2024 14:16:08 +1000 Subject: [PATCH] Re-query speaker name when multiple speaker names returned during Group Chat speaker selection (#2304) * Added requery_on_multiple_speaker_names to GroupChat and updated _finalize_speaker to requery on multiple speaker names (if enabled) * Removed unnecessary comments * Update to current main * Tweak error message. * Comment clarity * Expanded description of Group Chat requery_on_multiple_speaker_names * Reworked to two-way nested chat for speaker selection with default of 2 retries. * Adding validation of new GroupChat attributes * Updates as per @ekzhu's suggestions * Update groupchat - Added select_speaker_auto_multiple_template and select_speaker_auto_none_template - Added max_attempts comment - Re-instated support for role_for_select_speaker_messages - * Update conversable_agent.py Added ability to force override role for a message to support select speaker prompt. * Update test_groupchat.py Updated existing select_speaker test functions as underlying approach has changed, added necessary tests for new functionality. * Removed block for manual selection in select_speaker function. * Catered for no-selection during manual selection mode --------- Co-authored-by: Chi Wang --- autogen/agentchat/conversable_agent.py | 5 + autogen/agentchat/groupchat.py | 390 +++++++++++++++++++++++-- test/agentchat/test_groupchat.py | 369 ++++++++++++++++++++++- 3 files changed, 732 insertions(+), 32 deletions(-) diff --git a/autogen/agentchat/conversable_agent.py b/autogen/agentchat/conversable_agent.py index 262fc513d235..b04222f514e2 100644 --- a/autogen/agentchat/conversable_agent.py +++ b/autogen/agentchat/conversable_agent.py @@ -576,6 +576,11 @@ def _append_oai_message(self, message: Union[Dict, str], role, conversation_id: if message.get("role") in ["function", "tool"]: oai_message["role"] = message.get("role") + elif "override_role" in message: + # If we have a direction to override the role then set the + # role accordingly. Used to customise the role for the + # select speaker prompt. + oai_message["role"] = message.get("override_role") else: oai_message["role"] = role diff --git a/autogen/agentchat/groupchat.py b/autogen/agentchat/groupchat.py index f5b6106863aa..864924550803 100644 --- a/autogen/agentchat/groupchat.py +++ b/autogen/agentchat/groupchat.py @@ -7,6 +7,7 @@ from ..code_utils import content_str from ..exception_utils import AgentNameConflict, NoEligibleSpeaker, UndefinedNextAgent +from ..formatting_utils import colored from ..graph_utils import check_graph_validity, invert_disallowed_to_allowed from ..io.base import IOStream from ..runtime_logging import log_new_agent, logging_enabled @@ -28,13 +29,28 @@ class GroupChat: When set to True and when a message is a function call suggestion, the next speaker will be chosen from an agent which contains the corresponding function name in its `function_map`. - - select_speaker_message_template: customize the select speaker message (used in "auto" speaker selection), which appears first in the message context and generally includes the agent descriptions and list of agents. The string value will be converted to an f-string, use "{roles}" to output the agent's and their role descriptions and "{agentlist}" for a comma-separated list of agent names in square brackets. The default value is: + - select_speaker_message_template: customize the select speaker message (used in "auto" speaker selection), which appears first in the message context and generally includes the agent descriptions and list of agents. If the string contains "{roles}" it will replaced with the agent's and their role descriptions. If the string contains "{agentlist}" it will be replaced with a comma-separated list of agent names in square brackets. The default value is: "You are in a role play game. The following roles are available: {roles}. Read the following conversation. Then select the next role from {agentlist} to play. Only return the role." - - select_speaker_prompt_template: customize the select speaker prompt (used in "auto" speaker selection), which appears last in the message context and generally includes the list of agents and guidance for the LLM to select the next agent. The string value will be converted to an f-string, use "{agentlist}" for a comma-separated list of agent names in square brackets. The default value is: + - select_speaker_prompt_template: customize the select speaker prompt (used in "auto" speaker selection), which appears last in the message context and generally includes the list of agents and guidance for the LLM to select the next agent. If the string contains "{agentlist}" it will be replaced with a comma-separated list of agent names in square brackets. The default value is: "Read the above conversation. Then select the next role from {agentlist} to play. Only return the role." + - select_speaker_auto_multiple_template: customize the follow-up prompt used when selecting a speaker fails with a response that contains multiple agent names. This prompt guides the LLM to return just one agent name. Applies only to "auto" speaker selection method. If the string contains "{agentlist}" it will be replaced with a comma-separated list of agent names in square brackets. The default value is: + "You provided more than one name in your text, please return just the name of the next speaker. To determine the speaker use these prioritised rules: + 1. If the context refers to themselves as a speaker e.g. "As the..." , choose that speaker's name + 2. If it refers to the "next" speaker name, choose that name + 3. Otherwise, choose the first provided speaker's name in the context + The names are case-sensitive and should not be abbreviated or changed. + Respond with ONLY the name of the speaker and DO NOT provide a reason." + - select_speaker_auto_none_template: customize the follow-up prompt used when selecting a speaker fails with a response that contains no agent names. This prompt guides the LLM to return an agent name and provides a list of agent names. Applies only to "auto" speaker selection method. If the string contains "{agentlist}" it will be replaced with a comma-separated list of agent names in square brackets. The default value is: + "You didn't choose a speaker. As a reminder, to determine the speaker use these prioritised rules: + 1. If the context refers to themselves as a speaker e.g. "As the..." , choose that speaker's name + 2. If it refers to the "next" speaker name, choose that name + 3. Otherwise, choose the first provided speaker's name in the context + The names are case-sensitive and should not be abbreviated or changed. + The only names that are accepted are {agentlist}. + Respond with ONLY the name of the speaker and DO NOT provide a reason." - speaker_selection_method: the method for selecting the next speaker. Default is "auto". Could be any of the following (case insensitive), will raise ValueError if not recognized: - "auto": the next speaker is selected automatically by LLM. @@ -51,6 +67,15 @@ def custom_speaker_selection_func( last_speaker: Agent, groupchat: GroupChat ) -> Union[Agent, str, None]: ``` + - max_retries_for_selecting_speaker: the maximum number of times the speaker selection requery process will run. + If, during speaker selection, multiple agent names or no agent names are returned by the LLM as the next agent, it will be queried again up to the maximum number + of times until a single agent is returned or it exhausts the maximum attempts. + Applies only to "auto" speaker selection method. + Default is 2. + - select_speaker_auto_verbose: whether to output the select speaker responses and selections + If set to True, the outputs from the two agents in the nested select speaker chat will be output, along with + whether the responses were successful, or not, in selecting an agent + Applies only to "auto" speaker selection method. - allow_repeat_speaker: whether to allow the same speaker to speak consecutively. Default is True, in which case all speakers are allowed to speak consecutively. If `allow_repeat_speaker` is a list of Agents, then only those listed agents are allowed to repeat. @@ -77,6 +102,7 @@ def custom_speaker_selection_func( admin_name: Optional[str] = "Admin" func_call_filter: Optional[bool] = True speaker_selection_method: Union[Literal["auto", "manual", "random", "round_robin"], Callable] = "auto" + max_retries_for_selecting_speaker: Optional[int] = 2 allow_repeat_speaker: Optional[Union[bool, List[Agent]]] = None allowed_or_disallowed_speaker_transitions: Optional[Dict] = None speaker_transitions_type: Literal["allowed", "disallowed", None] = None @@ -89,6 +115,20 @@ def custom_speaker_selection_func( select_speaker_prompt_template: str = ( "Read the above conversation. Then select the next role from {agentlist} to play. Only return the role." ) + select_speaker_auto_multiple_template: str = """You provided more than one name in your text, please return just the name of the next speaker. To determine the speaker use these prioritised rules: + 1. If the context refers to themselves as a speaker e.g. "As the..." , choose that speaker's name + 2. If it refers to the "next" speaker name, choose that name + 3. Otherwise, choose the first provided speaker's name in the context + The names are case-sensitive and should not be abbreviated or changed. + Respond with ONLY the name of the speaker and DO NOT provide a reason.""" + select_speaker_auto_none_template: str = """You didn't choose a speaker. As a reminder, to determine the speaker use these prioritised rules: + 1. If the context refers to themselves as a speaker e.g. "As the..." , choose that speaker's name + 2. If it refers to the "next" speaker name, choose that name + 3. Otherwise, choose the first provided speaker's name in the context + The names are case-sensitive and should not be abbreviated or changed. + The only names that are accepted are {agentlist}. + Respond with ONLY the name of the speaker and DO NOT provide a reason.""" + select_speaker_auto_verbose: Optional[bool] = False role_for_select_speaker_messages: Optional[str] = "system" _VALID_SPEAKER_SELECTION_METHODS = ["auto", "manual", "random", "round_robin"] @@ -178,7 +218,7 @@ def __post_init__(self): agents=self.agents, ) - # Check select_speaker_message_template and select_speaker_prompt_template have values + # Check select speaker messages, prompts, roles, and retries have values if self.select_speaker_message_template is None or len(self.select_speaker_message_template) == 0: raise ValueError("select_speaker_message_template cannot be empty or None.") @@ -188,6 +228,27 @@ def __post_init__(self): if self.role_for_select_speaker_messages is None or len(self.role_for_select_speaker_messages) == 0: raise ValueError("role_for_select_speaker_messages cannot be empty or None.") + if self.select_speaker_auto_multiple_template is None or len(self.select_speaker_auto_multiple_template) == 0: + raise ValueError("select_speaker_auto_multiple_template cannot be empty or None.") + + if self.select_speaker_auto_none_template is None or len(self.select_speaker_auto_none_template) == 0: + raise ValueError("select_speaker_auto_none_template cannot be empty or None.") + + if self.max_retries_for_selecting_speaker is None or len(self.role_for_select_speaker_messages) == 0: + raise ValueError("role_for_select_speaker_messages cannot be empty or None.") + + # Validate max select speakers retries + if self.max_retries_for_selecting_speaker is None or not isinstance( + self.max_retries_for_selecting_speaker, int + ): + raise ValueError("max_retries_for_selecting_speaker cannot be None or non-int") + elif self.max_retries_for_selecting_speaker < 0: + raise ValueError("max_retries_for_selecting_speaker must be greater than or equal to zero") + + # Validate select_speaker_auto_verbose + if self.select_speaker_auto_verbose is None or not isinstance(self.select_speaker_auto_verbose, bool): + raise ValueError("select_speaker_auto_verbose cannot be None or non-bool") + @property def agent_names(self) -> List[str]: """Return the names of the agents in the group chat.""" @@ -450,33 +511,34 @@ def _prepare_and_select_agents( select_speaker_messages[-1] = dict(select_speaker_messages[-1], function_call=None) if select_speaker_messages[-1].get("tool_calls", False): select_speaker_messages[-1] = dict(select_speaker_messages[-1], tool_calls=None) - select_speaker_messages = select_speaker_messages + [ - { - "role": self.role_for_select_speaker_messages, - "content": self.select_speaker_prompt(graph_eligible_agents), - } - ] return selected_agent, graph_eligible_agents, select_speaker_messages def select_speaker(self, last_speaker: Agent, selector: ConversableAgent) -> Agent: - """Select the next speaker.""" + """Select the next speaker (with requery).""" + + # Prepare the list of available agents and select an agent if selection method allows (non-auto) selected_agent, agents, messages = self._prepare_and_select_agents(last_speaker) if selected_agent: return selected_agent - # auto speaker selection - selector.update_system_message(self.select_speaker_msg(agents)) - final, name = selector.generate_oai_reply(messages) - return self._finalize_speaker(last_speaker, final, name, agents) + elif self.speaker_selection_method == "manual": + # An agent has not been selected while in manual mode, so move to the next agent + return self.next_agent(last_speaker) + + # auto speaker selection with 2-agent chat + return self._auto_select_speaker(last_speaker, selector, messages, agents) async def a_select_speaker(self, last_speaker: Agent, selector: ConversableAgent) -> Agent: - """Select the next speaker.""" + """Select the next speaker (with requery), asynchronously.""" + selected_agent, agents, messages = self._prepare_and_select_agents(last_speaker) if selected_agent: return selected_agent - # auto speaker selection - selector.update_system_message(self.select_speaker_msg(agents)) - final, name = await selector.a_generate_oai_reply(messages) - return self._finalize_speaker(last_speaker, final, name, agents) + elif self.speaker_selection_method == "manual": + # An agent has not been selected while in manual mode, so move to the next agent + return self.next_agent(last_speaker) + + # auto speaker selection with 2-agent chat + return await self.a_auto_select_speaker(last_speaker, selector, messages, agents) def _finalize_speaker(self, last_speaker: Agent, final: bool, name: str, agents: Optional[List[Agent]]) -> Agent: if not final: @@ -496,6 +558,296 @@ def _finalize_speaker(self, last_speaker: Agent, final: bool, name: str, agents: agent = self.agent_by_name(name) return agent if agent else self.next_agent(last_speaker, agents) + def _auto_select_speaker( + self, + last_speaker: Agent, + selector: ConversableAgent, + messages: Optional[List[Dict]], + agents: Optional[List[Agent]], + ) -> Agent: + """Selects next speaker for the "auto" speaker selection method. Utilises its own two-agent chat to determine the next speaker and supports requerying. + + Speaker selection for "auto" speaker selection method: + 1. Create a two-agent chat with a speaker selector agent and a speaker validator agent, like a nested chat + 2. Inject the group messages into the new chat + 3. Run the two-agent chat, evaluating the result of response from the speaker selector agent: + - If a single agent is provided then we return it and finish. If not, we add an additional message to this nested chat in an attempt to guide the LLM to a single agent response + 4. Chat continues until a single agent is nominated or there are no more attempts left + 5. If we run out of turns and no single agent can be determined, the next speaker in the list of agents is returned + + Args: + last_speaker Agent: The previous speaker in the group chat + selector ConversableAgent: + messages Optional[List[Dict]]: Current chat messages + agents Optional[List[Agent]]: Valid list of agents for speaker selection + + Returns: + Dict: a counter for mentioned agents. + """ + + # If no agents are passed in, assign all the group chat's agents + if agents is None: + agents = self.agents + + # The maximum number of speaker selection attempts (including requeries) + # is the initial speaker selection attempt plus the maximum number of retries. + # We track these and use them in the validation function as we can't + # access the max_turns from within validate_speaker_name. + max_attempts = 1 + self.max_retries_for_selecting_speaker + attempts_left = max_attempts + attempt = 0 + + # Registered reply function for checking_agent, checks the result of the response for agent names + def validate_speaker_name(recipient, messages, sender, config) -> Tuple[bool, Union[str, Dict, None]]: + + # The number of retries left, starting at max_retries_for_selecting_speaker + nonlocal attempts_left + nonlocal attempt + + attempt = attempt + 1 + attempts_left = attempts_left - 1 + + return self._validate_speaker_name(recipient, messages, sender, config, attempts_left, attempt, agents) + + # Two-agent chat for speaker selection + + # Agent for checking the response from the speaker_select_agent + checking_agent = ConversableAgent("checking_agent", default_auto_reply=max_attempts) + + # Register the speaker validation function with the checking agent + checking_agent.register_reply( + [ConversableAgent, None], + reply_func=validate_speaker_name, # Validate each response + remove_other_reply_funcs=True, + ) + + # Agent for selecting a single agent name from the response + speaker_selection_agent = ConversableAgent( + "speaker_selection_agent", + system_message=self.select_speaker_msg(agents), + chat_messages={checking_agent: messages}, + llm_config=selector.llm_config, + human_input_mode="NEVER", # Suppresses some extra terminal outputs, outputs will be handled by select_speaker_auto_verbose + ) + + # Run the speaker selection chat + result = checking_agent.initiate_chat( + speaker_selection_agent, + cache=None, # don't use caching for the speaker selection chat + message={ + "content": self.select_speaker_prompt(agents), + "override_role": self.role_for_select_speaker_messages, + }, + max_turns=2 + * max(1, max_attempts), # Limiting the chat to the number of attempts, including the initial one + clear_history=False, + silent=not self.select_speaker_auto_verbose, # Base silence on the verbose attribute + ) + + return self._process_speaker_selection_result(result, last_speaker, agents) + + async def a_auto_select_speaker( + self, + last_speaker: Agent, + selector: ConversableAgent, + messages: Optional[List[Dict]], + agents: Optional[List[Agent]], + ) -> Agent: + """(Asynchronous) Selects next speaker for the "auto" speaker selection method. Utilises its own two-agent chat to determine the next speaker and supports requerying. + + Speaker selection for "auto" speaker selection method: + 1. Create a two-agent chat with a speaker selector agent and a speaker validator agent, like a nested chat + 2. Inject the group messages into the new chat + 3. Run the two-agent chat, evaluating the result of response from the speaker selector agent: + - If a single agent is provided then we return it and finish. If not, we add an additional message to this nested chat in an attempt to guide the LLM to a single agent response + 4. Chat continues until a single agent is nominated or there are no more attempts left + 5. If we run out of turns and no single agent can be determined, the next speaker in the list of agents is returned + + Args: + last_speaker Agent: The previous speaker in the group chat + selector ConversableAgent: + messages Optional[List[Dict]]: Current chat messages + agents Optional[List[Agent]]: Valid list of agents for speaker selection + + Returns: + Dict: a counter for mentioned agents. + """ + + # If no agents are passed in, assign all the group chat's agents + if agents is None: + agents = self.agents + + # The maximum number of speaker selection attempts (including requeries) + # We track these and use them in the validation function as we can't + # access the max_turns from within validate_speaker_name + max_attempts = 1 + self.max_retries_for_selecting_speaker + attempts_left = max_attempts + attempt = 0 + + # Registered reply function for checking_agent, checks the result of the response for agent names + def validate_speaker_name(recipient, messages, sender, config) -> Tuple[bool, Union[str, Dict, None]]: + + # The number of retries left, starting at max_retries_for_selecting_speaker + nonlocal attempts_left + nonlocal attempt + + attempt = attempt + 1 + attempts_left = attempts_left - 1 + + return self._validate_speaker_name(recipient, messages, sender, config, attempts_left, attempt, agents) + + # Two-agent chat for speaker selection + + # Agent for checking the response from the speaker_select_agent + checking_agent = ConversableAgent("checking_agent", default_auto_reply=max_attempts) + + # Register the speaker validation function with the checking agent + checking_agent.register_reply( + [ConversableAgent, None], + reply_func=validate_speaker_name, # Validate each response + remove_other_reply_funcs=True, + ) + + # Agent for selecting a single agent name from the response + speaker_selection_agent = ConversableAgent( + "speaker_selection_agent", + system_message=self.select_speaker_msg(agents), + chat_messages={checking_agent: messages}, + llm_config=selector.llm_config, + human_input_mode="NEVER", # Suppresses some extra terminal outputs, outputs will be handled by select_speaker_auto_verbose + ) + + # Run the speaker selection chat + result = await checking_agent.a_initiate_chat( + speaker_selection_agent, + cache=None, # don't use caching for the speaker selection chat + message=self.select_speaker_prompt(agents), + max_turns=2 + * max(1, max_attempts), # Limiting the chat to the number of attempts, including the initial one + clear_history=False, + silent=not self.select_speaker_auto_verbose, # Base silence on the verbose attribute + ) + + return self._process_speaker_selection_result(result, last_speaker, agents) + + def _validate_speaker_name( + self, recipient, messages, sender, config, attempts_left, attempt, agents + ) -> Tuple[bool, Union[str, Dict, None]]: + """Validates the speaker response for each round in the internal 2-agent + chat within the auto select speaker method. + + Used by auto_select_speaker and a_auto_select_speaker. + """ + + # Output the query and requery results + if self.select_speaker_auto_verbose: + iostream = IOStream.get_default() + + # Validate the speaker name selected + select_name = messages[-1]["content"].strip() + + mentions = self._mentioned_agents(select_name, agents) + + if len(mentions) == 1: + + # Success on retry, we have just one name mentioned + selected_agent_name = next(iter(mentions)) + + # Add the selected agent to the response so we can return it + messages.append({"role": "user", "content": f"[AGENT SELECTED]{selected_agent_name}"}) + + if self.select_speaker_auto_verbose: + iostream.print( + colored( + f">>>>>>>> Select speaker attempt {attempt} of {attempt + attempts_left} successfully selected: {selected_agent_name}", + "green", + ), + flush=True, + ) + + elif len(mentions) > 1: + # More than one name on requery so add additional reminder prompt for next retry + + if self.select_speaker_auto_verbose: + iostream.print( + colored( + f">>>>>>>> Select speaker attempt {attempt} of {attempt + attempts_left} failed as it included multiple agent names.", + "red", + ), + flush=True, + ) + + if attempts_left: + # Message to return to the chat for the next attempt + agentlist = f"{[agent.name for agent in agents]}" + + return True, { + "content": self.select_speaker_auto_multiple_template.format(agentlist=agentlist), + "override_role": self.role_for_select_speaker_messages, + } + else: + # Final failure, no attempts left + messages.append( + { + "role": "user", + "content": f"[AGENT SELECTION FAILED]Select speaker attempt #{attempt} of {attempt + attempts_left} failed as it returned multiple names.", + } + ) + + else: + # No names at all on requery so add additional reminder prompt for next retry + + if self.select_speaker_auto_verbose: + iostream.print( + colored( + f">>>>>>>> Select speaker attempt #{attempt} failed as it did not include any agent names.", + "red", + ), + flush=True, + ) + + if attempts_left: + # Message to return to the chat for the next attempt + agentlist = f"{[agent.name for agent in agents]}" + + return True, { + "content": self.select_speaker_auto_none_template.format(agentlist=agentlist), + "override_role": self.role_for_select_speaker_messages, + } + else: + # Final failure, no attempts left + messages.append( + { + "role": "user", + "content": f"[AGENT SELECTION FAILED]Select speaker attempt #{attempt} of {attempt + attempts_left} failed as it did not include any agent names.", + } + ) + + return True, None + + def _process_speaker_selection_result(self, result, last_speaker: ConversableAgent, agents: Optional[List[Agent]]): + """Checks the result of the auto_select_speaker function, returning the + agent to speak. + + Used by auto_select_speaker and a_auto_select_speaker.""" + if len(result.chat_history) > 0: + + # Use the final message, which will have the selected agent or reason for failure + final_message = result.chat_history[-1]["content"] + + if "[AGENT SELECTED]" in final_message: + + # Have successfully selected an agent, return it + return self.agent_by_name(final_message.replace("[AGENT SELECTED]", "")) + + else: # "[AGENT SELECTION FAILED]" + + # Failed to select an agent, so we'll select the next agent in the list + next_agent = self.next_agent(last_speaker, agents) + + # No agent, return the failed reason + return next_agent + def _participant_roles(self, agents: List[Agent] = None) -> str: # Default to all agents registered if agents is None: diff --git a/test/agentchat/test_groupchat.py b/test/agentchat/test_groupchat.py index 8a4758d2d37b..a4689bd539f7 100755 --- a/test/agentchat/test_groupchat.py +++ b/test/agentchat/test_groupchat.py @@ -1196,28 +1196,46 @@ def test_role_for_select_speaker_messages(): agents=[agent1, agent2], messages=[{"role": "user", "content": "Let's have a chat!"}], max_round=3, + role_for_select_speaker_messages="system", ) - # Run the select agents function to get the select speaker messages - selected_agent, agents, messages = groupchat._prepare_and_select_agents(agent1) + # Replicate the _auto_select_speaker nested chat. + + # Agent for checking the response from the speaker_select_agent + checking_agent = autogen.ConversableAgent("checking_agent") + + # Agent for selecting a single agent name from the response + speaker_selection_agent = autogen.ConversableAgent( + "speaker_selection_agent", + llm_config=None, + human_input_mode="NEVER", # Suppresses some extra terminal outputs, outputs will be handled by select_speaker_auto_verbose + ) + + # The role_for_select_speaker_message is put into the initiate_chat of the nested two-way chat + # into a message attribute called 'override_role'. This is evaluated in Conversable Agent's _append_oai_message function + # e.g.: message={'content':self.select_speaker_prompt(agents),'override_role':self.role_for_select_speaker_messages}, + message = {"content": "A prompt goes here.", "override_role": groupchat.role_for_select_speaker_messages} + checking_agent._append_oai_message(message, "assistant", speaker_selection_agent) # Test default is "system" - assert len(messages) == 2 - assert messages[-1]["role"] == "system" + assert len(checking_agent.chat_messages) == 1 + assert checking_agent.chat_messages[speaker_selection_agent][-1]["role"] == "system" # Test as "user" groupchat.role_for_select_speaker_messages = "user" - selected_agent, agents, messages = groupchat._prepare_and_select_agents(agent1) + message = {"content": "A prompt goes here.", "override_role": groupchat.role_for_select_speaker_messages} + checking_agent._append_oai_message(message, "assistant", speaker_selection_agent) - assert len(messages) == 2 - assert messages[-1]["role"] == "user" + assert len(checking_agent.chat_messages) == 1 + assert checking_agent.chat_messages[speaker_selection_agent][-1]["role"] == "user" # Test as something unusual groupchat.role_for_select_speaker_messages = "SockS" - selected_agent, agents, messages = groupchat._prepare_and_select_agents(agent1) + message = {"content": "A prompt goes here.", "override_role": groupchat.role_for_select_speaker_messages} + checking_agent._append_oai_message(message, "assistant", speaker_selection_agent) - assert len(messages) == 2 - assert messages[-1]["role"] == "SockS" + assert len(checking_agent.chat_messages) == 1 + assert checking_agent.chat_messages[speaker_selection_agent][-1]["role"] == "SockS" # Test empty string and None isn't accepted @@ -1307,7 +1325,7 @@ def test_select_speaker_message_and_prompt_templates(): speaker_selection_method="auto", max_round=10, select_speaker_message_template="Not empty.", - select_speaker_prompt_template=None, + select_speaker_prompt_template="", ) # Test with None @@ -1328,7 +1346,7 @@ def test_select_speaker_message_and_prompt_templates(): speaker_selection_method="auto", max_round=10, select_speaker_message_template="Not empty.", - select_speaker_prompt_template="", + select_speaker_prompt_template=None, ) @@ -1426,6 +1444,328 @@ def test_speaker_selection_agent_name_match(): assert result == {} +def test_speaker_selection_auto_process_result(): + """ + Tests the return result of the 2-agent chat used for speaker selection for the auto method. + The last message of the messages passed in will contain a pass or fail. + If passed, the message will contain the name of the correct agent and that agent will be returned. + If failed, the message will contain the reason for failure for the last attempt and the next + agent in the sequence will be returned. + """ + cmo = autogen.ConversableAgent( + name="Chief_Marketing_Officer", + human_input_mode="NEVER", + llm_config=False, + default_auto_reply="This is alice speaking.", + ) + pm = autogen.ConversableAgent( + name="Product_Manager", + human_input_mode="NEVER", + llm_config=False, + default_auto_reply="This is bob speaking.", + function_map={"test_func": lambda x: x}, + ) + + agent_list = [cmo, pm] + groupchat = autogen.GroupChat(agents=agent_list, messages=[], max_round=3) + + chat_result = autogen.ChatResult( + chat_id=None, + chat_history=[ + { + "content": "Let's get this meeting started. First the Product_Manager will create 3 new product ideas.", + "name": "Chairperson", + "role": "assistant", + }, + {"content": "You are an expert at finding the next speaker.", "role": "assistant"}, + {"content": "Product_Manager", "role": "user"}, + {"content": "UPDATED_BELOW", "role": "user"}, + ], + ) + + ### Agent selected successfully + chat_result.chat_history[3]["content"] = "[AGENT SELECTED]Product_Manager" + + # Product_Manager should be returned + assert groupchat._process_speaker_selection_result(chat_result, cmo, agent_list) == pm + + ### Agent not selected successfully + chat_result.chat_history[3][ + "content" + ] = "[AGENT SELECTION FAILED]Select speaker attempt #3 of 3 failed as it did not include any agent names." + + # The next speaker in the list will be selected, which will be the Product_Manager (as the last speaker is the Chief_Marketing_Officer) + assert groupchat._process_speaker_selection_result(chat_result, cmo, agent_list) == pm + + ### Invalid result messages, will return the next agent + chat_result.chat_history[3]["content"] = "This text should not be here." + + # The next speaker in the list will be selected, which will be the Chief_Marketing_Officer (as the last speaker is the Product_Maanger) + assert groupchat._process_speaker_selection_result(chat_result, pm, agent_list) == cmo + + +def test_speaker_selection_validate_speaker_name(): + """ + Tests the speaker name validation function used to evaluate the return result of the LLM + during speaker selection in 'auto' mode. + + Function: _validate_speaker_name + + If a single agent name is returned by the LLM, it will add a relevant message to the chat messages and return True, None + If multiple agent names are returned and there are attempts left, it will return a message to be used to prompt the LLM to try again + If multiple agent names are return and there are no attempts left, it will add a relevant message to the chat messages and return True, None + If no agent names are returned and there are attempts left, it will return a message to be used to prompt the LLM to try again + If no agent names are returned and there are no attempts left, it will add a relevant message to the chat messages and return True, None + + When returning a message, it will include the 'override_role' key and value to support the GroupChat role_for_select_speaker_messages attribute + """ + + # Group Chat setup + cmo = autogen.ConversableAgent( + name="Chief_Marketing_Officer", + human_input_mode="NEVER", + llm_config=False, + default_auto_reply="This is alice speaking.", + ) + pm = autogen.ConversableAgent( + name="Product_Manager", + human_input_mode="NEVER", + llm_config=False, + default_auto_reply="This is bob speaking.", + function_map={"test_func": lambda x: x}, + ) + + agent_list = [cmo, pm] + agent_list_string = f"{[agent.name for agent in agent_list]}" + groupchat = autogen.GroupChat(agents=agent_list, messages=[], max_round=3) + + # Speaker Selection 2-agent chat setup + + # Agent for selecting a single agent name from the response + speaker_selection_agent = autogen.ConversableAgent( + "speaker_selection_agent", + ) + + # Agent for checking the response from the speaker_select_agent + checking_agent = autogen.ConversableAgent("checking_agent") + + # Select speaker messages + select_speaker_messages = [ + { + "content": "Let's get this meeting started. First the Product_Manager will create 3 new product ideas.", + "name": "Chairperson", + "role": "assistant", + }, + {"content": "You are an expert at finding the next speaker.", "role": "assistant"}, + {"content": "UPDATED_BELOW", "role": "user"}, + ] + + ### Single agent name returned + attempts_left = 2 + attempt = 1 + select_speaker_messages[-1]["content"] = "Product_Manager is the next to speak" + + result = groupchat._validate_speaker_name( + recipient=checking_agent, + messages=select_speaker_messages, + sender=speaker_selection_agent, + config=None, + attempts_left=attempts_left, + attempt=attempt, + agents=agent_list, + ) + + assert result == (True, None) + assert select_speaker_messages[-1]["content"] == "[AGENT SELECTED]Product_Manager" + + select_speaker_messages.pop(-1) # Remove the last message before the next test + + ### Multiple agent names returned with attempts left + attempts_left = 2 + attempt = 1 + select_speaker_messages[-1]["content"] = "Product_Manager must speak after the Chief_Marketing_Officer" + + result = groupchat._validate_speaker_name( + recipient=checking_agent, + messages=select_speaker_messages, + sender=speaker_selection_agent, + config=None, + attempts_left=attempts_left, + attempt=attempt, + agents=agent_list, + ) + + assert result == ( + True, + { + "content": groupchat.select_speaker_auto_multiple_template.format(agentlist=agent_list_string), + "override_role": groupchat.role_for_select_speaker_messages, + }, + ) + + ### Multiple agent names returned with no attempts left + attempts_left = 0 + attempt = 1 + select_speaker_messages[-1]["content"] = "Product_Manager must speak after the Chief_Marketing_Officer" + + result = groupchat._validate_speaker_name( + recipient=checking_agent, + messages=select_speaker_messages, + sender=speaker_selection_agent, + config=None, + attempts_left=attempts_left, + attempt=attempt, + agents=agent_list, + ) + + assert result == (True, None) + assert ( + select_speaker_messages[-1]["content"] + == f"[AGENT SELECTION FAILED]Select speaker attempt #{attempt} of {attempt + attempts_left} failed as it returned multiple names." + ) + + select_speaker_messages.pop(-1) # Remove the last message before the next test + + ### No agent names returned with attempts left + attempts_left = 3 + attempt = 2 + select_speaker_messages[-1]["content"] = "The PM must speak after the CMO" + + result = groupchat._validate_speaker_name( + recipient=checking_agent, + messages=select_speaker_messages, + sender=speaker_selection_agent, + config=None, + attempts_left=attempts_left, + attempt=attempt, + agents=agent_list, + ) + + assert result == ( + True, + { + "content": groupchat.select_speaker_auto_none_template.format(agentlist=agent_list_string), + "override_role": groupchat.role_for_select_speaker_messages, + }, + ) + + ### Multiple agents returned with no attempts left + attempts_left = 0 + attempt = 3 + select_speaker_messages[-1]["content"] = "The PM must speak after the CMO" + + result = groupchat._validate_speaker_name( + recipient=checking_agent, + messages=select_speaker_messages, + sender=speaker_selection_agent, + config=None, + attempts_left=attempts_left, + attempt=attempt, + agents=agent_list, + ) + + assert result == (True, None) + assert ( + select_speaker_messages[-1]["content"] + == f"[AGENT SELECTION FAILED]Select speaker attempt #{attempt} of {attempt + attempts_left} failed as it did not include any agent names." + ) + + +def test_select_speaker_auto_messages(): + """ + In this test, two agents are part of a group chat which has customized select speaker "auto" multiple and no-name prompt messages. Both valid and empty string values will be used. + The expected behaviour is that the customized speaker selection "auto" messages will override the default values or throw exceptions if empty. + """ + + agent1 = autogen.ConversableAgent( + "Alice", + description="A wonderful employee named Alice.", + human_input_mode="NEVER", + llm_config=False, + ) + agent2 = autogen.ConversableAgent( + "Bob", + description="An amazing employee named Bob.", + human_input_mode="NEVER", + llm_config=False, + ) + + # Customised message for select speaker auto method where multiple agent names are returned + custom_multiple_names_msg = "You mentioned multiple names but we need just one. Select the best one. A reminder that the options are {agentlist}." + + # Customised message for select speaker auto method where no agent names are returned + custom_no_names_msg = "You forgot to select a single names and we need one, and only one. Select the best one. A reminder that the options are {agentlist}." + + # Test empty is_termination_msg function + groupchat = autogen.GroupChat( + agents=[agent1, agent2], + messages=[], + speaker_selection_method="auto", + max_round=10, + select_speaker_auto_multiple_template=custom_multiple_names_msg, + select_speaker_auto_none_template=custom_no_names_msg, + ) + + # Test using the _validate_speaker_name function, checking for the correct string and agentlist to be included + agents = [agent1, agent2] + + messages = [{"content": "Alice and Bob should both speak.", "name": "speaker_selector", "role": "user"}] + assert groupchat._validate_speaker_name(None, messages, None, None, 1, 1, agents) == ( + True, + { + "content": custom_multiple_names_msg.replace("{agentlist}", "['Alice', 'Bob']"), + "override_role": groupchat.role_for_select_speaker_messages, + }, + ) + + messages = [{"content": "Fred should both speak.", "name": "speaker_selector", "role": "user"}] + assert groupchat._validate_speaker_name(None, messages, None, None, 1, 1, agents) == ( + True, + { + "content": custom_no_names_msg.replace("{agentlist}", "['Alice', 'Bob']"), + "override_role": groupchat.role_for_select_speaker_messages, + }, + ) + + # Test with empty strings + with pytest.raises(ValueError, match="select_speaker_auto_multiple_template cannot be empty or None."): + groupchat = autogen.GroupChat( + agents=[agent1, agent2], + messages=[], + speaker_selection_method="auto", + max_round=10, + select_speaker_auto_multiple_template="", + ) + + with pytest.raises(ValueError, match="select_speaker_auto_none_template cannot be empty or None."): + groupchat = autogen.GroupChat( + agents=[agent1, agent2], + messages=[], + speaker_selection_method="auto", + max_round=10, + select_speaker_auto_none_template="", + ) + + # Test with None + with pytest.raises(ValueError, match="select_speaker_auto_multiple_template cannot be empty or None."): + groupchat = autogen.GroupChat( + agents=[agent1, agent2], + messages=[], + speaker_selection_method="auto", + max_round=10, + select_speaker_auto_multiple_template=None, + ) + + with pytest.raises(ValueError, match="select_speaker_auto_none_template cannot be empty or None."): + groupchat = autogen.GroupChat( + agents=[agent1, agent2], + messages=[], + speaker_selection_method="auto", + max_round=10, + select_speaker_auto_none_template=None, + ) + + if __name__ == "__main__": # test_func_call_groupchat() # test_broadcast() @@ -1443,5 +1783,8 @@ def test_speaker_selection_agent_name_match(): # test_custom_speaker_selection_overrides_transition_graph() # test_role_for_select_speaker_messages() # test_select_speaker_message_and_prompt_templates() - test_speaker_selection_agent_name_match() + # test_speaker_selection_agent_name_match() + test_speaker_selection_auto_process_result() + test_speaker_selection_validate_speaker_name() + test_select_speaker_auto_messages() # pass