autogen_agentchat.base._chat_agent 源代码

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Mapping, Sequence

from autogen_core import CancellationToken, ComponentBase
from pydantic import BaseModel

from ..messages import BaseAgentEvent, BaseChatMessage
from ._task import TaskRunner




@dataclass(kw_only=True)
class Response:
    """调用 :meth:`ChatAgent.on_messages` 方法返回的响应。"""

    chat_message: BaseChatMessage
    """由代理生成的聊天消息作为响应。"""

    inner_messages: Sequence[BaseAgentEvent | BaseChatMessage] | None = None
    """代理生成的内部消息，可以是 :class:`BaseAgentEvent`
    或 :class:`BaseChatMessage`。"""





class ChatAgent(ABC, TaskRunner, ComponentBase[BaseModel]):
    """聊天代理的协议。"""

    component_type = "agent"

    @property
    @abstractmethod
    def name(self) -> str:
        """代理的名称。团队使用该名称来唯一标识
        代理。在团队内它应该是唯一的。"""
        ...

    @property
    @abstractmethod
    def description(self) -> str:
        """代理的描述。团队使用该描述来
        决定使用哪些代理。该描述应
        说明代理的能力以及如何与之交互。"""
        ...

    @property
    @abstractmethod
    def produced_message_types(self) -> Sequence[type[BaseChatMessage]]:
        """代理在 :attr:`Response.chat_message` 字段中生成的消息类型。它们必须是 :class:`BaseChatMessage` 类型。"""
        ...



    @abstractmethod
    async def on_messages(self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken) -> Response:
        """处理传入消息并返回响应。"""
        ...




    @abstractmethod
    def on_messages_stream(
        self, messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken
    ) -> AsyncGenerator[BaseAgentEvent | BaseChatMessage | Response, None]:
        """处理传入消息并返回内部消息流，最后一项是响应。"""
        ...




    @abstractmethod
    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        """将代理重置到初始化状态。"""
        ...




    @abstractmethod
    async def on_pause(self, cancellation_token: CancellationToken) -> None:
        """当代理暂停时调用。调用此方法时，代理可能正在 :meth:`on_messages` 或
        :meth:`on_messages_stream` 中运行。"""
        ...




    @abstractmethod
    async def on_resume(self, cancellation_token: CancellationToken) -> None:
        """当代理恢复时调用。调用此方法时，代理可能正在 :meth:`on_messages` 或
        :meth:`on_messages_stream` 中运行。"""
        ...




    @abstractmethod
    async def save_state(self) -> Mapping[str, Any]:
        """保存代理状态以便后续恢复"""
        ...




    @abstractmethod
    async def load_state(self, state: Mapping[str, Any]) -> None:
        """从保存的状态恢复代理"""
        ...




    @abstractmethod
    async def close(self) -> None:
        """释放代理持有的所有资源"""
        ...