GroupChat

def reset(self):
    """Reset the group chat."""
    self.messages.clear()

def append(self, message: dict[str, Any], speaker: Agent):
    """Append a message to the group chat.
    We cast the content to str here so that it can be managed by text-based
    model.
    """
    # set the name to speaker's name if the role is not function
    # if the role is tool, it is OK to modify the name
    if message["role"] != "function":
        message["name"] = speaker.name
    if not isinstance(message["content"], str) and not isinstance(message["content"], list):
        message["content"] = str(message["content"])
    message["content"] = content_str(message["content"])
    self.messages.append(message)

def agent_by_name(
    self, name: str, recursive: bool = False, raise_on_name_conflict: bool = False
) -> Optional[Agent]:
    """Returns the agent with a given name. If recursive is True, it will search in nested teams."""
    agents = self.nested_agents() if recursive else self.agents
    filtered_agents = [agent for agent in agents if agent.name == name]

    if raise_on_name_conflict and len(filtered_agents) > 1:
        raise AgentNameConflictError()

    return filtered_agents[0] if filtered_agents else None

def nested_agents(self) -> list[Agent]:
    """Returns all agents in the group chat manager."""
    agents = self.agents.copy()
    for agent in agents:
        if isinstance(agent, GroupChatManager):
            # Recursive call for nested teams
            agents.extend(agent.groupchat.nested_agents())
    return agents

def next_agent(self, agent: Agent, agents: Optional[list[Agent]] = None) -> Agent:
    """Return the next agent in the list."""
    if agents is None:
        agents = self.agents

    # Ensure the provided list of agents is a subset of self.agents
    if not set(agents).issubset(set(self.agents)):
        raise UndefinedNextAgentError()

    # What index is the agent? (-1 if not present)
    idx = self.agent_names.index(agent.name) if agent.name in self.agent_names else -1

    # Return the next agent
    if agents == self.agents:
        return agents[(idx + 1) % len(agents)]
    else:
        offset = idx + 1
        for i in range(len(self.agents)):
            if self.agents[(offset + i) % len(self.agents)] in agents:
                return self.agents[(offset + i) % len(self.agents)]

    # Explicitly handle cases where no valid next agent exists in the provided subset.
    raise UndefinedNextAgentError()

def select_speaker_msg(self, agents: Optional[list[Agent]] = None) -> str:
    """Return the system message for selecting the next speaker. This is always the *first* message in the context."""
    if agents is None:
        agents = self.agents

    roles = self._participant_roles(agents)
    agentlist = f"{[agent.name for agent in agents]}"

    return_msg = self.select_speaker_message_template.format(roles=roles, agentlist=agentlist)
    return return_msg

def select_speaker_prompt(self, agents: Optional[list[Agent]] = None) -> str:
    """Return the floating system prompt selecting the next speaker.
    This is always the *last* message in the context.
    Will return None if the select_speaker_prompt_template is None.
    """
    if self.select_speaker_prompt_template is None:
        return None

    if agents is None:
        agents = self.agents

    agentlist = f"{[agent.name for agent in agents]}"

    return_prompt = f"{self.select_speaker_prompt_template}".replace("{agentlist}", agentlist)
    return return_prompt

def introductions_msg(self, agents: Optional[list[Agent]] = None) -> str:
    """Return the system message for selecting the next speaker. This is always the *first* message in the context."""
    if agents is None:
        agents = self.agents

    # Use the class attribute instead of a hardcoded string
    intro_msg = self.DEFAULT_INTRO_MSG
    participant_roles = self._participant_roles(agents)

    return f"{intro_msg}\n\n{participant_roles}"

def manual_select_speaker(self, agents: Optional[list[Agent]] = None) -> Union[Agent, None]:
    """Manually select the next speaker."""
    iostream = IOStream.get_default()

    if agents is None:
        agents = self.agents

    iostream.send(SelectSpeakerMessage(agents=agents))

    try_count = 0
    # Assume the user will enter a valid number within 3 tries, otherwise use auto selection to avoid blocking.
    while try_count <= 3:
        try_count += 1
        if try_count >= 3:
            iostream.send(SelectSpeakerTryCountExceededMessage(try_count=try_count, agents=agents))
            break
        try:
            i = iostream.input(
                "Enter the number of the next speaker (enter nothing or `q` to use auto selection): "
            )
            if i == "" or i == "q":
                break
            i = int(i)
            if i > 0 and i <= len(agents):
                return agents[i - 1]
            else:
                raise ValueError
        except ValueError:
            iostream.send(SelectSpeakerInvalidInputMessage(agents=agents))
    return None

def random_select_speaker(self, agents: Optional[list[Agent]] = None) -> Union[Agent, None]:
    """Randomly select the next speaker."""
    if agents is None:
        agents = self.agents
    return random.choice(agents)

def select_speaker(self, last_speaker: Agent, selector: ConversableAgent) -> Agent:
    """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
    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 (with requery), asynchronously."""
    selected_agent, agents, messages = self._prepare_and_select_agents(last_speaker)
    if selected_agent:
        return selected_agent
    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)

async def a_auto_select_speaker(
    self,
    last_speaker: Agent,
    selector: ConversableAgent,
    messages: Optional[list[dict[str, Any]]],
    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: The previous speaker in the group chat
        selector: The ConversableAgent that initiated the speaker selection
        messages: Current chat messages
        agents: Valid list of agents for speaker selection

    Returns:
        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, Optional[Union[str, dict[str, Any]]]]:
        # 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, speaker_selection_agent = self._create_internal_agents(
        agents, max_attempts, messages, validate_speaker_name, selector
    )

    # Create the starting message
    if self.select_speaker_prompt_template is not None:
        start_message = {
            "content": self.select_speaker_prompt(agents),
            "override_role": self.role_for_select_speaker_messages,
        }
    else:
        start_message = messages[-1]

    # Add the message transforms, if any, to the speaker selection agent
    if self._speaker_selection_transforms is not None:
        self._speaker_selection_transforms.add_to_agent(speaker_selection_agent)

    # 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=start_message,
        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)