From 1cd2645efcde6f18554e2e692e3921e3345019e8 Mon Sep 17 00:00:00 2001 From: eavanvalkenburg Date: Mon, 21 Sep 2026 13:11:49 +0200 Subject: [PATCH] Create Magentic managers per workflow --- .../_magentic.py | 22 +++- .../orchestrations/tests/test_magentic.py | 117 +++++++++++++++++- 2 files changed, 128 insertions(+), 11 deletions(-) diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py index a4218c2798b..fc828b0e095 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_magentic.py @@ -10,6 +10,7 @@ from collections.abc import Callable, Sequence from dataclasses import dataclass, field from enum import Enum +from functools import partial from typing import Any, ClassVar, Literal, TypeVar, cast from agent_framework import ( @@ -1435,9 +1436,14 @@ def __init__( name: Optional workflow identifier. Defaults to ``"Magentic"``. participants: Sequence of agent or executor instances for the workflow. manager: Pre-configured manager instance (subclass of MagenticManagerBase). - manager_factory: Callable that returns a new MagenticManagerBase instance. - manager_agent: Agent instance for creating a StandardMagenticManager. - manager_agent_factory: Callable that returns a new agent instance for creating a StandardMagenticManager. + The same instance, including its mutable state, is reused by every workflow + built with this builder. Do not share a stateful manager across concurrent or + interleaved workflows; use manager_factory to create a fresh manager per build. + manager_factory: Callable invoked on each build to return a new MagenticManagerBase instance. + manager_agent: Agent instance used to create a new StandardMagenticManager on each build. + Each workflow has its own manager, but the supplied agent is shared. + manager_agent_factory: Callable invoked on each build to return a new agent instance + for a new StandardMagenticManager. task_ledger: Optional custom task ledger (used with manager_agent/manager_agent_factory). task_ledger_facts_prompt: Custom prompt for extracting facts. task_ledger_plan_prompt: Custom prompt for generating initial plan. @@ -1691,7 +1697,8 @@ def _log_warning_if_constructor_args_provided() -> None: self._manager = manager _log_warning_if_constructor_args_provided() elif manager_agent is not None: - self._manager = StandardMagenticManager( + self._manager_factory = partial( + StandardMagenticManager, agent=manager_agent, task_ledger=task_ledger, task_ledger_facts_prompt=task_ledger_facts_prompt, @@ -1781,7 +1788,12 @@ def _resolve_participants(self) -> list[Executor]: return executors def build(self) -> Workflow: - """Build a Magentic workflow with the orchestrator and all agent executors.""" + """Build a Magentic workflow with the orchestrator and all agent executors. + + A new standard manager is created for each build when configured with manager_agent + or manager_agent_factory. An explicitly supplied manager instance is reused unchanged; + manager_factory is called on each build to obtain a manager. + """ mark_feature_used(FeatureIndex.ORCHESTRATION_MAGENTIC) logger.info(f"Building Magentic workflow with {len(self._participants)} participants") diff --git a/python/packages/orchestrations/tests/test_magentic.py b/python/packages/orchestrations/tests/test_magentic.py index 5e085799acf..e9f8df1fea1 100644 --- a/python/packages/orchestrations/tests/test_magentic.py +++ b/python/packages/orchestrations/tests/test_magentic.py @@ -1102,12 +1102,17 @@ def agent_factory() -> SupportsAgentRun: assert event_count > 0 -def test_magentic_agent_factory_uses_default_max_stall_count() -> None: +@pytest.mark.parametrize("use_agent_factory", [False, True]) +def test_magentic_manager_agent_uses_default_max_stall_count(use_agent_factory: bool) -> None: def agent_factory() -> SupportsAgentRun: return cast(SupportsAgentRun, StubManagerAgent()) participant = StubAgent("agentA", "reply from agentA") - workflow = MagenticBuilder(participants=[participant], manager_agent_factory=agent_factory).build() + workflow = MagenticBuilder( + participants=[participant], + manager_agent=None if use_agent_factory else StubManagerAgent(), + manager_agent_factory=agent_factory if use_agent_factory else None, + ).build() orchestrator = next(e for e in workflow.executors.values() if isinstance(e, MagenticOrchestrator)) manager = orchestrator._manager # type: ignore[reportPrivateUsage] @@ -1116,6 +1121,103 @@ def agent_factory() -> SupportsAgentRun: assert manager.max_stall_count == 3 +def test_magentic_manager_agent_reusable_builder() -> None: + agent = StubManagerAgent() + builder = MagenticBuilder(participants=[StubAgent("agentA", "reply")], manager_agent=agent) + + assert builder._manager is None + workflow1 = builder.build() + workflow2 = builder.build() + orchestrator1 = next(e for e in workflow1.executors.values() if isinstance(e, MagenticOrchestrator)) + orchestrator2 = next(e for e in workflow2.executors.values() if isinstance(e, MagenticOrchestrator)) + manager1 = orchestrator1._manager + manager2 = orchestrator2._manager + + assert isinstance(manager1, StandardMagenticManager) + assert isinstance(manager2, StandardMagenticManager) + assert manager1 is not manager2 + assert manager1._agent is manager2._agent is agent + assert manager1.task_ledger is None + assert manager2.task_ledger is None + + +def test_magentic_supplied_manager_reusable_builder_preserves_identity() -> None: + manager = FakeManager() + builder = MagenticBuilder(participants=[StubAgent("agentA", "reply")], manager=manager) + + workflow1 = builder.build() + workflow2 = builder.build() + orchestrator1 = next(e for e in workflow1.executors.values() if isinstance(e, MagenticOrchestrator)) + orchestrator2 = next(e for e in workflow2.executors.values() if isinstance(e, MagenticOrchestrator)) + + assert orchestrator1 is not orchestrator2 + assert orchestrator1._manager is orchestrator2._manager is manager + + +@pytest.mark.parametrize("use_agent_factory", [False, True]) +async def test_magentic_reusable_builder_keeps_interleaved_task_ledgers_separate(use_agent_factory: bool) -> None: + class RecordingChatClient(BaseChatClient): + def __init__(self) -> None: + super().__init__() + self.calls: list[list[Message]] = [] + + @override + def _inner_get_response(self, *, messages, stream, options, **kwargs): # type: ignore[override] + self.calls.append(list(messages)) + + async def _get() -> ChatResponse: + return ChatResponse(messages=Message("assistant", [messages[-1].text])) + + return _get() + + client = RecordingChatClient() + agent = Agent(name="manager", client=client) + builder = MagenticBuilder( + participants=[StubAgent("agentA", "reply")], + manager_agent=None if use_agent_factory else agent, + manager_agent_factory=(lambda: agent) if use_agent_factory else None, + enable_plan_review=True, + task_ledger_facts_prompt="FACTS:{task}", + task_ledger_plan_prompt="PLAN", + task_ledger_facts_update_prompt="UPDATE:{task}:{old_facts}", + task_ledger_plan_update_prompt="REVISED PLAN", + ) + workflow_a = builder.build() + workflow_b = builder.build() + + events_a = [event async for event in workflow_a.run("A_ONLY", stream=True)] + events_b = [event async for event in workflow_b.run("B_ONLY", stream=True)] + review_a = next(event for event in events_a if event.type == "request_info") + review_b = next(event for event in events_b if event.type == "request_info") + assert isinstance(review_a.data, MagenticPlanReviewRequest) + assert isinstance(review_b.data, MagenticPlanReviewRequest) + assert len(client.calls) == 4 + + revised_events = [ + event + async for event in workflow_a.run( + stream=True, responses={review_a.request_id: review_a.data.revise("Revise the plan")} + ) + ] + assert len(client.calls) == 6 + assert client.calls[4][-1].text == "UPDATE:A_ONLY:FACTS:A_ONLY" + assert all("B_ONLY" not in message.text for call in client.calls[4:] for message in call) + revised_review = next(event for event in revised_events if event.type == "request_info") + assert isinstance(revised_review.data, MagenticPlanReviewRequest) + assert "UPDATE:A_ONLY:FACTS:A_ONLY" in revised_review.data.plan.text + assert "B_ONLY" not in revised_review.data.plan.text + + _ = [ + event + async for event in workflow_b.run( + stream=True, responses={review_b.request_id: review_b.data.revise("Revise the plan")} + ) + ] + assert len(client.calls) == 8 + assert client.calls[6][-1].text == "UPDATE:B_ONLY:FACTS:B_ONLY" + assert all("A_ONLY" not in message.text for call in client.calls[6:] for message in call) + + async def test_magentic_manager_factory_reusable_builder(): """Test that the builder can be reused to build multiple workflows with manager factory.""" factory_call_count = 0 @@ -1140,10 +1242,12 @@ def manager_factory() -> MagenticManagerBase: orchestrator1 = next(e for e in wf1.executors.values() if isinstance(e, MagenticOrchestrator)) orchestrator2 = next(e for e in wf2.executors.values() if isinstance(e, MagenticOrchestrator)) assert orchestrator1 is not orchestrator2 + assert orchestrator1._manager is not orchestrator2._manager -def test_magentic_agent_factory_with_standard_manager_options(): - """Test that agent_factory properly passes through standard manager options.""" +@pytest.mark.parametrize("use_agent_factory", [False, True]) +def test_magentic_manager_agent_with_standard_manager_options(use_agent_factory: bool) -> None: + """Both manager agent configurations pass through standard manager options.""" factory_call_count = 0 def agent_factory() -> SupportsAgentRun: @@ -1174,7 +1278,8 @@ def agent_factory() -> SupportsAgentRun: participant = StubAgent("agentA", "reply from agentA") workflow = MagenticBuilder( participants=[participant], - manager_agent_factory=agent_factory, + manager_agent=None if use_agent_factory else StubManagerAgent(), + manager_agent_factory=agent_factory if use_agent_factory else None, task_ledger=custom_task_ledger, max_stall_count=custom_max_stall_count, max_reset_count=custom_max_reset_count, @@ -1189,7 +1294,7 @@ def agent_factory() -> SupportsAgentRun: ).build() # Factory should be called during build - assert factory_call_count == 1 + assert factory_call_count == (1 if use_agent_factory else 0) # Get the orchestrator and verify the manager has the custom options orchestrator = next(e for e in workflow.executors.values() if isinstance(e, MagenticOrchestrator))