Per-(process, instance) registry of transition target / condition classes.
Constructed pre-populated with the built-ins (AgentTarget / RoundRobinTarget / StayTarget / RevertToInitiatorTarget / TerminateTarget and Always / FromSpeaker / ToolCalled).
Tests / multi-tenant callers that need isolation construct their own and pass to TransitionGraph.loads(data, registry=). The module-level register_target / register_condition helpers delegate to :meth:default — a class-cached lazily-initialised singleton — for the common single-tenant case.
Source code in autogen/beta/network/transitions.py
| def __init__(self) -> None:
self._targets: dict[str, type[TransitionTarget]] = {cls.name: cls for cls in _BUILTIN_TARGETS}
self._conditions: dict[str, type[TransitionCondition]] = {cls.name: cls for cls in _BUILTIN_CONDITIONS}
|
classmethod
Return the lazily-initialised process-wide default registry.
Mutated by the module-level register_target / register_condition helpers. Tests that need isolation should construct a fresh TransitionRegistry instead and pass it to TransitionGraph.loads(..., registry=).
Source code in autogen/beta/network/transitions.py
| @classmethod
def default(cls) -> "TransitionRegistry":
"""Return the lazily-initialised process-wide default registry.
Mutated by the module-level ``register_target`` /
``register_condition`` helpers. Tests that need isolation should
construct a fresh ``TransitionRegistry`` instead and pass it to
``TransitionGraph.loads(..., registry=)``.
"""
if cls._DEFAULT is None:
cls._DEFAULT = cls()
return cls._DEFAULT
|
register_target(target_cls)
Register a custom :class:TransitionTarget. Re-registers replace.
Source code in autogen/beta/network/transitions.py
| def register_target(self, target_cls: type[TransitionTarget]) -> None:
"""Register a custom :class:`TransitionTarget`. Re-registers replace."""
self._targets[target_cls.name] = target_cls
|
register_condition(condition_cls)
Register a custom :class:TransitionCondition. Re-registers replace.
Source code in autogen/beta/network/transitions.py
| def register_condition(self, condition_cls: type[TransitionCondition]) -> None:
"""Register a custom :class:`TransitionCondition`. Re-registers replace."""
self._conditions[condition_cls.name] = condition_cls
|
Source code in autogen/beta/network/transitions.py
| def target_from_dict(self, data: dict[str, Any] | None) -> TransitionTarget:
if data is None:
return TerminateTarget()
cls = self._targets.get(data["name"])
if cls is None:
raise WorkflowGraphError(f"no transition target registered for {data['name']!r}")
return cls(**data.get("args", {}))
|
condition_from_dict(data)
Source code in autogen/beta/network/transitions.py
| def condition_from_dict(self, data: dict[str, Any]) -> TransitionCondition:
cls = self._conditions.get(data["name"])
if cls is None:
raise WorkflowGraphError(f"no transition condition registered for {data['name']!r}")
return cls(**data.get("args", {}))
|