Repository: microsoft/autogen
Pull request: feat: add support for list of messages as team task input
<!-- Thank you for your contribution! Please review https://microsoft.github.io/autogen/docs/Contribute before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

Currently, a team only supports a single message as input for a task. This limits the ability to prepopulate context for agents, especially when working with previous conversation histories or nested chats. This PR adds support for passing a list of messages as task input, enabling better integration with applications where users might type multiple messages before triggering the team run.

Key changes:
- Added support for `List[ChatMessage]` as valid task input
- Messages are processed sequentially, maintaining conversation flow
- Added validation for message lists (empty lists, invalid message types)
- Maintained backward compatibility with existing string and single message inputs
- Added comprehensive test coverage for the new functionality

## Related issue number

Resolves #4472

## Checks

- [x] I've included any doc changes needed for https://microsoft.github.io/autogen/. See https://microsoft.github.io/autogen/docs/Contribute#documentation to build and test documentation locally.
- [x] I've added tests (if relevant) corresponding to the changes introduced in this PR.
- [x] I've made sure all auto checks have passed.

Problem statement:
Team to support a list of messages as input to a task
Currently a team only support a single message for task.

```python
result = await team.run(task="...")
```

It should support a list of messages, to prepopulate the context of the agents.

```python
result = await team.run(task=[TextMessage(...), TextMessage(...)])
```

This allows passing a previous conversation history from another team. 

This is also useful for nested chats to prepopulate the team participants with messages before generating a response.

Hints from issue discussion:
Interesting idea .. 
This sounds like what `team.load_state()` does in #4100. 
Also, in the above example 

```python
result = await team.run(task=[TextMessage(...), TextMessage(...)])
```

Does this populate the context for ALL agents or some agents etc? 
In PR #4436 ,`team_state` has an `agent_states` field that is a dict and maps to each agent. This way specific list of messages are mapped to the actual agent from which they were serialized (or created for) in `team.load_state()`

```python
from autogen_agentchat.state import BaseTeamState

# convert team state to dict (save to file if needed)
team_state_dict = vars(team_state)
print(team_state_dict)      

# convert back to team state object
team_state = BaseTeamState(**vars(team_state)) 

# load team state
await agent_team.load_state(team_state)
stream = agent_team.run_stream(task="What was the last line of the poem you wrote?")
await Console(stream)
```
I think the list of messages as task is to allow better integration with applications. E.g., in a scenario where users might type multiple messages before the application triggering the team run, and nest chats. 

The load and save states are designed for persistence and editing of agent states. 

They both serve different purposes.

Commit message: feat: enhance task handling to support single and multiple messages in group chat

Patch to review:
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py b/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py
index f617b3823451..d0d9aefe70a7 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/base/_task.py
@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import AsyncGenerator, Protocol, Sequence
+from typing import AsyncGenerator, List, Protocol, Sequence
 
 from autogen_core import CancellationToken
 
@@ -23,7 +23,7 @@ class TaskRunner(Protocol):
     async def run(
         self,
         *,
-        task: str | ChatMessage | None = None,
+        task: str | ChatMessage | List[ChatMessage] | None = None,
         cancellation_token: CancellationToken | None = None,
     ) -> TaskResult:
         """Run the task and return the result.
@@ -36,7 +36,7 @@ async def run(
     def run_stream(
         self,
         *,
-        task: str | ChatMessage | None = None,
+        task: str | ChatMessage | List[ChatMessage] | None = None,
         cancellation_token: CancellationToken | None = None,
     ) -> AsyncGenerator[AgentMessage | TaskResult, None]:
         """Run the task and produces a stream of messages and the final result
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
index fdb79b1197f3..8a9080caa163 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
@@ -149,7 +149,11 @@ async def collect_output_messages(
             if isinstance(message, GroupChatTermination):
                 self._stop_reason = message.message.content
                 return
-            await self._output_message_queue.put(message.message)
+
+            # Handle single message or list of messages
+            messages = message.message if isinstance(message.message, List) else [message.message]
+            for msg in messages:
+                await self._output_message_queue.put(msg)
 
         await ClosureAgent.register_closure(
             runtime,
@@ -164,7 +168,7 @@ async def collect_output_messages(
     async def run(
         self,
         *,
-        task: str | ChatMessage | None = None,
+        task: str | ChatMessage | List[ChatMessage] | None = None,
         cancellation_token: CancellationToken | None = None,
     ) -> TaskResult:
         """Run the team and return the result. The base implementation uses
@@ -172,7 +176,7 @@ async def run(
         Once the team is stopped, the termination condition is reset.
 
         Args:
-            task (str | ChatMessage | None): The task to run the team with.
+            task (str | ChatMessage | List[ChatMessage] | None): The task to run the team with. Can be a string, a single :class:`ChatMessage` , or a list of :class:`ChatMessage`.
             cancellation_token (CancellationToken | None): The cancellation token to kill the task immediately.
                 Setting the cancellation token potentially put the team in an inconsistent state,
                 and it may not reset the termination condition.
@@ -263,7 +267,7 @@ async def main() -> None:
     async def run_stream(
         self,
         *,
-        task: str | ChatMessage | None = None,
+        task: str | ChatMessage | List[ChatMessage] | None = None,
         cancellation_token: CancellationToken | None = None,
     ) -> AsyncGenerator[AgentMessage | TaskResult, None]:
         """Run the team and produces a stream of messages and the final result
@@ -271,7 +275,7 @@ async def run_stream(
         team is stopped, the termination condition is reset.
 
         Args:
-            task (str | ChatMessage | None): The task to run the team with.
+            task (str | ChatMessage | List[ChatMessage] | None): The task to run the team with. Can be a string, a single :class:`ChatMessage` , or a list of :class:`ChatMessage`.
             cancellation_token (CancellationToken | None): The cancellation token to kill the task immediately.
                 Setting the cancellation token potentially put the team in an inconsistent state,
                 and it may not reset the termination condition.
@@ -360,8 +364,17 @@ async def main() -> None:
             pass
         elif isinstance(task, str):
             first_chat_message = TextMessage(content=task, source="user")
-        elif isinstance(task, TextMessage | MultiModalMessage | StopMessage | HandoffMessage):
+        elif isinstance(task, (TextMessage, MultiModalMessage, StopMessage, HandoffMessage)):
             first_chat_message = task
+        elif isinstance(task, list):
+            if not task:
+                raise ValueError("Task list cannot be empty")
+            if not all(isinstance(msg, (TextMessage, MultiModalMessage, StopMessage, HandoffMessage)) for msg in task):
+                raise ValueError("All messages in task list must be valid ChatMessage types")
+            first_chat_message = task[0]
+            # Queue remaining messages for processing
+            for msg in task[1:]:
+                await self._output_message_queue.put(msg)
         else:
             raise ValueError(f"Invalid task type: {type(task)}")
 
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
index aefe4f8d49d8..4b2294da44a9 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
@@ -74,20 +74,28 @@ async def handle_start(self, message: GroupChatStart, ctx: MessageContext) -> No
         await self.validate_group_state(message.message)
 
         if message.message is not None:
-            # Log the start message.
-            await self.publish_message(message, topic_id=DefaultTopicId(type=self._output_topic_type))
+            messages_to_process = [message.message] if isinstance(message.message, ChatMessage) else message.message
 
-            # Relay the start message to the participants.
-            await self.publish_message(
-                message, topic_id=DefaultTopicId(type=self._group_topic_type), cancellation_token=ctx.cancellation_token
-            )
+            # Log and relay each message
+            for msg in messages_to_process:
+                # Log the message
+                await self.publish_message(
+                    GroupChatStart(message=msg), topic_id=DefaultTopicId(type=self._output_topic_type)
+                )
+
+                # Relay the message to participants
+                await self.publish_message(
+                    GroupChatStart(message=msg),
+                    topic_id=DefaultTopicId(type=self._group_topic_type),
+                    cancellation_token=ctx.cancellation_token,
+                )
 
-            # Append the user message to the message thread.
-            self._message_thread.append(message.message)
+                # Append to message thread
+                self._message_thread.append(msg)
 
-            # Check if the conversation should be terminated.
+            # Check termination condition after processing all messages
             if self._termination_condition is not None:
-                stop_message = await self._termination_condition([message.message])
+                stop_message = await self._termination_condition(messages_to_process)
                 if stop_message is not None:
                     await self.publish_message(
                         GroupChatTermination(message=stop_message),
@@ -97,7 +105,7 @@ async def handle_start(self, message: GroupChatStart, ctx: MessageContext) -> No
                     await self._termination_condition.reset()
                     return
 
-        # Select a speaker to start the conversation.
+        # Select a speaker to start/continue the conversation
         speaker_topic_type_future = asyncio.ensure_future(self.select_speaker(self._message_thread))
         # Link the select speaker future to the cancellation token.
         ctx.cancellation_token.link_future(speaker_topic_type_future)
diff --git a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_events.py b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_events.py
index 4ae4d892cace..fa0f556547b2 100644
--- a/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_events.py
+++ b/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_events.py
@@ -1,3 +1,5 @@
+from typing import List
+
 from pydantic import BaseModel
 
 from ...base import Response
@@ -7,8 +9,8 @@
 class GroupChatStart(BaseModel):
     """A request to start a group chat."""
 
-    message: ChatMessage | None = None
-    """An optional user message to start the group chat."""
+    message: ChatMessage | List[ChatMessage] | None = None
+    """An optional user message or list of messages to start the group chat."""
 
 
 class GroupChatAgentResponse(BaseModel):
diff --git a/python/packages/autogen-agentchat/tests/test_group_chat.py b/python/packages/autogen-agentchat/tests/test_group_chat.py
index 5c1d681fca07..b5eb7ff3f69d 100644
--- a/python/packages/autogen-agentchat/tests/test_group_chat.py
+++ b/python/packages/autogen-agentchat/tests/test_group_chat.py
@@ -942,3 +942,50 @@ async def test_swarm_with_handoff_termination() -> None:
     assert result.messages[1].content == "Transferred to second_agent."
     assert result.messages[2].content == "Transferred to third_agent."
     assert result.messages[3].content == "Transferred to non_existing_agent."
+
+
+@pytest.mark.asyncio
+async def test_round_robin_group_chat_with_message_list() -> None:
+    # Create a simple team with echo agents
+    agent1 = _EchoAgent("Agent1", "First agent")
+    agent2 = _EchoAgent("Agent2", "Second agent")
+    termination = MaxMessageTermination(4)  # Stop after 4 messages
+    team = RoundRobinGroupChat([agent1, agent2], termination_condition=termination)
+
+    # Create a list of messages
+    messages = [
+        TextMessage(content="Message 1", source="user"),
+        TextMessage(content="Message 2", source="user"),
+        TextMessage(content="Message 3", source="user"),
+    ]
+
+    # Run the team with the message list
+    result = await team.run(task=messages)
+
+    # Verify the messages were processed in order
+    assert len(result.messages) == 6  # All initial messages + echoes until termination
+    assert result.messages[0].content == "Message 2"  # Second message from queue
+    assert result.messages[1].content == "Message 3"  # Third message from queue
+    assert result.messages[2].content == "Message 1"  # First message (processed first)
+    assert result.messages[3].content == "Message 1"  # Echo from first agent
+    assert result.messages[4].content == "Message 1"  # Echo from second agent
+    assert result.messages[5].content == "Message 1"  # Echo from first agent (before termination)
+    assert result.stop_reason == "Maximum number of messages 4 reached, current message count: 4"
+
+    # Test with streaming
+    await team.reset()
+    index = 0
+    async for message in team.run_stream(task=messages):
+        if isinstance(message, TaskResult):
+            assert message == result
+        else:
+            assert message == result.messages[index]
+            index += 1
+
+    # Test with invalid message list
+    with pytest.raises(ValueError, match="All messages in task list must be valid ChatMessage types"):
+        await team.run(task=["not a message"])
+
+    # Test with empty message list
+    with pytest.raises(ValueError, match="Task list cannot be empty"):
+        await team.run(task=[])


Code review comments:
diff_hunk: @@ -74,20 +74,28 @@ async def handle_start(self, message: GroupChatStart, ctx: MessageContext) -> No
         await self.validate_group_state(message.message)
 
         if message.message is not None:
-            # Log the start message.
-            await self.publish_message(message, topic_id=DefaultTopicId(type=self._output_topic_type))
+            messages_to_process = [message.message] if isinstance(message.message, ChatMessage) else message.message
 
-            # Relay the start message to the participants.
-            await self.publish_message(
-                message, topic_id=DefaultTopicId(type=self._group_topic_type), cancellation_token=ctx.cancellation_token
-            )
+            # Log and relay each message
+            for msg in messages_to_process:
+                # Log the message
+                await self.publish_message(
+                    GroupChatStart(message=msg), topic_id=DefaultTopicId(type=self._output_topic_type)
+                )
+
+                # Relay the message to participants
+                await self.publish_message(
+                    GroupChatStart(message=msg),
+                    topic_id=DefaultTopicId(type=self._group_topic_type),
+                    cancellation_token=ctx.cancellation_token,
+                )
 
-            # Append the user message to the message thread.
-            self._message_thread.append(message.message)
+                # Append to message thread
+                self._message_thread.append(msg)
 
-            # Check if the conversation should be terminated.
+            # Check termination condition after processing all messages
             if self._termination_condition is not None:
-                stop_message = await self._termination_condition([message.message])
+                stop_message = await self._termination_condition(messages_to_process)
path: python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat_manager.py
text: @user1:
The termination condition should be checked after each message is processed to avoid unintended behavior.
```suggestion
                stop_message = await self._termination_condition([msg])
```

diff_hunk: @@ -360,8 +360,17 @@ async def main() -> None:
             pass
         elif isinstance(task, str):
             first_chat_message = TextMessage(content=task, source="user")
-        elif isinstance(task, TextMessage | MultiModalMessage | StopMessage | HandoffMessage):
+        elif isinstance(task, (TextMessage, MultiModalMessage, StopMessage, HandoffMessage)):
             first_chat_message = task
+        elif isinstance(task, list):
+            if not task:
+                raise ValueError("Task list cannot be empty")
+            if not all(isinstance(msg, (TextMessage, MultiModalMessage, StopMessage, HandoffMessage)) for msg in task):
+                raise ValueError("All messages in task list must be valid ChatMessage types")
+            first_chat_message = task[0]
path: python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_base_group_chat.py
text: @author:
- Modified `GroupChatStart` event type in `_events.py` to support list of messages
- Updated `BaseGroupChatManager.handle_start()` to properly handle list of messages
- Added logic to populate team members with messages
- Implemented proper termination condition checking for message lists