Add documentation to RoutedAgent and message_hander, remove outdated doc content. (#485)

This commit is contained in:
Eric Zhu 2024-09-12 03:26:33 -07:00 committed by GitHub
parent 461b8a8bbd
commit cb9ae44f23
3 changed files with 56 additions and 67 deletions

View File

@ -1,66 +0,0 @@
# Using Type Routed Agent
To make it easier to implement agents that respond to certain message types there is a base class called {py:class}`~autogen_core.components.RoutedAgent`. This class provides a simple decorator pattern for associating message types with message handlers.
The decorator {py:func}`autogen_core.components.message_handler` should be added to functions in the class that are intended to handle messages. These functions have a specific signature that needs to be followed for it to be recognized as a message handler.
- The function must be an `async` function.
- The function must be decorated with the `message_handler` decorator.
- The function must have exactly 3 arguments.
- `self`
- `message`: The message to be handled, this must be type hinted with the message type that it is intended to handle.
- `cancellation_token`: A {py:class}`autogen_core.base.CancellationToken` object
- The function must be type hinted with what message types it can return.
```{tip}
Handlers can handle more than one message type by accepting a Union of the message types. It can also return more than one message type by returning a Union of the message types.
```
## Example
The following is an example of a simple agent that broadcasts the fact it received messages, and resets its internal counter when it receives a reset message.
One important thing to point out is that when an agent is constructed it must be passed a runtime object. This allows the agent to communicate with other agents via the runtime.
```python
from dataclasses import dataclass
from typing import List, Union
from autogen_core.components import RoutedAgent, message_handler, Image
from autogen_core.base import AgentRuntime, CancellationToken
@dataclass
class TextMessage:
content: str
source: str
@dataclass
class MultiModalMessage:
content: List[Union[str, Image]]
source: str
@dataclass
class Reset:
pass
class MyAgent(RoutedAgent):
def __init__(self):
super().__init__(description="I am a demo agent")
self._received_count = 0
@message_handler()
async def on_text_message(
self, message: TextMessage | MultiModalMessage, cancellation_token: CancellationToken
) -> None:
self._received_count += 1
await self.publish_message(
TextMessage(
content=f"I received a message from {message.source}. Message received #{self._received_count}",
source=self.metadata["type"],
)
)
@message_handler()
async def on_reset(self, message: Reset, cancellation_token: CancellationToken) -> None:
self._received_count = 0
```

View File

@ -50,7 +50,6 @@ from `Agent and Multi-Agent Application <core-concepts/agent-and-multi-agent-app
:caption: Cookbook :caption: Cookbook
:hidden: :hidden:
cookbook/type-routed-agent
cookbook/azure-openai-with-aad-auth cookbook/azure-openai-with-aad-auth
cookbook/termination-with-intervention cookbook/termination-with-intervention
cookbook/extracting-results-with-an-agent cookbook/extracting-results-with-an-agent

View File

@ -91,6 +91,27 @@ def message_handler(
] ]
| MessageHandler[ReceivesT, ProducesT] | MessageHandler[ReceivesT, ProducesT]
): ):
"""Decorator for message handlers.
Add this decorator to methods in a :class:`RoutedAgent` class that are intended to handle messages.
These methods must have a specific signature that needs to be followed for it to be valid:
- The method must be an `async` method.
- The method must be decorated with the `@message_handler` decorator.
- The method must have exactly 3 arguments:
1. `self`
2. `message`: The message to be handled, this must be type-hinted with the message type that it is intended to handle.
3. `ctx`: A :class:`autogen_core.base.MessageContext` object.
- The method must be type hinted with what message types it can return as a response, or it can return `None` if it does not return anything.
Handlers can handle more than one message type by accepting a Union of the message types. It can also return more than one message type by returning a Union of the message types.
Args:
func: The function to be decorated.
strict: If `True`, the handler will raise an exception if the message type or return type is not in the target types. If `False`, it will log a warning instead.
match: A function that takes the message and the context as arguments and returns a boolean. This is used for secondary routing after the message type. For handlers addressing the same message type, the match function is applied in alphabetical order of the handlers and the first matching handler will be called while the rest are skipped. If `None`, the first handler in alphabetical order matching the same message type will be called.
"""
def decorator( def decorator(
func: Callable[[Any, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]], func: Callable[[Any, ReceivesT, MessageContext], Coroutine[Any, Any, ProducesT]],
) -> MessageHandler[ReceivesT, ProducesT]: ) -> MessageHandler[ReceivesT, ProducesT]:
@ -149,6 +170,34 @@ def message_handler(
class RoutedAgent(BaseAgent): class RoutedAgent(BaseAgent):
"""A base class for agents that route messages to handlers based on the type of the message
and optional matching functions.
To create a routed agent, subclass this class and add message handlers as methods decorated with
the :func:`message_handler` decorator.
Example:
.. code-block:: python
from autogen_core.base import MessageContext
from autogen_core.components import RoutedAgent, message_handler
# Assume Message, MessageWithContent, and Response are defined elsewhere.
class MyAgent(RoutedAgent):
def __init__(self):
super().__init__("MyAgent")
@message_handler
async def handle_message(self, message: Message, ctx: MessageContext) -> Response:
return Response()
@message_handler(match=lambda message, ctx: message.content == "special")
async def handle_special_message(self, message: MessageWithContent, ctx: MessageContext) -> Response:
return Response()
"""
def __init__(self, description: str) -> None: def __init__(self, description: str) -> None:
# Self is already bound to the handlers # Self is already bound to the handlers
self._handlers: Dict[ self._handlers: Dict[
@ -172,6 +221,9 @@ class RoutedAgent(BaseAgent):
super().__init__(description) super().__init__(description)
async def on_message(self, message: Any, ctx: MessageContext) -> Any | None: async def on_message(self, message: Any, ctx: MessageContext) -> Any | None:
"""Handle a message by routing it to the appropriate message handler.
Do not override this method in subclasses. Instead, add message handlers as methods decorated with
the :func:`message_handler` decorator."""
key_type: Type[Any] = type(message) # type: ignore key_type: Type[Any] = type(message) # type: ignore
handlers = self._handlers.get(key_type) # type: ignore handlers = self._handlers.get(key_type) # type: ignore
if handlers is not None: if handlers is not None:
@ -183,11 +235,15 @@ class RoutedAgent(BaseAgent):
return await self.on_unhandled_message(message, ctx) # type: ignore return await self.on_unhandled_message(message, ctx) # type: ignore
async def on_unhandled_message(self, message: Any, ctx: MessageContext) -> None: async def on_unhandled_message(self, message: Any, ctx: MessageContext) -> None:
"""Called when a message is received that does not have a matching message handler.
The default implementation logs an info message."""
logger.info(f"Unhandled message: {message}") logger.info(f"Unhandled message: {message}")
# Deprecation warning for TypeRoutedAgent # Deprecation warning for TypeRoutedAgent
class TypeRoutedAgent(RoutedAgent): class TypeRoutedAgent(RoutedAgent):
"""Deprecated. Use :class:`RoutedAgent` instead."""
def __init__(self, description: str) -> None: def __init__(self, description: str) -> None:
warnings.warn("TypeRoutedAgent is deprecated. Use RoutedAgent instead.", DeprecationWarning, stacklevel=2) warnings.warn("TypeRoutedAgent is deprecated. Use RoutedAgent instead.", DeprecationWarning, stacklevel=2)
super().__init__(description) super().__init__(description)