自定义代理#

您可能会有一些行为不属于预设的代理。在这种情况下，您可以构建自定义代理。

AgentChat 中的所有代理都继承自 BaseChatAgent 类，并实现以下抽象方法和属性：

on_messages(): 定义代理响应消息的行为的抽象方法。当代理被要求在 run() 中提供响应时，会调用此方法。它返回一个 Response 对象。
on_reset(): 将代理重置为初始状态的抽象方法。当代理被要求重置自身时，会调用此方法。
produced_message_types: 代理在其响应中可以产生的可能的 ChatMessage 消息类型列表。

可选地，您可以实现 on_messages_stream() 方法来流式传输代理生成的消息。如果未实现此方法，代理将使用 on_messages_stream() 的默认实现，该实现调用 on_messages() 方法并生成响应中的所有消息。

倒计时代理#

在这个例子中，我们创建一个简单的代理，它从给定的数字倒数到零，并生成一个包含当前计数的消息流。

from typing import AsyncGenerator, List, Sequence

from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import AgentMessage, ChatMessage, TextMessage
from autogen_core.base import CancellationToken


class CountDownAgent(BaseChatAgent):
    def __init__(self, name: str, count: int = 10):
        super().__init__(name, "A simple agent that counts down.")
        self._count = count

    @property
    def produced_message_types(self) -> List[type[ChatMessage]]:
        return [TextMessage]

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        # Calls the on_messages_stream.
        response: Response | None = None
        async for message in self.on_messages_stream(messages, cancellation_token):
            if isinstance(message, Response):
                response = message
        assert response is not None
        return response

    async def on_messages_stream(
        self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken
    ) -> AsyncGenerator[AgentMessage | Response, None]:
        inner_messages: List[AgentMessage] = []
        for i in range(self._count, 0, -1):
            msg = TextMessage(content=f"{i}...", source=self.name)
            inner_messages.append(msg)
            yield msg
        # The response is returned at the end of the stream.
        # It contains the final message and all the inner messages.
        yield Response(chat_message=TextMessage(content="Done!", source=self.name), inner_messages=inner_messages)

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass


async def run_countdown_agent() -> None:
    # Create a countdown agent.
    countdown_agent = CountDownAgent("countdown")

    # Run the agent with a given task and stream the response.
    async for message in countdown_agent.on_messages_stream([], CancellationToken()):
        if isinstance(message, Response):
            print(message.chat_message.content)
        else:
            print(message.content)


# Use asyncio.run(run_countdown_agent()) when running in a script.
await run_countdown_agent()

10...
9...
8...
7...
6...
5...
4...
3...
2...
1...
Done!

UserProxyAgent#

构建自定义代理的一个常见用例是创建一个作为用户代理的Agent。

在下面的例子中，我们展示了如何实现一个 UserProxyAgent - 这是一个通过控制台要求用户输入文本，然后将该消息作为响应返回的代理。

import asyncio
from typing import List, Sequence

from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import ChatMessage
from autogen_core.base import CancellationToken


class UserProxyAgent(BaseChatAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name, "A human user.")

    @property
    def produced_message_types(self) -> List[type[ChatMessage]]:
        return [TextMessage]

    async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response:
        user_input = await asyncio.get_event_loop().run_in_executor(None, input, "Enter your response: ")
        return Response(chat_message=TextMessage(content=user_input, source=self.name))

    async def on_reset(self, cancellation_token: CancellationToken) -> None:
        pass


async def run_user_proxy_agent() -> None:
    user_proxy_agent = UserProxyAgent(name="user_proxy_agent")
    response = await user_proxy_agent.on_messages([], CancellationToken())
    print(response.chat_message.content)


# Use asyncio.run(run_user_proxy_agent()) when running in a script.
await run_user_proxy_agent()

hi