autogen_core._subscription 源代码

from __future__ import annotations

from typing import Awaitable, Callable, Protocol, runtime_checkable

from ._agent_id import AgentId
from ._topic import TopicId


[文档] @runtime_checkable class Subscription(Protocol): """订阅定义了代理感兴趣的主题。""" @property def id(self) -> str: """获取订阅的ID。 实现应返回订阅的唯一ID。通常这是一个UUID。 Returns: str: 订阅的ID。 """ ... def __eq__(self, other: object) -> bool: """检查两个订阅是否相等。 Args: other (object): 要比较的另一个订阅。 Returns: bool: 如果订阅相等则为True,否则为False。 """ if not isinstance(other, Subscription): return False return self.id == other.id
[文档] def is_match(self, topic_id: TopicId) -> bool: """检查给定的 topic_id 是否匹配该订阅。 Args: topic_id (TopicId): 要检查的 TopicId。 Returns: bool: 如果 topic_id 匹配订阅则返回 True,否则返回 False。 """ ...
[文档] def map_to_agent(self, topic_id: TopicId) -> AgentId: """将 topic_id 映射到一个代理。仅当 `is_match` 对给定 topic_id 返回 True 时才应调用此方法。 Args: topic_id (TopicId): 要映射的 TopicId。 Returns: AgentId: 应处理该 topic_id 的代理 ID。 Raises: CantHandleException: 如果订阅无法处理该 topic_id。 """ ...
# Helper alias to represent the lambdas used to define subscriptions UnboundSubscription = Callable[[], list[Subscription] | Awaitable[list[Subscription]]]