agentchat.groupchat

群组聊天​

@dataclass
class GroupChat()

（预览中）一个包含以下数据字段的群聊类：

• agents: 参与的代理列表。
• messages: 群聊中的消息列表。
• max_round: 最大轮数。
• admin_name: 如果有管理员代理，则为其名称。默认为"Admin"。KeyBoardInterrupt将使管理员代理接管。
• func_call_filter: 是否强制执行函数调用过滤器。默认是 True。 当设置为 True 并且消息是函数调用建议时， 下一位发言者将从在其function_map中包含相应函数名称的代理中选择。
• select_speaker_message_template: 自定义选择发言者的消息（用于“自动”选择发言者），该消息首先出现在消息上下文中，通常包括代理描述和代理列表。如果字符串包含“{roles}”，它将被替换为代理及其角色描述。如果字符串包含“{agentlist}”，它将被替换为方括号中以逗号分隔的代理名称列表。默认值为： "你正在玩一个角色扮演游戏。以下角色可用： {roles}。 阅读以下对话。 然后从{agentlist}中选择下一个要扮演的角色。只需返回角色。"
• select_speaker_prompt_template: 自定义选择发言者的提示（用于“自动”发言者选择），该提示最后出现在消息上下文中，通常包括代理列表和指导LLM选择下一个代理的指南。如果字符串中包含“{agentlist}”，它将被替换为方括号中的代理名称逗号分隔列表。默认值为： "阅读以上对话。然后从{agentlist}中选择下一个角色。只返回角色。" 如果不想使用此提示，请将其设置为None。如果设置为None，请确保选择发言者的指令在select_speaker_message_template字符串中。
• select_speaker_auto_multiple_template: 自定义在选择发言人时，如果响应中包含多个代理名称，用于后续提示的模板。该提示引导LLM仅返回一个代理名称。仅适用于"auto"发言人选择方法。如果字符串中包含"{agentlist}"，它将被替换为方括号中的逗号分隔的代理名称列表。默认值为： "你在文本中提供了多个名称，请仅返回下一个发言人的名称。确定发言人时请使用以下优先级规则：

1. 如果在上下文中说话者自称为“作为…”，请选择该说话者的名称
2. 如果它指的是“下一个”说话者的名字，选择那个名字
3. 否则，选择上下文中提供的第一个说话者的名字 名字区分大小写，不应缩写或更改。 仅用说话者的名字回应，不要提供理由。

• select_speaker_auto_none_template: 自定义当选择发言人失败且响应中不包含任何代理名称时使用的后续提示。该提示指导LLM返回一个代理名称，并提供一个代理名称列表。仅适用于“auto”发言人选择方法。如果字符串中包含“{agentlist}”，它将被替换为包含在方括号中的逗号分隔的代理名称列表。默认值为： “您没有选择发言人。作为提醒，请使用这些优先规则来确定发言人：”

1. 如果在上下文中说话者自称为“作为…”，请选择该说话者的名称
2. 如果它指的是“下一个”说话者的名字，选择那个名字
3. 否则，选择上下文中提供的第一个发言者名称 名称区分大小写，不应缩写或更改。 唯一接受的名称是{agentlist}。 仅回答发言者名称，不要提供理由。

• speaker_selection_method: 用于选择下一个发言者的方法。默认为 "auto"。可以是以下任意一种（不区分大小写），如果未识别到，将引发 ValueError：
• "auto"：下一个发言者由LLM自动选择。
• "manual"：下一个发言者由用户手动选择。
• "random": 下一个发言者是随机选择的。
• "round_robin": 下一个发言者以循环方式选择，即按照 agents 中提供的相同顺序进行迭代。
• 一个定制的说话者选择函数（Callable）：该函数将被调用来选择下一个说话者。 函数应采用最后一个说话者和群聊作为输入，并返回以下之一：

1. 一个 Agent 类，它必须是群聊中的一个代理人。
2. 一个字符串从['auto', 'manual', 'random', 'round_robin']中选择要使用的默认方法。
3. 无，这将优雅地终止对话。

• max_retries_for_selecting_speaker: 选择发言者的最大重试次数。 如果在选择发言者时，LLM返回了多个代理名称或没有代理名称作为下一个代理，它将再次查询，直到返回单个代理或达到最大尝试次数。 仅适用于“自动”发言者选择方法。 默认值为2。
• select_speaker_transform_messages: （可选）应用于嵌套选择说话者代理间聊天消息的消息转换。 接受一个 TransformMessages 对象，默认为 None，仅在说话者选择方法为 "auto" 时使用。
• select_speaker_auto_verbose: 是否输出选择发言者的响应和选择 如果设置为True，将输出嵌套选择发言者聊天中两个代理的输出，以及 在选择代理时，响应是否成功 仅适用于"auto"发言者选择方法。
• allow_repeat_speaker: 是否允许同一个发言者连续发言。 默认为True，在这种情况下，所有发言者都可以连续发言。 如果allow_repeat_speaker是一个Agent列表，那么只有列出的Agent可以重复。 如果设置为False，则不允许任何发言者重复。 allow_repeat_speaker和allowed_or_disallowed_speaker_transitions是互斥的。
• allowed_or_disallowed_speaker_transitions: dict. 键是源代理，值是关键代理可以/不能转到的代理， 取决于speaker_transitions_type。默认值为None，这意味着所有代理都可以转到所有其他代理。 allow_repeat_speaker 和 allowed_or_disallowed_speaker_transitions 是互斥的。
• speaker_transitions_type: 确定speaker_transitions_type是一个包含允许代理列表还是禁止代理列表的字典。 "allowed"表示allowed_or_disallowed_speaker_transitions是一个包含允许代理列表的字典。 如果设置为"disallowed"，则allowed_or_disallowed_speaker_transitions是一个包含禁止代理列表的字典。 如果allowed_or_disallowed_speaker_transitions不为None，则必须提供此参数。
• enable_clear_history: 通过用户提示中提供“清除历史”短语，启用手动清除代理消息历史的功能。这是一个实验性功能。 有关更多信息，请参阅 GroupChatManager.clear_agents_history 函数的描述。
• send_introductions: 在群聊开始时发送一轮介绍，以便代理知道他们可以与谁交谈（默认值：False）
• select_speaker_auto_model_client_cls: 用于在“auto”说话者选择期间使用的内部说话者选择代理的自定义模型客户端类（可选）
• select_speaker_auto_llm_config: 用于在‘auto’发言人选择期间使用的内部发言人选择代理的LLM配置（可选）
• role_for_select_speaker_messages: 设置在“自动”模式下选择发言者时的角色名称，通常为“用户”或“系统”。(默认: '系统') python def custom_speaker_selection_func( last_speaker: Agent, groupchat: GroupChat ) -> Union[Agent, str, None]:

代理名称​

@property
def agent_names() -> List[str]

返回群聊中的代理名称。

重置​

def reset()

重置群聊。

append​

def append(message: Dict, speaker: Agent)

向群聊添加一条消息。 我们在这里将内容转换为字符串，以便它可以由基于文本的模型管理。

agent_by_name​

def agent_by_name(name: str,
                  recursive: bool = False,
                  raise_on_name_conflict: bool = False) -> Optional[Agent]

返回具有给定名称的代理。如果recursive为True，它将在嵌套团队中搜索。

嵌套代理​

def nested_agents() -> List[Agent]

返回群聊管理器中的所有代理。

下一个代理​

def next_agent(agent: Agent, agents: Optional[List[Agent]] = None) -> Agent

返回列表中的下一个代理。

选择发言者消息​

def select_speaker_msg(agents: Optional[List[Agent]] = None) -> str

返回选择下一位发言者的系统消息。这始终是上下文中的第一条消息。

选择发言者提示​

def select_speaker_prompt(agents: Optional[List[Agent]] = None) -> str

返回选择下一个发言者的浮动系统提示。 这始终是上下文中的最后一条消息。 如果select_speaker_prompt_template为None，将返回None。

介绍信息​

def introductions_msg(agents: Optional[List[Agent]] = None) -> str

返回选择下一位发言者的系统消息。这始终是上下文中的第一条消息。

手动选择发言人​

def manual_select_speaker(
        agents: Optional[List[Agent]] = None) -> Union[Agent, None]

手动选择下一个发言者。

随机选择演讲者​

def random_select_speaker(
        agents: Optional[List[Agent]] = None) -> Union[Agent, None]

随机选择下一个发言者。

选择发言者​

def select_speaker(last_speaker: Agent, selector: ConversableAgent) -> Agent

选择下一个发言人（带重新查询）。

a_select_speaker​

async def a_select_speaker(last_speaker: Agent,
                           selector: ConversableAgent) -> Agent

选择下一个发言者（带重新查询），异步进行。

a_auto_select_speaker​

async def a_auto_select_speaker(last_speaker: Agent,
                                selector: ConversableAgent,
                                messages: Optional[List[Dict]],
                                agents: Optional[List[Agent]]) -> Agent

（异步）为“auto”发言者选择方法选择下一个发言者。利用其自己的双代理聊天来确定下一个发言者，并支持重新查询。

“自动”演讲者选择方法的演讲者选择：

1. 创建一个包含发言者选择代理和发言者验证代理的双代理聊天，类似于嵌套聊天
2. 将群组消息注入到新聊天中
3. 运行双代理聊天，评估来自说话者选择代理的响应结果：

• 如果提供了单个代理，则返回它并完成。如果没有，我们在此嵌套聊天中添加一条额外消息，尝试引导LLM生成单个代理响应

4. 聊天继续进行，直到指定了一个代理或没有更多尝试机会
5. 如果我们用完了轮次，并且没有确定一个单独的代理，则返回代理列表中的下一个发言者

参数:

last_speaker Agent: 群聊中的上一个发言者 selector ConversableAgent: messages Optional[List[Dict]]: 当前的聊天消息 agents Optional[List[Agent]]: 用于选择发言者的有效代理列表

返回:

• Dict - 一个用于记录提及代理的计数器。

GroupChatManager​

class GroupChatManager(ConversableAgent)

（预览中）一个聊天管理器代理，可以管理多个代理的群聊。

群聊​

@property
def groupchat() -> GroupChat

返回由群聊管理器管理的群聊。

chat_messages_for_summary​

def chat_messages_for_summary(agent: Agent) -> List[Dict]

群聊中的消息列表作为摘要的对话。该代理被忽略。

最后的发言者​

@property
def last_speaker() -> Agent

返回向群聊管理器发送最后一条消息的代理。

在群聊中，代理总是会向群聊管理器发送消息，然后群聊管理器会将消息发送给群聊中的所有其他代理。因此，当代理接收到消息时，消息总是来自群聊管理器。利用这一特性，接收消息的代理可以知道实际发送消息的人是谁。

示例:

from autogen import ConversableAgent
from autogen import GroupChat, GroupChatManager


def print_messages(recipient, messages, sender, config):
    # Print the message immediately
    print(
        f"Sender: {sender.name} | Recipient: {recipient.name} | Message: {messages[-1].get('content')}"
    )
    print(f"Real Sender: {sender.last_speaker.name}")
    assert sender.last_speaker.name in messages[-1].get("content")
    return False, None  # Required to ensure the agent communication flow continues


agent_a = ConversableAgent("agent A", default_auto_reply="I'm agent A.")
agent_b = ConversableAgent("agent B", default_auto_reply="I'm agent B.")
agent_c = ConversableAgent("agent C", default_auto_reply="I'm agent C.")
for agent in [agent_a, agent_b, agent_c]:
    agent.register_reply(
        [ConversableAgent, None], reply_func=print_messages, config=None
    )
group_chat = GroupChat(
    [agent_a, agent_b, agent_c],
    messages=[],
    max_round=6,
    speaker_selection_method="random",
    allow_repeat_speaker=True,
)
chat_manager = GroupChatManager(group_chat)
groupchat_result = agent_a.initiate_chat(
    chat_manager, message="Hi, there, I'm agent A."
)

run_chat​

def run_chat(messages: Optional[List[Dict]] = None,
             sender: Optional[Agent] = None,
             config: Optional[GroupChat] = None) -> Tuple[bool, Optional[str]]

运行一个群聊。

a_run_chat​

async def a_run_chat(messages: Optional[List[Dict]] = None,
                     sender: Optional[Agent] = None,
                     config: Optional[GroupChat] = None)

异步运行群组聊天。

继续​

def resume(messages: Union[List[Dict], str],
           remove_termination_string: Union[str, Callable[[str], str]] = None,
           silent: Optional[bool] = False) -> Tuple[ConversableAgent, Dict]

从之前的消息作为起点恢复群聊。需要按照原始群聊的方式建立代理、群聊和群聊管理器。

参数:

• messages Union[List[Dict], str]: 之前聊天消息的内容，可以是Json字符串或消息字典的列表。
• remove_termination_string (str or function): 从最后一条消息中移除终止字符串，以防止立即终止 如果提供了一个字符串，则此字符串将从最后一条消息中移除。 如果提供了一个函数，则最后一条消息将传递给此函数。
• silent (bool or None): (实验性) 是否打印此对话的消息。默认为 False。

返回:

• Tuple[ConversableAgent, Dict]: 一个包含最后一个发言的代理及其消息的元组

a_resume​

async def a_resume(
        messages: Union[List[Dict], str],
        remove_termination_string: Union[str, Callable[[str], str]] = None,
        silent: Optional[bool] = False) -> Tuple[ConversableAgent, Dict]

使用之前的消息作为起点，异步恢复群聊。需要按照原始群聊的方式建立代理、群聊和群聊管理器。

参数:

• messages Union[List[Dict], str]: 之前聊天消息的内容，可以是Json字符串或消息字典的列表。
• remove_termination_string (str 或 function): 从最后一条消息中移除终止字符串，以防止立即终止 如果提供了字符串，则该字符串将从最后一条消息中移除。 如果提供了函数，则最后一条消息将传递给该函数，函数在处理后返回字符串。
• silent (bool or None): (实验性) 是否打印此对话的消息。默认为 False。

返回:

• Tuple[ConversableAgent, Dict]: 一个包含最后一个发言的代理及其消息的元组

messages_from_string​

def messages_from_string(message_string: str) -> List[Dict]

读取以 Json 格式保存的消息状态以进行恢复，并将其作为消息列表返回

参数:

• message_string: Json字符串，保存的状态

返回:

• List[Dict]: 消息列表

messages_to_string​

def messages_to_string(messages: List[Dict]) -> str

将提供的消息转换为可用于恢复聊天的Json字符串。状态由一系列消息组成。

参数:

• messages (List[Dict]): 要转换为字符串的消息集合

返回:

• str: 消息的JSON表示，可以持久化以便稍后恢复

清除代理历史记录​

def clear_agents_history(reply: dict, groupchat: GroupChat) -> str

清除所有代理或选定代理的消息历史。可以保留一定数量的最近消息。 当用户在回复中手动提供“清除历史记录”短语时，将调用该功能。 当提供“清除历史记录”时，所有代理的消息历史记录将被清除。 当提供“清除历史记录 <代理名称>”时，所选代理的消息历史记录将被清除。 当提供“清除历史记录 <要保留的消息数量>”时，所有代理的消息历史记录将被清除，除了最后 <要保留的消息数量> 条消息。 当提供“清除历史记录 <代理名称> <要保留的消息数量>”时，所选代理的消息历史记录将被清除，除了最后 <要保留的消息数量> 条消息。 在将回复传递到聊天之前，“清除历史记录”短语和可选参数将从回复中删除。

参数:

• reply dict - 要分析的回复消息字典。
• groupchat GroupChat - GroupChat 对象。