mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Fix spurious Magentic custom manager warning (#6261)
* Fix magentic manager warning
* Use typing_extensions.Sentinel for _MISSING sentinel value
Replace the bare object() sentinel with typing_extensions.Sentinel per
PEP 661 (now final). Sentinel provides a proper name and repr
('<_MISSING>') and is the idiomatic approach going forward.
Refs #4306
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
* fix: correct Sentinel type annotation for max_stall_count param (#6261)
Use int | Sentinel for max_stall_count parameter type annotation instead
of int with cast(Any, _MISSING) to properly express that the parameter
can hold either an int or the _MISSING sentinel value. This fixes the
pyright reportUnnecessaryComparison errors caused by the types int and
Sentinel having no overlap.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
* Rename _MISSING sentinel to UNSET in orchestrations
The sentinel is user-visible as a default in public init signatures, so
use UNSET (no leading underscore) instead of the private _MISSING name.
Drop the now-unnecessary reportPrivateUsage ignores on the UNSET imports.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---------
Co-authored-by: Copilot <copilot@github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
co-authored by
Copilot
Copilot
parent
fe08574a7c
commit
4268080c20
@@ -1,5 +1,6 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from collections.abc import AsyncIterable, Awaitable, Sequence
|
||||
from dataclasses import dataclass
|
||||
@@ -987,6 +988,33 @@ def test_magentic_builder_requires_exactly_one_manager_option():
|
||||
MagenticBuilder(participants=[agent], manager=manager, manager_factory=manager_factory)
|
||||
|
||||
|
||||
def test_magentic_with_custom_manager_does_not_warn_without_standard_manager_options(caplog: Any) -> None:
|
||||
caplog.set_level(logging.WARNING, logger="agent_framework_orchestrations._magentic")
|
||||
|
||||
MagenticBuilder(participants=[StubAgent("agentA", "reply")], manager=FakeManager())
|
||||
|
||||
assert "Custom manager provided; all other manager arguments will be ignored." not in caplog.text
|
||||
|
||||
|
||||
def test_magentic_with_custom_manager_factory_does_not_warn_without_standard_manager_options(caplog: Any) -> None:
|
||||
caplog.set_level(logging.WARNING, logger="agent_framework_orchestrations._magentic")
|
||||
|
||||
def manager_factory() -> MagenticManagerBase:
|
||||
return FakeManager()
|
||||
|
||||
MagenticBuilder(participants=[StubAgent("agentA", "reply")], manager_factory=manager_factory)
|
||||
|
||||
assert "Custom manager provided; all other manager arguments will be ignored." not in caplog.text
|
||||
|
||||
|
||||
def test_magentic_with_custom_manager_warns_when_standard_manager_option_is_provided(caplog: Any) -> None:
|
||||
caplog.set_level(logging.WARNING, logger="agent_framework_orchestrations._magentic")
|
||||
|
||||
MagenticBuilder(participants=[StubAgent("agentA", "reply")], manager=FakeManager(), max_stall_count=3)
|
||||
|
||||
assert "Custom manager provided; all other manager arguments will be ignored." in caplog.text
|
||||
|
||||
|
||||
async def test_magentic_with_manager_factory():
|
||||
"""Test workflow creation using manager_factory."""
|
||||
factory_call_count = 0
|
||||
@@ -1037,6 +1065,20 @@ async def test_magentic_with_agent_factory():
|
||||
assert event_count > 0
|
||||
|
||||
|
||||
def test_magentic_agent_factory_uses_default_max_stall_count() -> 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()
|
||||
|
||||
orchestrator = next(e for e in workflow.executors.values() if isinstance(e, MagenticOrchestrator))
|
||||
manager = orchestrator._manager # type: ignore[reportPrivateUsage]
|
||||
|
||||
assert isinstance(manager, StandardMagenticManager)
|
||||
assert manager.max_stall_count == 3
|
||||
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user