from typing import Callable, Type, TypeVar, overload
from ..types.agent import AgentType
from ..agent.base_agent import BaseAgent, subscription_factory
from .subscription_context import SubscriptionInstantiationContext
from .topic_subscription import TopicSubscription
from ..types.exceptions import CantHandleException
[docs]
class DefaultSubscription(TopicSubscription):
"""The default subscription is designed to be a sensible default for applications that only need global scope for agents.
This topic by default uses the "default" topic type and attempts to detect the agent type to use based on the instantiation context.
Args:
topic (str, optional): The topic to subscribe to. Defaults to "default".
agent_type (str, optional): The agent type to use for the subscription. Defaults to None, in which case it will attempt to detect the agent type based on the instantiation context.
"""
def __init__(self, topic: str = "default", agent_type: str | AgentType | None = None):
if agent_type is None:
try:
agent_type = SubscriptionInstantiationContext.agent_type().type
except RuntimeError as e:
raise CantHandleException(
"If agent_type is not specified DefaultSubscription must be created within the subscription callback in AgentEngine.register"
) from e
super().__init__(topic, agent_type)
BaseAgentType = TypeVar("BaseAgentType", bound="BaseAgent")
@overload
def default_subscription() -> Callable[[Type[BaseAgentType]], Type[BaseAgentType]]: ...
@overload
def default_subscription(cls: Type[BaseAgentType]) -> Type[BaseAgentType]: ...
[docs]
def default_subscription(
cls: Type[BaseAgentType] | None = None,
) -> Callable[[Type[BaseAgentType]], Type[BaseAgentType]] | Type[BaseAgentType]:
if cls is None:
return subscription_factory(lambda: [DefaultSubscription()])
else:
return subscription_factory(lambda: [DefaultSubscription()])(cls)
[docs]
def topic_subscription(topic: str) -> Callable[[Type[BaseAgentType]], Type[BaseAgentType]]:
return subscription_factory(lambda: [DefaultSubscription(topic=topic)])