autogen_core._agent_proxy 源代码

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Awaitable, Mapping

from ._agent_id import AgentId
from ._agent_metadata import AgentMetadata
from ._cancellation_token import CancellationToken

if TYPE_CHECKING:
    from ._agent_runtime import AgentRuntime




class AgentProxy:
    """一个辅助类，允许你使用 :class:`~autogen_core.AgentId` 来代替其关联的 :class:`~autogen_core.Agent`"""

    def __init__(self, agent: AgentId, runtime: AgentRuntime):
        self._agent = agent
        self._runtime = runtime

    @property
    def id(self) -> AgentId:
        """此代理的目标代理"""
        return self._agent

    @property
    def metadata(self) -> Awaitable[AgentMetadata]:
        """该代理的元数据"""
        return self._runtime.agent_metadata(self._agent)



    async def send_message(
        self,
        message: Any,
        *,
        sender: AgentId,
        cancellation_token: CancellationToken | None = None,
        message_id: str | None = None,
    ) -> Any:
        return await self._runtime.send_message(
            message,
            recipient=self._agent,
            sender=sender,
            cancellation_token=cancellation_token,
            message_id=message_id,
        )




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




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

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