autogen_core._agent 源代码

from typing import TYPE_CHECKING, Any, Mapping, Protocol, runtime_checkable

from ._agent_id import AgentId
from ._agent_metadata import AgentMetadata
from ._message_context import MessageContext

# Forward declaration for type checking only
if TYPE_CHECKING:
    from ._agent_runtime import AgentRuntime




@runtime_checkable
class Agent(Protocol):
    @property
    def metadata(self) -> AgentMetadata:
        """代理的元数据。"""
        ...

    @property
    def id(self) -> AgentId:
        """代理的ID。"""
        ...



    async def bind_id_and_runtime(self, id: AgentId, runtime: "AgentRuntime") -> None:
        """用于将Agent实例绑定到`AgentRuntime`的函数。

        Args:
            agent_id (AgentId): 代理的ID。
            runtime (AgentRuntime): 要绑定代理的AgentRuntime实例。
        """
        ...




    async def on_message(self, message: Any, ctx: MessageContext) -> Any:
        """代理的消息处理器。此方法应仅由运行时调用，而非其他代理。

        Args:
            message (Any): 接收到的消息。类型为`subscriptions`中的一种。
            ctx (MessageContext): 消息上下文。

        Returns:
            Any: 对消息的响应。可为None。

        Raises:
            asyncio.CancelledError: 如果消息被取消。
            CantHandleException: 如果代理无法处理该消息。
        """
        ...




    async def save_state(self) -> Mapping[str, Any]:
        """保存代理的状态。结果必须是可JSON序列化的。"""
        ...




    async def load_state(self, state: Mapping[str, Any]) -> None:
        """从`save_state`获取的代理状态进行加载。

        Args:
            state (Mapping[str, Any]): 代理的状态。必须是可JSON序列化的。
        """

        ...




    async def close(self) -> None:
        """当运行时关闭时调用"""
        ...