Python: [BREAKING] Python: Intro group chat and refactor orchestrations. Fix as_agent(). Standardize orchestration start msg types.

python/packages/core/agent_framework/_workflows/__init__.py

@@ -52,29 +52,29 @@
     handler,
 )
 from ._function_executor import FunctionExecutor, executor
+from ._group_chat import (
+    DEFAULT_MANAGER_INSTRUCTIONS,
+    DEFAULT_MANAGER_STRUCTURED_OUTPUT_PROMPT,
+    GroupChatBuilder,
+    GroupChatDirective,
+    GroupChatStateSnapshot,
+    ManagerDirectiveModel,
+)
 from ._handoff import HandoffBuilder, HandoffUserInputRequest
 from ._magentic import (
     MagenticAgentDeltaEvent,
-    MagenticAgentExecutor,
     MagenticAgentMessageEvent,
     MagenticBuilder,
-    MagenticCallbackEvent,
-    MagenticCallbackMode,
     MagenticContext,
     MagenticFinalResultEvent,
     MagenticManagerBase,
-    MagenticOrchestratorExecutor,
     MagenticOrchestratorMessageEvent,
     MagenticPlanReviewDecision,
     MagenticPlanReviewReply,
     MagenticPlanReviewRequest,
-    MagenticProgressLedger,
-    MagenticProgressLedgerItem,
-    MagenticRequestMessage,
-    MagenticResponseMessage,
-    MagenticStartMessage,
     StandardMagenticManager,
 )
+from ._orchestration_state import OrchestrationState
 from ._request_info_executor import (
     PendingRequestDetails,
     RequestInfoExecutor,
@@ -105,6 +105,8 @@
 from ._workflow_executor import WorkflowExecutor
 
 __all__ = [
+    "DEFAULT_MANAGER_INSTRUCTIONS",
+    "DEFAULT_MANAGER_STRUCTURED_OUTPUT_PROMPT",
     "DEFAULT_MAX_ITERATIONS",
     "AgentExecutor",
     "AgentExecutorRequest",
@@ -128,30 +130,26 @@
     "FileCheckpointStorage",
     "FunctionExecutor",
     "GraphConnectivityError",
+    "GroupChatBuilder",
+    "GroupChatDirective",
+    "GroupChatStateSnapshot",
     "HandoffBuilder",
     "HandoffUserInputRequest",
     "InMemoryCheckpointStorage",
     "InProcRunnerContext",
     "MagenticAgentDeltaEvent",
-    "MagenticAgentExecutor",
     "MagenticAgentMessageEvent",
     "MagenticBuilder",
-    "MagenticCallbackEvent",
-    "MagenticCallbackMode",
     "MagenticContext",
     "MagenticFinalResultEvent",
     "MagenticManagerBase",
-    "MagenticOrchestratorExecutor",
     "MagenticOrchestratorMessageEvent",
     "MagenticPlanReviewDecision",
     "MagenticPlanReviewReply",
     "MagenticPlanReviewRequest",
-    "MagenticProgressLedger",
-    "MagenticProgressLedgerItem",
-    "MagenticRequestMessage",
-    "MagenticResponseMessage",
-    "MagenticStartMessage",
+    "ManagerDirectiveModel",
     "Message",
+    "OrchestrationState",
     "PendingRequestDetails",
     "RequestInfoEvent",
     "RequestInfoExecutor",

python/packages/core/agent_framework/_workflows/__init__.pyi

@@ -50,29 +50,28 @@ from ._executor import (
     handler,
 )
 from ._function_executor import FunctionExecutor, executor
+from ._group_chat import (
+    DEFAULT_MANAGER_INSTRUCTIONS,
+    DEFAULT_MANAGER_STRUCTURED_OUTPUT_PROMPT,
+    GroupChatBuilder,
+    GroupChatDirective,
+    GroupChatStateSnapshot,
+)
 from ._handoff import HandoffBuilder, HandoffUserInputRequest
 from ._magentic import (
     MagenticAgentDeltaEvent,
-    MagenticAgentExecutor,
     MagenticAgentMessageEvent,
     MagenticBuilder,
-    MagenticCallbackEvent,
-    MagenticCallbackMode,
     MagenticContext,
     MagenticFinalResultEvent,
     MagenticManagerBase,
-    MagenticOrchestratorExecutor,
     MagenticOrchestratorMessageEvent,
     MagenticPlanReviewDecision,
     MagenticPlanReviewReply,
     MagenticPlanReviewRequest,
-    MagenticProgressLedger,
-    MagenticProgressLedgerItem,
-    MagenticRequestMessage,
-    MagenticResponseMessage,
-    MagenticStartMessage,
     StandardMagenticManager,
 )
+from ._orchestration_state import OrchestrationState
 from ._request_info_executor import (
     PendingRequestDetails,
     RequestInfoExecutor,
@@ -103,6 +102,8 @@ from ._workflow_context import WorkflowContext
 from ._workflow_executor import WorkflowExecutor
 
 __all__ = [
+    "DEFAULT_MANAGER_INSTRUCTIONS",
+    "DEFAULT_MANAGER_STRUCTURED_OUTPUT_PROMPT",
     "DEFAULT_MAX_ITERATIONS",
     "AgentExecutor",
     "AgentExecutorRequest",
@@ -126,30 +127,25 @@ __all__ = [
     "FileCheckpointStorage",
     "FunctionExecutor",
     "GraphConnectivityError",
+    "GroupChatBuilder",
+    "GroupChatDirective",
+    "GroupChatStateSnapshot",
     "HandoffBuilder",
     "HandoffUserInputRequest",
     "InMemoryCheckpointStorage",
     "InProcRunnerContext",
     "MagenticAgentDeltaEvent",
-    "MagenticAgentExecutor",
     "MagenticAgentMessageEvent",
     "MagenticBuilder",
-    "MagenticCallbackEvent",
-    "MagenticCallbackMode",
     "MagenticContext",
     "MagenticFinalResultEvent",
     "MagenticManagerBase",
-    "MagenticOrchestratorExecutor",
     "MagenticOrchestratorMessageEvent",
     "MagenticPlanReviewDecision",
     "MagenticPlanReviewReply",
     "MagenticPlanReviewRequest",
-    "MagenticProgressLedger",
-    "MagenticProgressLedgerItem",
-    "MagenticRequestMessage",
-    "MagenticResponseMessage",
-    "MagenticStartMessage",
     "Message",
+    "OrchestrationState",
     "PendingRequestDetails",
     "RequestInfoEvent",
     "RequestInfoExecutor",

python/packages/core/agent_framework/_workflows/_agent.py

@@ -3,7 +3,7 @@
 import json
 import logging
 import uuid
-from collections.abc import AsyncIterable, Sequence
+from collections.abc import AsyncIterable
 from dataclasses import dataclass
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast
@@ -19,7 +19,6 @@
     FunctionCallContent,
     FunctionResultContent,
     Role,
-    TextContent,
     UsageDetails,
 )
 
@@ -29,6 +28,7 @@
     RequestInfoEvent,
     WorkflowEvent,
 )
+from ._message_utils import normalize_messages_input
 
 if TYPE_CHECKING:
     from ._workflow import Workflow
@@ -131,7 +131,7 @@ async def run(
         """
         # Collect all streaming updates
         response_updates: list[AgentRunResponseUpdate] = []
-        input_messages = self._normalize_messages(messages)
+        input_messages = normalize_messages_input(messages)
         thread = thread or self.get_new_thread()
         response_id = str(uuid.uuid4())
 
@@ -165,7 +165,7 @@ async def run_stream(
         Yields:
             AgentRunResponseUpdate objects representing the workflow execution progress.
         """
-        input_messages = self._normalize_messages(messages)
+        input_messages = normalize_messages_input(messages)
         thread = thread or self.get_new_thread()
         response_updates: list[AgentRunResponseUpdate] = []
         response_id = str(uuid.uuid4())
@@ -225,28 +225,6 @@ async def _run_stream_impl(
             if update:
                 yield update
 
-    def _normalize_messages(
-        self,
-        messages: str | ChatMessage | Sequence[str] | Sequence[ChatMessage] | None = None,
-    ) -> list[ChatMessage]:
-        """Normalize input messages to a list of ChatMessage objects."""
-        if messages is None:
-            return []
-
-        if isinstance(messages, str):
-            return [ChatMessage(role=Role.USER, contents=[TextContent(text=messages)])]
-
-        if isinstance(messages, ChatMessage):
-            return [messages]
-
-        normalized: list[ChatMessage] = []
-        for msg in messages:
-            if isinstance(msg, str):
-                normalized.append(ChatMessage(role=Role.USER, contents=[TextContent(text=msg)]))
-            elif isinstance(msg, ChatMessage):
-                normalized.append(msg)
-        return normalized
-
     def _convert_workflow_event_to_agent_update(
         self,
         response_id: str,

python/packages/core/agent_framework/_workflows/_agent_executor.py

@@ -12,6 +12,7 @@
     AgentRunUpdateEvent,  # type: ignore[reportPrivateUsage]
 )
 from ._executor import Executor, handler
+from ._message_utils import normalize_messages_input
 from ._workflow_context import WorkflowContext
 
 logger = logging.getLogger(__name__)
@@ -167,7 +168,7 @@ async def from_response(
     @handler
     async def from_str(self, text: str, ctx: WorkflowContext[AgentExecutorResponse, AgentRunResponse]) -> None:
         """Accept a raw user prompt string and run the agent (one-shot)."""
-        self._cache = [ChatMessage(role="user", text=text)]  # type: ignore[arg-type]
+        self._cache = normalize_messages_input(text)
         await self._run_agent_and_emit(ctx)
 
     @handler
@@ -177,15 +178,50 @@ async def from_message(
         ctx: WorkflowContext[AgentExecutorResponse, AgentRunResponse],
     ) -> None:
         """Accept a single ChatMessage as input."""
-        self._cache = [message]
+        self._cache = normalize_messages_input(message)
         await self._run_agent_and_emit(ctx)
 
     @handler
     async def from_messages(
         self,
-        messages: list[ChatMessage],
+        messages: list[str | ChatMessage],
         ctx: WorkflowContext[AgentExecutorResponse, AgentRunResponse],
     ) -> None:
-        """Accept a list of ChatMessage objects as conversation context."""
-        self._cache = list(messages)
+        """Accept a list of chat inputs (strings or ChatMessage) as conversation context."""
+        self._cache = normalize_messages_input(messages)
         await self._run_agent_and_emit(ctx)
+
+    def snapshot_state(self) -> dict[str, Any]:
+        """Capture current executor state for checkpointing.
+
+        Returns:
+            Dict containing serialized cache state
+        """
+        from ._conversation_state import encode_chat_messages
+
+        return {
+            "cache": encode_chat_messages(self._cache),
+        }
+
+    def restore_state(self, state: dict[str, Any]) -> None:
+        """Restore executor state from checkpoint.
+
+        Args:
+            state: Checkpoint data dict
+        """
+        from ._conversation_state import decode_chat_messages
+
+        cache_payload = state.get("cache")
+        if cache_payload:
+            try:
+                self._cache = decode_chat_messages(cache_payload)
+            except Exception as exc:
+                logger.warning("Failed to restore cache: %s", exc)
+                self._cache = []
+        else:
+            self._cache = []
+
+    def reset(self) -> None:
+        """Reset the internal cache of the executor."""
+        logger.debug("AgentExecutor %s: Resetting cache", self.id)
+        self._cache.clear()