[BREAKING] Python: Refactor orchestrations

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

@@ -6,6 +6,13 @@
     AgentExecutorRequest,
     AgentExecutorResponse,
 )
+from ._agent_utils import resolve_agent_id
+from ._base_group_chat_orchestrator import (
+    BaseGroupChatOrchestrator,
+    GroupChatRequestMessage,
+    GroupChatRequestSentEvent,
+    GroupChatResponseReceivedEvent,
+)
 from ._checkpoint import (
     CheckpointStorage,
     FileCheckpointStorage,
@@ -56,37 +63,30 @@
 )
 from ._function_executor import FunctionExecutor, executor
 from ._group_chat import (
-    DEFAULT_MANAGER_INSTRUCTIONS,
-    DEFAULT_MANAGER_STRUCTURED_OUTPUT_PROMPT,
+    AgentBasedGroupChatOrchestrator,
     GroupChatBuilder,
-    GroupChatDirective,
-    GroupChatStateSnapshot,
-    ManagerDirectiveModel,
-    ManagerSelectionRequest,
-    ManagerSelectionResponse,
+    GroupChatState,
 )
-from ._handoff import HandoffBuilder, HandoffUserInputRequest
+from ._handoff import HandoffAgentUserRequest, HandoffBuilder, HandoffSentEvent
 from ._magentic import (
-    MAGENTIC_EVENT_TYPE_AGENT_DELTA,
-    MAGENTIC_EVENT_TYPE_ORCHESTRATOR,
     ORCH_MSG_KIND_INSTRUCTION,
     ORCH_MSG_KIND_NOTICE,
     ORCH_MSG_KIND_TASK_LEDGER,
     ORCH_MSG_KIND_USER_TASK,
     MagenticBuilder,
     MagenticContext,
-    MagenticHumanInputRequest,
-    MagenticHumanInterventionDecision,
-    MagenticHumanInterventionKind,
-    MagenticHumanInterventionReply,
-    MagenticHumanInterventionRequest,
     MagenticManagerBase,
-    MagenticStallInterventionDecision,
-    MagenticStallInterventionReply,
-    MagenticStallInterventionRequest,
+    MagenticOrchestrator,
+    MagenticOrchestratorEvent,
+    MagenticOrchestratorEventType,
+    MagenticPlanReviewRequest,
+    MagenticPlanReviewResponse,
+    MagenticProgressLedger,
+    MagenticProgressLedgerItem,
+    MagenticResetSignal,
     StandardMagenticManager,
 )
-from ._orchestration_request_info import AgentInputRequest, AgentResponseReviewRequest, RequestInfoInterceptor
+from ._orchestration_request_info import AgentRequestInfoResponse
 from ._orchestration_state import OrchestrationState
 from ._request_info_mixin import response_handler
 from ._runner import Runner
@@ -112,22 +112,19 @@
 from ._workflow_executor import SubWorkflowRequestMessage, SubWorkflowResponseMessage, WorkflowExecutor
 
 __all__ = [
-    "DEFAULT_MANAGER_INSTRUCTIONS",
-    "DEFAULT_MANAGER_STRUCTURED_OUTPUT_PROMPT",
     "DEFAULT_MAX_ITERATIONS",
-    "MAGENTIC_EVENT_TYPE_AGENT_DELTA",
-    "MAGENTIC_EVENT_TYPE_ORCHESTRATOR",
     "ORCH_MSG_KIND_INSTRUCTION",
     "ORCH_MSG_KIND_NOTICE",
     "ORCH_MSG_KIND_TASK_LEDGER",
     "ORCH_MSG_KIND_USER_TASK",
+    "AgentBasedGroupChatOrchestrator",
     "AgentExecutor",
     "AgentExecutorRequest",
     "AgentExecutorResponse",
-    "AgentInputRequest",
-    "AgentResponseReviewRequest",
+    "AgentRequestInfoResponse",
     "AgentRunEvent",
     "AgentRunUpdateEvent",
+    "BaseGroupChatOrchestrator",
     "Case",
     "CheckpointStorage",
     "ConcurrentBuilder",
@@ -146,30 +143,29 @@
     "FunctionExecutor",
     "GraphConnectivityError",
     "GroupChatBuilder",
-    "GroupChatDirective",
-    "GroupChatStateSnapshot",
+    "GroupChatRequestMessage",
+    "GroupChatRequestSentEvent",
+    "GroupChatResponseReceivedEvent",
+    "GroupChatState",
+    "HandoffAgentUserRequest",
     "HandoffBuilder",
-    "HandoffUserInputRequest",
+    "HandoffSentEvent",
     "InMemoryCheckpointStorage",
     "InProcRunnerContext",
     "MagenticBuilder",
     "MagenticContext",
-    "MagenticHumanInputRequest",
-    "MagenticHumanInterventionDecision",
-    "MagenticHumanInterventionKind",
-    "MagenticHumanInterventionReply",
-    "MagenticHumanInterventionRequest",
     "MagenticManagerBase",
-    "MagenticStallInterventionDecision",
-    "MagenticStallInterventionReply",
-    "MagenticStallInterventionRequest",
-    "ManagerDirectiveModel",
-    "ManagerSelectionRequest",
-    "ManagerSelectionResponse",
+    "MagenticOrchestrator",
+    "MagenticOrchestratorEvent",
+    "MagenticOrchestratorEventType",
+    "MagenticPlanReviewRequest",
+    "MagenticPlanReviewResponse",
+    "MagenticProgressLedger",
+    "MagenticProgressLedgerItem",
+    "MagenticResetSignal",
     "Message",
     "OrchestrationState",
     "RequestInfoEvent",
-    "RequestInfoInterceptor",
     "Runner",
     "RunnerContext",
     "SequentialBuilder",
@@ -208,6 +204,7 @@
     "executor",
     "get_checkpoint_summary",
     "handler",
+    "resolve_agent_id",
     "response_handler",
     "validate_workflow_graph",
 ]

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

@@ -10,6 +10,7 @@
 from .._agents import AgentProtocol, ChatAgent
 from .._threads import AgentThread
 from .._types import AgentRunResponse, AgentRunResponseUpdate, ChatMessage
+from ._agent_utils import resolve_agent_id
 from ._checkpoint_encoding import decode_checkpoint_value, encode_checkpoint_value
 from ._const import WORKFLOW_RUN_KWARGS_KEY
 from ._conversation_state import encode_chat_messages
@@ -88,16 +89,20 @@ def __init__(
             id: A unique identifier for the executor. If None, the agent's name will be used if available.
         """
         # Prefer provided id; else use agent.name if present; else generate deterministic prefix
-        exec_id = id or agent.name
+        exec_id = id or resolve_agent_id(agent)
         if not exec_id:
-            raise ValueError("Agent must have a name or an explicit id must be provided.")
+            raise ValueError("Agent must have a non-empty name or id or an explicit id must be provided.")
         super().__init__(exec_id)
         self._agent = agent
         self._agent_thread = agent_thread or self._agent.get_new_thread()
         self._pending_agent_requests: dict[str, FunctionApprovalRequestContent] = {}
         self._pending_responses_to_agent: list[FunctionApprovalResponseContent] = []
         self._output_response = output_response
+
+        # AgentExecutor maintains an internal cache of messages in between runs
         self._cache: list[ChatMessage] = []
+        # This tracks the full conversation after each run
+        self._full_conversation: list[ChatMessage] = []
 
     @property
     def output_response(self) -> bool:
@@ -227,6 +232,7 @@ async def on_checkpoint_save(self) -> dict[str, Any]:
 
         return {
             "cache": encode_chat_messages(self._cache),
+            "full_conversation": encode_chat_messages(self._full_conversation),
             "agent_thread": serialized_thread,
             "pending_agent_requests": encode_checkpoint_value(self._pending_agent_requests),
             "pending_responses_to_agent": encode_checkpoint_value(self._pending_responses_to_agent),
@@ -251,6 +257,16 @@ async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
         else:
             self._cache = []
 
+        full_conversation_payload = state.get("full_conversation")
+        if full_conversation_payload:
+            try:
+                self._full_conversation = decode_chat_messages(full_conversation_payload)
+            except Exception as exc:
+                logger.warning("Failed to restore full conversation: %s", exc)
+                self._full_conversation = []
+        else:
+            self._full_conversation = []
+
         thread_payload = state.get("agent_thread")
         if thread_payload:
             try:
@@ -289,6 +305,12 @@ async def _run_agent_and_emit(self, ctx: WorkflowContext[AgentExecutorResponse,
             # Non-streaming mode: use run() and emit single event
             response = await self._run_agent(cast(WorkflowContext, ctx))
 
+        # Always extend full conversation with cached messages plus agent outputs
+        # (agent_run_response.messages) after each run. This is to avoid losing context
+        # when agent did not complete and the cache is cleared when responses come back.
+        # Do not mutate response.messages so AgentRunEvent remains faithful to the raw output.
+        self._full_conversation.extend(list(self._cache) + (list(response.messages) if response else []))
+
         if response is None:
             # Agent did not complete (e.g., waiting for user input); do not emit response
             logger.info("AgentExecutor %s: Agent did not complete, awaiting user input", self.id)
@@ -297,12 +319,7 @@ async def _run_agent_and_emit(self, ctx: WorkflowContext[AgentExecutorResponse,
         if self._output_response:
             await ctx.yield_output(response)
 
-        # Always construct a full conversation snapshot from inputs (cache)
-        # plus agent outputs (agent_run_response.messages). Do not mutate
-        # response.messages so AgentRunEvent remains faithful to the raw output.
-        full_conversation: list[ChatMessage] = list(self._cache) + list(response.messages)
-
-        agent_response = AgentExecutorResponse(self.id, response, full_conversation=full_conversation)
+        agent_response = AgentExecutorResponse(self.id, response, full_conversation=self._full_conversation)
         await ctx.send_message(agent_response)
         self._cache.clear()
 

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

@@ -0,0 +1,17 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from .._agents import AgentProtocol
+
+
+def resolve_agent_id(agent: AgentProtocol) -> str:
+    """Resolve the unique identifier for an agent.
+
+    Prefers the `.name` attribute if set; otherwise falls back to `.id`.
+
+    Args:
+        agent: The agent whose identifier is to be resolved.
+
+    Returns:
+        The resolved unique identifier for the agent.
+    """
+    return agent.name if agent.name else agent.id