aphilippov/python-server-api
0
1import logging2import random3import re4import sys5from dataclasses import dataclass6from typing import Dict, List, Optional, Union, Tuple7 8from ..code_utils import content_str9from .agent import Agent10from .conversable_agent import ConversableAgent11 12logger = logging.getLogger(__name__)13 14 15@dataclass16class GroupChat:17 """(In preview) A group chat class that contains the following data fields:18 - agents: a list of participating agents.19 - messages: a list of messages in the group chat.20 - max_round: the maximum number of rounds.21 - admin_name: the name of the admin agent if there is one. Default is "Admin".22 KeyBoardInterrupt will make the admin agent take over.23 - func_call_filter: whether to enforce function call filter. Default is True.24 When set to True and when a message is a function call suggestion,25 the next speaker will be chosen from an agent which contains the corresponding function name26 in its `function_map`.27 - speaker_selection_method: the method for selecting the next speaker. Default is "auto".28 Could be any of the following (case insensitive), will raise ValueError if not recognized:29 - "auto": the next speaker is selected automatically by LLM.30 - "manual": the next speaker is selected manually by user input.31 - "random": the next speaker is selected randomly.32 - "round_robin": the next speaker is selected in a round robin fashion, i.e., iterating in the same order as provided in `agents`.33 - 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. If set to False, then no speakers are allowed to repeat.34 """35 36 agents: List[Agent]37 messages: List[Dict]38 max_round: Optional[int] = 1039 admin_name: Optional[str] = "Admin"40 func_call_filter: Optional[bool] = True41 speaker_selection_method: Optional[str] = "auto"42 allow_repeat_speaker: Optional[Union[bool, List[Agent]]] = True43 44 _VALID_SPEAKER_SELECTION_METHODS = ["auto", "manual", "random", "round_robin"]45 46 @property47 def agent_names(self) -> List[str]:48 """Return the names of the agents in the group chat."""49 return [agent.name for agent in self.agents]50 51 def reset(self):52 """Reset the group chat."""53 self.messages.clear()54 55 def append(self, message: Dict):56 """Append a message to the group chat.57 We cast the content to str here so that it can be managed by text-based58 model.59 """60 message["content"] = content_str(message["content"])61 self.messages.append(message)62 63 def agent_by_name(self, name: str) -> Agent:64 """Returns the agent with a given name."""65 return self.agents[self.agent_names.index(name)]66 67 def next_agent(self, agent: Agent, agents: Optional[List[Agent]] = None) -> Agent:68 """Return the next agent in the list."""69 if agents is None:70 agents = self.agents71 72 # What index is the agent? (-1 if not present)73 idx = self.agent_names.index(agent.name) if agent.name in self.agent_names else -174 75 # Return the next agent76 if agents == self.agents:77 return agents[(idx + 1) % len(agents)]78 else:79 offset = idx + 180 for i in range(len(self.agents)):81 if self.agents[(offset + i) % len(self.agents)] in agents:82 return self.agents[(offset + i) % len(self.agents)]83 84 def select_speaker_msg(self, agents: Optional[List[Agent]] = None) -> str:85 """Return the system message for selecting the next speaker. This is always the *first* message in the context."""86 if agents is None:87 agents = self.agents88 return f"""You are in a role play game. The following roles are available:89{self._participant_roles(agents)}.90 91Read the following conversation.92Then select the next role from {[agent.name for agent in agents]} to play. Only return the role."""93 94 def select_speaker_prompt(self, agents: Optional[List[Agent]] = None) -> str:95 """Return the floating system prompt selecting the next speaker. This is always the *last* message in the context."""96 if agents is None:97 agents = self.agents98 return f"Read the above conversation. Then select the next role from {[agent.name for agent in agents]} to play. Only return the role."99 100 def manual_select_speaker(self, agents: Optional[List[Agent]] = None) -> Union[Agent, None]:101 """Manually select the next speaker."""102 if agents is None:103 agents = self.agents104 105 print("Please select the next speaker from the following list:")106 _n_agents = len(agents)107 for i in range(_n_agents):108 print(f"{i+1}: {agents[i].name}")109 try_count = 0110 # Assume the user will enter a valid number within 3 tries, otherwise use auto selection to avoid blocking.111 while try_count <= 3:112 try_count += 1113 if try_count >= 3:114 print(f"You have tried {try_count} times. The next speaker will be selected automatically.")115 break116 try:117 i = input("Enter the number of the next speaker (enter nothing or `q` to use auto selection): ")118 if i == "" or i == "q":119 break120 i = int(i)121 if i > 0 and i <= _n_agents:122 return agents[i - 1]123 else:124 raise ValueError125 except ValueError:126 print(f"Invalid input. Please enter a number between 1 and {_n_agents}.")127 return None128 129 def _prepare_and_select_agents(self, last_speaker: Agent) -> Tuple[Optional[Agent], List[Agent]]:130 if self.speaker_selection_method.lower() not in self._VALID_SPEAKER_SELECTION_METHODS:131 raise ValueError(132 f"GroupChat speaker_selection_method is set to '{self.speaker_selection_method}'. "133 f"It should be one of {self._VALID_SPEAKER_SELECTION_METHODS} (case insensitive). "134 )135 136 # If provided a list, make sure the agent is in the list137 allow_repeat_speaker = (138 self.allow_repeat_speaker139 if isinstance(self.allow_repeat_speaker, bool)140 else last_speaker in self.allow_repeat_speaker141 )142 143 agents = self.agents144 n_agents = len(agents)145 # Warn if GroupChat is underpopulated146 if n_agents < 2:147 raise ValueError(148 f"GroupChat is underpopulated with {n_agents} agents. "149 "Please add more agents to the GroupChat or use direct communication instead."150 )151 elif n_agents == 2 and self.speaker_selection_method.lower() != "round_robin" and allow_repeat_speaker:152 logger.warning(153 f"GroupChat is underpopulated with {n_agents} agents. "154 "It is recommended to set speaker_selection_method to 'round_robin' or allow_repeat_speaker to False."155 "Or, use direct communication instead."156 )157 158 if (159 self.func_call_filter160 and self.messages161 and ("function_call" in self.messages[-1] or "tool_calls" in self.messages[-1])162 ):163 funcs = []164 if "function_call" in self.messages[-1]:165 funcs += [self.messages[-1]["function_call"]["name"]]166 if "tool_calls" in self.messages[-1]:167 funcs += [168 tool["function"]["name"] for tool in self.messages[-1]["tool_calls"] if tool["type"] == "function"169 ]170 171 # find agents with the right function_map which contains the function name172 agents = [agent for agent in self.agents if agent.can_execute_function(funcs)]173 if len(agents) == 1:174 # only one agent can execute the function175 return agents[0], agents176 elif not agents:177 # find all the agents with function_map178 agents = [agent for agent in self.agents if agent.function_map]179 if len(agents) == 1:180 return agents[0], agents181 elif not agents:182 raise ValueError(183 f"No agent can execute the function {', '.join(funcs)}. "184 "Please check the function_map of the agents."185 )186 # remove the last speaker from the list to avoid selecting the same speaker if allow_repeat_speaker is False187 agents = agents if allow_repeat_speaker else [agent for agent in agents if agent != last_speaker]188 189 if self.speaker_selection_method.lower() == "manual":190 selected_agent = self.manual_select_speaker(agents)191 elif self.speaker_selection_method.lower() == "round_robin":192 selected_agent = self.next_agent(last_speaker, agents)193 elif self.speaker_selection_method.lower() == "random":194 selected_agent = random.choice(agents)195 else:196 selected_agent = None197 return selected_agent, agents198 199 def select_speaker(self, last_speaker: Agent, selector: ConversableAgent):200 """Select the next speaker."""201 selected_agent, agents = self._prepare_and_select_agents(last_speaker)202 if selected_agent:203 return selected_agent204 # auto speaker selection205 selector.update_system_message(self.select_speaker_msg(agents))206 207 # If last message is a tool call or function call, blank the call so the api doesn't throw208 messages = self.messages.copy()209 if messages[-1].get("function_call", False):210 messages[-1] = dict(messages[-1], function_call=None)211 if messages[-1].get("tool_calls", False):212 messages[-1] = dict(messages[-1], tool_calls=None)213 context = messages + [{"role": "system", "content": self.select_speaker_prompt(agents)}]214 final, name = selector.generate_oai_reply(context)215 216 if not final:217 # the LLM client is None, thus no reply is generated. Use round robin instead.218 return self.next_agent(last_speaker, agents)219 220 # If exactly one agent is mentioned, use it. Otherwise, leave the OAI response unmodified221 mentions = self._mentioned_agents(name, agents)222 if len(mentions) == 1:223 name = next(iter(mentions))224 else:225 logger.warning(226 f"GroupChat select_speaker failed to resolve the next speaker's name. This is because the speaker selection OAI call returned:\n{name}"227 )228 229 # Return the result230 try:231 return self.agent_by_name(name)232 except ValueError:233 return self.next_agent(last_speaker, agents)234 235 async def a_select_speaker(self, last_speaker: Agent, selector: ConversableAgent):236 """Select the next speaker."""237 selected_agent, agents = self._prepare_and_select_agents(last_speaker)238 if selected_agent:239 return selected_agent240 # auto speaker selection241 selector.update_system_message(self.select_speaker_msg(agents))242 final, name = await selector.a_generate_oai_reply(243 self.messages244 + [245 {246 "role": "system",247 "content": f"Read the above conversation. Then select the next role from {[agent.name for agent in agents]} to play. Only return the role.",248 }249 ]250 )251 if not final:252 # the LLM client is None, thus no reply is generated. Use round robin instead.253 return self.next_agent(last_speaker, agents)254 255 # If exactly one agent is mentioned, use it. Otherwise, leave the OAI response unmodified256 mentions = self._mentioned_agents(name, agents)257 if len(mentions) == 1:258 name = next(iter(mentions))259 else:260 logger.warning(261 f"GroupChat select_speaker failed to resolve the next speaker's name. This is because the speaker selection OAI call returned:\n{name}"262 )263 264 # Return the result265 try:266 return self.agent_by_name(name)267 except ValueError:268 return self.next_agent(last_speaker, agents)269 270 def _participant_roles(self, agents: List[Agent] = None) -> str:271 # Default to all agents registered272 if agents is None:273 agents = self.agents274 275 roles = []276 for agent in agents:277 if agent.description.strip() == "":278 logger.warning(279 f"The agent '{agent.name}' has an empty description, and may not work well with GroupChat."280 )281 roles.append(f"{agent.name}: {agent.description}".strip())282 return "\n".join(roles)283 284 def _mentioned_agents(self, message_content: Union[str, List], agents: List[Agent]) -> Dict:285 """Counts the number of times each agent is mentioned in the provided message content.286 287 Args:288 message_content (Union[str, List]): The content of the message, either as a single string or a list of strings.289 agents (List[Agent]): A list of Agent objects, each having a 'name' attribute to be searched in the message content.290 291 Returns:292 Dict: a counter for mentioned agents.293 """294 # Cast message content to str295 if isinstance(message_content, dict):296 message_content = message_content["content"]297 message_content = content_str(message_content)298 299 mentions = dict()300 for agent in agents:301 regex = (302 r"(?<=\W)" + re.escape(agent.name) + r"(?=\W)"303 ) # Finds agent mentions, taking word boundaries into account304 count = len(re.findall(regex, f" {message_content} ")) # Pad the message to help with matching305 if count > 0:306 mentions[agent.name] = count307 return mentions308 309 310class GroupChatManager(ConversableAgent):311 """(In preview) A chat manager agent that can manage a group chat of multiple agents."""312 313 def __init__(314 self,315 groupchat: GroupChat,316 name: Optional[str] = "chat_manager",317 # unlimited consecutive auto reply by default318 max_consecutive_auto_reply: Optional[int] = sys.maxsize,319 human_input_mode: Optional[str] = "NEVER",320 system_message: Optional[Union[str, List]] = "Group chat manager.",321 **kwargs,322 ):323 if kwargs.get("llm_config") and (kwargs["llm_config"].get("functions") or kwargs["llm_config"].get("tools")):324 raise ValueError(325 "GroupChatManager is not allowed to make function/tool calls. Please remove the 'functions' or 'tools' config in 'llm_config' you passed in."326 )327 328 super().__init__(329 name=name,330 max_consecutive_auto_reply=max_consecutive_auto_reply,331 human_input_mode=human_input_mode,332 system_message=system_message,333 **kwargs,334 )335 # Order of register_reply is important.336 # Allow sync chat if initiated using initiate_chat337 self.register_reply(Agent, GroupChatManager.run_chat, config=groupchat, reset_config=GroupChat.reset)338 # Allow async chat if initiated using a_initiate_chat339 self.register_reply(Agent, GroupChatManager.a_run_chat, config=groupchat, reset_config=GroupChat.reset)340 341 def run_chat(342 self,343 messages: Optional[List[Dict]] = None,344 sender: Optional[Agent] = None,345 config: Optional[GroupChat] = None,346 ) -> Union[str, Dict, None]:347 """Run a group chat."""348 if messages is None:349 messages = self._oai_messages[sender]350 message = messages[-1]351 speaker = sender352 groupchat = config353 for i in range(groupchat.max_round):354 # set the name to speaker's name if the role is not function355 if message["role"] != "function":356 message["name"] = speaker.name357 358 groupchat.append(message)359 360 if self._is_termination_msg(message):361 # The conversation is over362 break363 # broadcast the message to all agents except the speaker364 for agent in groupchat.agents:365 if agent != speaker:366 self.send(message, agent, request_reply=False, silent=True)367 if i == groupchat.max_round - 1:368 # the last round369 break370 try:371 # select the next speaker372 speaker = groupchat.select_speaker(speaker, self)373 # let the speaker speak374 reply = speaker.generate_reply(sender=self)375 except KeyboardInterrupt:376 # let the admin agent speak if interrupted377 if groupchat.admin_name in groupchat.agent_names:378 # admin agent is one of the participants379 speaker = groupchat.agent_by_name(groupchat.admin_name)380 reply = speaker.generate_reply(sender=self)381 else:382 # admin agent is not found in the participants383 raise384 if reply is None:385 break386 # The speaker sends the message without requesting a reply387 speaker.send(reply, self, request_reply=False)388 message = self.last_message(speaker)389 return True, None390 391 async def a_run_chat(392 self,393 messages: Optional[List[Dict]] = None,394 sender: Optional[Agent] = None,395 config: Optional[GroupChat] = None,396 ):397 """Run a group chat asynchronously."""398 if messages is None:399 messages = self._oai_messages[sender]400 message = messages[-1]401 speaker = sender402 groupchat = config403 for i in range(groupchat.max_round):404 # set the name to speaker's name if the role is not function405 if message["role"] != "function":406 message["name"] = speaker.name407 408 groupchat.append(message)409 410 if self._is_termination_msg(message):411 # The conversation is over412 break413 414 # broadcast the message to all agents except the speaker415 for agent in groupchat.agents:416 if agent != speaker:417 await self.a_send(message, agent, request_reply=False, silent=True)418 if i == groupchat.max_round - 1:419 # the last round420 break421 try:422 # select the next speaker423 speaker = await groupchat.a_select_speaker(speaker, self)424 # let the speaker speak425 reply = await speaker.a_generate_reply(sender=self)426 except KeyboardInterrupt:427 # let the admin agent speak if interrupted428 if groupchat.admin_name in groupchat.agent_names:429 # admin agent is one of the participants430 speaker = groupchat.agent_by_name(groupchat.admin_name)431 reply = await speaker.a_generate_reply(sender=self)432 else:433 # admin agent is not found in the participants434 raise435 if reply is None:436 break437 # The speaker sends the message without requesting a reply438 await speaker.a_send(reply, self, request_reply=False)439 message = self.last_message(speaker)440 return True, None441