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]]]