wangtao
/
autogen
mirror of https://github.com/microsoft/autogen.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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 <wang.chi@microsoft.com>
This commit is contained in:
Mark Sze 2024-04-30 14:16:08 +10:00 committed by GitHub
parent 5e29ac84dc
commit 5b6ae324e2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 732 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
autogen/agentchat/conversable_agent.py
View File

@ -576,6 +576,11 @@ class ConversableAgent(LLMAgent):
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

390
autogen/agentchat/groupchat.py
View File

@ -7,6 +7,7 @@ from typing import Callable, Dict, List, Literal, Optional, Tuple, Union
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 @@ class GroupChat:
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 @@ class GroupChat:
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 @@ class GroupChat:
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 @@ class GroupChat:
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 @@ class GroupChat:
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 @@ class GroupChat:
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 @@ class GroupChat:
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:

369
test/agentchat/test_groupchat.py
View File

@ -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 @@ if __name__ == "__main__":
# 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