自定义代理#
您可能会有一些行为不属于预设的代理。 在这种情况下,您可以构建自定义代理。
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