Selector Group Chat#

选择器群聊#

SelectorGroupChat 实现了一种团队协调模式，参与者轮流发布消息，下一个发言者由生成式模型（LLM）根据对话上下文选择。这实现了动态的、上下文感知的多代理对话。

SelectorGroupChat 提供了几个关键特性：

使用 LLM 分析对话上下文来动态选择发言者
可配置的参与者角色和描述
可选的防止同一发言者连续发言
可自定义的选择提示

发言者选择过程#

聊天使用 LLM 通过以下方式选择下一个发言者：

分析对话历史
评估参与者角色和描述
使用可配置的提示模板进行选择
验证是否恰好选择了一个参与者

import asyncio
from typing import List, Sequence

from autogen_agentchat.agents import (
    BaseChatAgent,
    CodingAssistantAgent,
    ToolUseAssistantAgent,
)
from autogen_agentchat.base import Response
from autogen_agentchat.messages import ChatMessage, StopMessage, TextMessage
from autogen_agentchat.task import TextMentionTermination
from autogen_agentchat.teams import SelectorGroupChat
from autogen_core.base import CancellationToken
from autogen_core.components.tools import FunctionTool
from autogen_ext.models import OpenAIChatCompletionClient

定义代理#

UserProxyAgent 允许用户直接输入消息。该代理等待用户输入，并返回文本消息，如果用户决定终止对话，则返回停止消息。

class UserProxyAgent(BaseChatAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name, "A human user.")

    @property
    def produced_message_types(self) -> List[type[ChatMessage]]:
        return [TextMessage, StopMessage]

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        user_input = await asyncio.get_event_loop().run_in_executor(None, input, "Enter your response: ")
        if "TERMINATE" in user_input:
            return Response(chat_message=StopMessage(content="User has terminated the conversation.", source=self.name))
        return Response(chat_message=TextMessage(content=user_input, source=self.name))

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass

async def flight_search(start: str, destination: str, date: str) -> str:
    return "\n".join(
        [
            f"AC24 from {start} to {destination} on {date} is $500",
            f"UA23 from {start} to {destination} on {date} is $450",
            f"AL21 from {start} to {destination} on {date} is $400",
        ]
    )


async def flight_booking(flight: str, date: str) -> str:
    return f"Booked flight {flight} on {date}"

ToolUseAssistantAgent 负责调用外部工具。在这个例子中，定义了两个工具：flight_search 和 flight_booking。

此外，CodingAssistantAgent 作为一个通用的旅行助手，其行为在 system_message 中预先定义。

user_proxy = UserProxyAgent("User")
flight_broker = ToolUseAssistantAgent(
    "FlightBroker",
    description="An assistant for booking flights",
    model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"),
    registered_tools=[
        FunctionTool(flight_search, description="Search for flights"),
        FunctionTool(flight_booking, description="Book a flight"),
    ],
)
travel_assistant = CodingAssistantAgent(
    "TravelAssistant",
    description="A travel assistant",
    model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"),
    system_message="You are a travel assistant.",
)

termination = TextMentionTermination("TERMINATE")
team = SelectorGroupChat(
    [user_proxy, flight_broker, travel_assistant],
    model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"),
    termination_condition=termination,
)
await team.run(task="Help user plan a trip and book a flight.")