Autogen_core源码：_agent.py

_agent.py代码

from typing import Any, Mapping, Protocol, runtime_checkable

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


@runtime_checkable
class Agent(Protocol):
    @property
    def metadata(self) -> AgentMetadata:
        """Metadata of the agent."""
        ...

    @property
    def id(self) -> AgentId:
        """ID of the agent."""
        ...

    async def on_message(self, message: Any, ctx: MessageContext) -> Any:
        """Message handler for the agent. This should only be called by the runtime, not by other agents.

        Args:
            message (Any): Received message. Type is one of the types in `subscriptions`.
            ctx (MessageContext): Context of the message.

        Returns:
            Any: Response to the message. Can be None.

        Raises:
            asyncio.CancelledError: If the message was cancelled.
            CantHandleException: If the agent cannot handle the message.
        """
        ...

    async def save_state(self) -> Mapping[str, Any]:
        """Save the state of the agent. The result must be JSON serializable."""
        ...

    async def load_state(self, state: Mapping[str, Any]) -> None:
        """Load in the state of the agent obtained from `save_state`.

        Args:
            state (Mapping[str, Any]): State of the agent. Must be JSON serializable.
        """

        ...

    async def close(self) -> None:
        """Called when the runtime is closed"""
        ...

代码解释

这段代码定义了一个名为 Agent 的协议（Protocol），协议在 Python 中用于定义接口规范，它描述了一个对象应该具备的属性和方法，但不提供具体的实现。下面详细解释代码逻辑和功能：

导入模块

from typing import Any, Mapping, Protocol, runtime_checkable

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

• typing 模块中的 Any 表示任意类型，Mapping 表示映射类型（如字典），Protocol 用于定义协议，runtime_checkable 装饰器允许在运行时检查一个对象是否实现了该协议。
• 从 . 包中导入 AgentId、AgentMetadata 和 MessageContext 类，这些类可能用于表示代理的 ID、元数据和消息上下文。

定义 Agent 协议

@runtime_checkable
class Agent(Protocol):

• @runtime_checkable 装饰器使得可以在运行时使用 isinstance() 函数来检查一个对象是否实现了 Agent 协议。
• Agent 类继承自 Protocol，表示这是一个协议类。

定义属性

    @property
    def metadata(self) -> AgentMetadata:
        """Metadata of the agent."""
       ...

    @property
    def id(self) -> AgentId:
        """ID of the agent."""
       ...

• metadata 属性：返回一个 AgentMetadata 类型的对象，表示代理的元数据。
• id 属性：返回一个 AgentId 类型的对象，表示代理的 ID。

定义方法

on_message 方法

    async def on_message(self, message: Any, ctx: MessageContext) -> Any:
        """Message handler for the agent. This should only be called by the runtime, not by other agents.

        Args:
            message (Any): Received message. Type is one of the types in `subscriptions`.
            ctx (MessageContext): Context of the message.

        Returns:
            Any: Response to the message. Can be None.

        Raises:
            asyncio.CancelledError: If the message was cancelled.
            CantHandleException: If the agent cannot handle the message.
        """
       ...

• 这是一个异步方法，用于处理接收到的消息。
• message 参数表示接收到的消息，类型可以是任意类型。
• ctx 参数表示消息的上下文，类型为 MessageContext。
• 方法返回一个任意类型的对象，表示对消息的响应，也可以返回 None。
• 方法可能会抛出 asyncio.CancelledError 异常，表示消息被取消，还可能抛出 CantHandleException 异常，表示代理无法处理该消息。

save_state 方法

    async def save_state(self) -> Mapping[str, Any]:
        """Save the state of the agent. The result must be JSON serializable."""
       ...

• 这是一个异步方法，用于保存代理的状态。
• 方法返回一个映射类型的对象，键为字符串，值为任意类型，且该对象必须是 JSON 可序列化的。

load_state 方法

    async def load_state(self, state: Mapping[str, Any]) -> None:
        """Load in the state of the agent obtained from `save_state`.

        Args:
            state (Mapping[str, Any]): State of the agent. Must be JSON serializable.
        """
       ...

• 这是一个异步方法，用于加载代理的状态。
• state 参数是一个映射类型的对象，键为字符串，值为任意类型，且该对象必须是 JSON 可序列化的。

close 方法

    async def close(self) -> None:
        """Called when the runtime is closed"""
       ...

• 这是一个异步方法，当运行时关闭时调用，用于执行一些清理操作。

总结

这段代码定义了一个 Agent 协议，描述了一个代理对象应该具备的属性和方法，包括代理的元数据、ID、消息处理、状态保存和加载以及关闭操作。任何实现了这些属性和方法的类都可以被视为一个 Agent 对象，并且可以使用 isinstance() 函数在运行时进行检查。这个协议可以作为其他代理类的接口规范，确保它们具备必要的功能。

代码示例

示例 1：实现 Agent 协议的简单类

import asyncio
from typing import Any, Mapping
from autogen_core import AgentId,AgentMetadata,MessageContext,CancellationToken,Agent

class CantHandleException(Exception):
    pass

class SimpleAgent:
    def __init__(self, agent_id: AgentId, metadata: AgentMetadata):
        self._id = agent_id
        self._metadata = metadata
        self._state = {}

    @property
    def metadata(self) -> AgentMetadata:
        return self._metadata

    @property
    def id(self) -> AgentId:
        return self._id

    async def on_message(self, message: Any, ctx: MessageContext) -> Any:
        try:
            # 简单示例，只是返回消息的字符串形式
            return str(message)
        except Exception:
            raise CantHandleException("Can't handle this message")

    async def save_state(self) -> Mapping[str, Any]:
        return self._state

    async def load_state(self, state: Mapping[str, Any]) -> None:
        self._state = state

    async def close(self) -> None:
        print(f"Agent {self._id} is closing.")




# 使用示例
if __name__ == "__main__":
    agent_id = AgentId("test_agent","default")
    metadata = AgentMetadata(type="Test Agent", key = "default", description="A simple test agent")
    agent = SimpleAgent(agent_id, metadata)

    message = {"key": "value"}
    ctx = MessageContext(sender=AgentId("sender_agent","default"), topic_id=None, is_rpc=False, message_id="", cancellation_token=CancellationToken())

    async def test_agent():
        response = await agent.on_message(message, ctx)
        print(f"Response: {response}")

        state = await agent.save_state()
        print(f"Saved state: {state}")

        new_state = {"new_key": "new_value"}
        await agent.load_state(new_state)
        print(f"Loaded state: {await agent.save_state()}")

        await agent.close()

    await test_agent()

Response: {'key': 'value'}
Saved state: {}
Loaded state: {'new_key': 'new_value'}
Agent test_agent/default is closing.

示例 2：检查对象是否实现 Agent 协议

class SimpleAgent:
    def __init__(self, agent_id: AgentId, metadata: AgentMetadata):
        self._id = agent_id
        self._metadata = metadata

    @property
    def metadata(self) -> AgentMetadata:
        return self._metadata

    @property
    def id(self) -> AgentId:
        return self._id

    async def on_message(self, message: Any, ctx: MessageContext) -> Any:
        return str(message)

    async def save_state(self) -> Mapping[str, Any]:
        return {}

    async def load_state(self, state: Mapping[str, Any]) -> None:
        pass

    async def close(self) -> None:
        pass


agent_id = AgentId("test_agent","default")
metadata = AgentMetadata(type="Test Agent", key = "default", description="A simple test agent")
agent = SimpleAgent(agent_id, metadata)

if isinstance(agent, Agent):
    print("The object implements the Agent protocol.")
else:
    print("The object does not implement the Agent protocol.")

The object implements the Agent protocol.