Migrating from pyautogen 0.3 to autogen v0.4.0: How to Log Group Chat Messages in correct order using the new SelectorGroupChat

[aayushmittal2]

I’ve been successfully using pyautogen v0.3.2 to orchestrate multi-agent workflows and capture real-time group chat messages for logging and notifications. My current setup uses a custom GroupChatLogger class that periodically checks new messages, does some domain-specific processing (like extracting search links, building Markdown from JSON, sending notifications, etc.), and caches these logs in Redis.

Below is a simplified version of my custom logger and how it’s used in pyautogen v0.3.2 . This below code snippit works well and tracks the execution.

class GroupChatLogger:
    def __init__(
        self,
        manager,
 ....
        agents_names
    ):
        self.manager = manager
        ....
        self.last_message_idx = 0
        self.last_message_time = None
        self.agents_names = agents_names
        self.mem = set()
        self.stop_event = asyncio.Event()

    async def start(self):
        """Starts the periodic polling of the group chat."""
        while not self.stop_event.is_set():
            await self.print_new_messages()
            await asyncio.sleep(5)

    async def print_new_messages(self):
        """Looks for new messages in the group chat and logs them."""
        new_messages = self.manager.groupchat.messages[self.last_message_idx:]
        current_time = time.time()
        
        for idx, message in enumerate(new_messages, start=self.last_message_idx):
            time_diff = current_time - self.last_message_time if self.last_message_time else 0
            
            # Example: we do some domain extraction from message content
            domains = None
            if message.get('name') == "scrap_for_online_search_terms":
                # e.g., parse JSON from Redis to extract domain icons, etc.
                pass

            # We build a data payload that includes time spent, token counts, etc.
            data = {
                "agent": message.get('name', ''),
                "message_role": message.get('role'),
                "time_spent": f"{time_diff:.2f}",
                "token_count": ... # hypothetical method to count tokens
                # ...
            }
            
            # We log the data to Redis and send push notifications
            # redis_helper.set(...)
            # NotificationHelper().push_notification(...)

            self.last_message_time = current_time

        self.last_message_idx = len(self.manager.groupchat.messages)
        self.last_message_time = current_time

    async def stop(self):
        self.stop_event.set(
        
 async def get_autogen_fsm_response(
    initializer,
    graph_dict,
    agents,
    allow_repeat_speakers,
    prompt,
    ...
    agents_names={},
    final_agents=None
):
    # 1) Create the GroupChat
    group_chat = GroupChat(agents=agents, messages=[], ...)

    # 2) Create the Manager
    manager = GroupChatManager(groupchat=group_chat, ...)

    # 3) Initialize & start the logger in background
    logger_timer = GroupChatLogger(
        manager, 
.... , agents_names
    )
    asyncio.create_task(logger_timer.start())

    try:
        # 4) Perform the conversation
        chat_result = await asyncio.create_task(run_initiate_chat(initializer, manager, prompt))
        last_response = manager.groupchat.messages[-1].get('content')
    finally:
        # 5) Stop the logger
        await logger_timer.stop()

    return {
        "response": last_response,
        "summary": chat_result.summary,
    }

The New autogen v0.4.0 Approach

Now, with autogen v0.4.0, I’m using SelectorGroupChat and passing in a model_client, plus new termination conditions, etc. The SelectorGroupChat works well in itself but I am not able to figure how to setup the logger functionality to track the execution in correct flow and use the agent communcation message for notification.

Here’s an abbreviated snippet of how I’m creating and running the group chat:

async def get_autogen_fsm_response(
    prompt: str,
    initializer: UserProxyAgent,
    agent_details: List[UserProxyAgent | AssistantAgent],
    ...
    agents_names: dict = {},
    **kwargs
) -> Dict[str, Any]:
    try:
        model_client = AzureOpenAIChatCompletionClient(...)
        
        # Filter out duplicates
        seen_names = set()
        unique_agents = []
        for agent in agent_details:
            if agent.name not in seen_names and agent.name != "User":
                seen_names.add(agent.name)
                unique_agents.append(agent)

        # Termination conditions
        text_mention_termination = TextMentionTermination("TERMINATE")
        max_messages_termination = MaxMessageTermination(max_messages=15)
        termination = text_mention_termination | max_messages_termination

        # Ensure each agent has a unique name
        for i, agent in enumerate(unique_agents):
            if not agent.name:
                agent.name = f"Agent_{i}"

        # Build the group chat
        group_chat = SelectorGroupChat(
            unique_agents,
            model_client=model_client,
            termination_condition=termination
        )

        # New GroupChatLogger usage in v0.4
        logger = GroupChatLogger(
            group_chat,
            bot_id,
            conversation_id,
            message_id,
            agents_icons,
            agents_names,
        )
        logger_task = asyncio.create_task(logger.start())

        # Actually run the chat
        print(f"Running prompt: {prompt}")
        result = await Console(group_chat.run_stream(task=prompt))

        # Wait for logger
        await logger.stop()
        await logger_task

        last_message = result.messages[-1] if result.messages else None
        final_response = last_message.content if isinstance(last_message, TextMessage) else ""

        return {
            "response": final_response,
            "summary": last_message,
        }

    except Exception as e:
        raise e

How can I hook into each new message in SelectorGroupChat so I can retrieve them in real-time the way I did with manager.groupchat.messages in pyautogen?

Any examples or hints on how others are doing real-time logging within the new autogen v0.4.0 SelectorGroupChat would be super helpful!

[ekzhu]

@aayushmittal2, for SelectorGroupChat, you can use run_stream to get the messages in the order they are produced. In fact you are already using it. Instead of wrapping it around with Console, just use

from autogen_agentchat.base import TaskResult
task_result = None
async for message in group_chat.run_stream(task=prompt):
    print(message) # Or log it.
    if isinstance(message, TaskResult):
        task_result = message
# Use the task_result as before.

[ekzhu]

Read the doc for more info: https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/tutorial/teams.html#observing-a-team