微软AutoGen介绍：Custom Agents创建自己的Agents_custom-agent

介绍

大家好，这次给大家分享的内容是微软AutoGen框架一个重要的功能模块Custom Agents(自定义智能体)。本次分享的内容需要大家具备一定的面向对象编程经验。那么本次分享的Custom Agents又是什么呢，我们直接进入正题。

Custom Agents

我们可能会遇到行为不符合预设的智能体。在这种情况下，我们可以构建自定义智能体。

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

• on_messages()：这是一个抽象方法，用于定义智能体在收到消息时的响应行为。当在run()中要求智能体提供响应时，会调用此方法。它返回一个Response对象。
• on_reset()：此抽象方法用于将智能体重置为初始状态。当要求智能体自行重置时，会调用此方法。
• produced_message_types：智能体在其响应中可能生成的ChatMessage消息类型列表。

我们也可以选择实现on_messages_stream()方法，以便在智能体生成消息时将其流式输出。如果未实现此方法，智能体将使用on_messages_stream()的默认实现，该默认实现会调用on_messages()方法，并逐一生成响应中的所有消息。

CountDownAgent

我们写个示例来演示。我们创建一个简单的智能体，它从给定数字倒数到零，并生成一系列包含当前计数的消息流。

完整代码

import asynciofrom typing import AsyncGenerator, List, Sequence, Tuplefrom autogen_agentchat.agents import BaseChatAgentfrom autogen_agentchat.base import Responsefrom autogen_agentchat.messages import AgentEvent, ChatMessage, TextMessagefrom autogen_core import CancellationTokenclass CountDownAgent(BaseChatAgent): def __init__(self, name: str, count: int = 3): super().__init__(name, \"A simple agent that counts down.\") self._count = count @property def produced_message_types(self) -> Sequence[type[ChatMessage]]: return (TextMessage,) async def on_messages(self, messages: Sequence[ChatMessage], cancellation_token: CancellationToken) -> Response: # 调用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[AgentEvent | ChatMessage | Response, None]: inner_messages: List[AgentEvent | ChatMessage] = [] for i in range(self._count, 0, -1): msg = TextMessage(content=f\"{i}...\", source=self.name) inner_messages.append(msg) yield msg # 响应会在流结束时返回。 # 它包含最终消息以及所有内部消息。 yield Response(chat_message=TextMessage(content=\"Done!\", source=self.name), inner_messages=inner_messages) async def on_reset(self, cancellation_token: CancellationToken) -> None: passasync def run_countdown_