autogen_core._default_subscription 源代码

from typing import Callable, Type, TypeVar, overload

from ._agent_type import AgentType
from ._base_agent import BaseAgent, subscription_factory
from ._subscription_context import SubscriptionInstantiationContext
from ._type_subscription import TypeSubscription
from .exceptions import CantHandleException




class DefaultSubscription(TypeSubscription):
    """默认订阅设计用于那些仅需要全局作用域代理的应用程序，是一个合理的默认选择。

    该主题默认使用"default"主题类型，并尝试根据实例化上下文检测要使用的代理类型。

    Args:
        topic_type (str, optional): 要订阅的主题类型。默认为"default"。
        agent_type (str, optional): 用于订阅的代理类型。默认为None，此时将尝试根据实例化上下文检测代理类型。
    """

    def __init__(self, topic_type: 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 AgentRuntime.register"
                ) from e

        super().__init__(topic_type, 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]: ...




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)





def type_subscription(topic_type: str) -> Callable[[Type[BaseAgentType]], Type[BaseAgentType]]:
    return subscription_factory(lambda: [DefaultSubscription(topic_type=topic_type)])