Python: Use message cache in agent orchestrations (#12618)

### Motivation and Context

<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->
Some agent threads don't support adding messages. An example of this is
the copilot studio agent.

In an agent orchestration, an agent may receive multiple messages before
it is asked to generate a response. The currently approach in the
orchestrations would add new messages to an agent thread whenever an
agent receives a message, making the copilot studio agent not compatible
with some orchestrations.

### Description

<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->
Instead of adding messages to the agent thread, this PR introduces a
message cache to store messages temporarily before the agent is invoked.


### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [x] The code builds clean without any errors or warnings
- [x] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [x] All unit tests pass, and I have added new tests where possible
- [x] I didn't break anyone 😄

Author: TaoChenOSU

python/semantic_kernel/agents/orchestration/agent_actor_base.py

@@ -63,8 +63,8 @@ def __init__(
         self._streaming_agent_response_callback = streaming_agent_response_callback
 
         self._agent_thread: AgentThread | None = None
-        # Chat history to temporarily store messages before the agent thread is created
-        self._chat_history = ChatHistory()
+        # Chat history to temporarily store messages before each invoke.
+        self._message_cache: ChatHistory = ChatHistory()
 
         ActorBase.__init__(self, description=agent.description or "Semantic Kernel Agent")
 
@@ -146,7 +146,10 @@ def _create_messages(self, additional_messages: DefaultTypeAlias | None = None)
         Returns:
             list[ChatMessageContent]: A list of messages to be sent to the agent.
         """
-        base_messages = self._chat_history.messages[:] if self._agent_thread is None else []
+        base_messages = self._message_cache.messages[:]
+
+        # Clear the message cache for the next invoke.
+        self._message_cache.clear()
 
         if additional_messages is None:
             return base_messages

python/semantic_kernel/agents/orchestration/group_chat.py

@@ -107,27 +107,17 @@ async def _handle_start_message(self, message: GroupChatStartMessage, ctx: Messa
         """Handle the start message for the group chat."""
         logger.debug(f"{self.id}: Received group chat start message.")
         if isinstance(message.body, ChatMessageContent):
-            if self._agent_thread:
-                await self._agent_thread.on_new_message(message.body)
-            else:
-                self._chat_history.add_message(message.body)
+            self._message_cache.add_message(message.body)
         elif isinstance(message.body, list) and all(isinstance(m, ChatMessageContent) for m in message.body):
-            if self._agent_thread:
-                for m in message.body:
-                    await self._agent_thread.on_new_message(m)
-            else:
-                for m in message.body:
-                    self._chat_history.add_message(m)
+            for m in message.body:
+                self._message_cache.add_message(m)
         else:
             raise ValueError(f"Invalid message body type: {type(message.body)}. Expected {DefaultTypeAlias}.")
 
     @message_handler
     async def _handle_response_message(self, message: GroupChatResponseMessage, ctx: MessageContext) -> None:
         logger.debug(f"{self.id}: Received group chat response message.")
-        if self._agent_thread is not None:
-            await self._agent_thread.on_new_message(message.body)
-        else:
-            self._chat_history.add_message(message.body)
+        self._message_cache.add_message(message.body)
 
     @message_handler
     async def _handle_request_message(self, message: GroupChatRequestMessage, ctx: MessageContext) -> None:

python/semantic_kernel/agents/orchestration/handoffs.py

@@ -250,27 +250,18 @@ async def _complete_task(self, task_summary: str) -> None:
     async def _handle_start_message(self, message: HandoffStartMessage, cts: MessageContext) -> None:
         logger.debug(f"{self.id}: Received handoff start message.")
         if isinstance(message.body, ChatMessageContent):
-            if self._agent_thread:
-                await self._agent_thread.on_new_message(message.body)
-            else:
-                self._chat_history.add_message(message.body)
+            self._message_cache.add_message(message.body)
         elif isinstance(message.body, list) and all(isinstance(m, ChatMessageContent) for m in message.body):
             for m in message.body:
-                if self._agent_thread:
-                    await self._agent_thread.on_new_message(m)
-                else:
-                    self._chat_history.add_message(m)
+                self._message_cache.add_message(m)
         else:
             raise ValueError(f"Invalid message body type: {type(message.body)}. Expected {DefaultTypeAlias}.")
 
     @message_handler
     async def _handle_response_message(self, message: HandoffResponseMessage, cts: MessageContext) -> None:
         """Handle a response message from an agent in the handoff group."""
         logger.debug(f"{self.id}: Received handoff response message.")
-        if self._agent_thread is not None:
-            await self._agent_thread.on_new_message(message.body)
-        else:
-            self._chat_history.add_message(message.body)
+        self._message_cache.add_message(message.body)
 
     @message_handler
     async def _handle_request_message(self, message: HandoffRequestMessage, cts: MessageContext) -> None:

python/semantic_kernel/agents/orchestration/magentic.py

@@ -677,10 +677,7 @@ class MagenticAgentActor(AgentActorBase):
     @message_handler
     async def _handle_response_message(self, message: MagenticResponseMessage, ctx: MessageContext) -> None:
         logger.debug(f"{self.id}: Received response message.")
-        if self._agent_thread is not None:
-            await self._agent_thread.on_new_message(message.body)
-        else:
-            self._chat_history.add_message(message.body)
+        self._message_cache.add_message(message.body)
 
     @message_handler
     async def _handle_request_message(self, message: MagenticRequestMessage, ctx: MessageContext) -> None:
@@ -703,7 +700,7 @@ async def _handle_request_message(self, message: MagenticRequestMessage, ctx: Me
     async def _handle_reset_message(self, message: MagenticResetMessage, ctx: MessageContext) -> None:
         """Handle the reset message for the Magentic One group chat."""
         logger.debug(f"{self.id}: Received reset message.")
-        self._chat_history.clear()
+        self._message_cache.clear()
         if self._agent_thread:
             await self._agent_thread.delete()
             self._agent_thread = None