Add group chat pattern, create separate folder for patterns (#117)

* add tool use example; refactor example directory

* update

* add more examples

* fix

* fix

* doc

* move

* add group chat example, create patterns folder

python/examples/README.md

@@ -9,11 +9,7 @@ agents, runtime, and message passing APIs.
 
 - [`one_agent_direct.py`](core/one_agent_direct.py): A simple example of how to create a single agent powered by ChatCompletion model client. Communicate with the agent using async direct messaging API.
 - [`inner_outer_direct.py`](core/inner_outer_direct.py): A simple example of how to create an agent that calls an inner agent using async direct messaging API.
-- [`two_agents_pub_sub.py`](core/two_agents_pub_sub.py): An example of how to create two agents that communicate using publish-subscribe API.
-- [`mixture_of_agents_direct.py`](core/mixture_of_agents_direct.py): An example of how to create a [mixture of agents](https://github.com/togethercomputer/moa) that communicate using async direct messaging API.
-- [`mixture_of_agents_pub_sub.py`](core/mixture_of_agents_pub_sub.py): An example of how to create a [mixture of agents](https://github.com/togethercomputer/moa) that communicate using publish-subscribe API.
-- [`coder_reviewer_direct.py`](core/coder_reviewer_direct.py): An example of how to create a coder-reviewer reflection pattern using async direct messaging API.
-- [`coder_reviewer_pub_sub.py`](core/coder_reviewer_pub_sub.py): An example of how to create a coder-reviewer reflection pattern using publish-subscribe API.
+- [`two_agents_pub_sub_termination.py`](core/two_agents_pub_sub_termination.py): An example of how to create two agents that communicate using publish-subscribe API, and termination using an intervention handler.
 
 ## Tool use examples
 
@@ -24,6 +20,16 @@ We provide examples to illustrate how to use tools in AGNext:
 - [`coding_two_agent_pub_sub.py`](tool-use/coding_two_agent_pub_sub.py): a code execution example with two agents, one for calling tool and one for executing the tool, to demonstrate tool use and reflection on tool use. This example uses the publish-subscribe API.
 - [`custom_function_tool_one_agent_direct.py`](tool-use/custom_function_tool_one_agent_direct.py): a custom function tool example with one agent that calls and executes tools to demonstrate tool use and reflection on tool use. This example uses the async direct messaging API.
 
+## Pattern examples
+
+We provide examples to illustrate how multi-agent patterns can be implemented in AGNext:
+
+- [`mixture_of_agents_direct.py`](pattern/mixture_of_agents_direct.py): An example of how to create a [mixture of agents](https://github.com/togethercomputer/moa) that communicate using async direct messaging API.
+- [`mixture_of_agents_pub_sub.py`](pattern/mixture_of_agents_pub_sub.py): An example of how to create a [mixture of agents](https://github.com/togethercomputer/moa) that communicate using publish-subscribe API.
+- [`coder_reviewer_direct.py`](pattern/coder_reviewer_direct.py): An example of how to create a coder-reviewer reflection pattern using async direct messaging API.
+- [`coder_reviewer_pub_sub.py`](pattern/coder_reviewer_pub_sub.py): An example of how to create a coder-reviewer reflection pattern using publish-subscribe API.
+- [`group_chat_pub_sub.py`](pattern/group_chat_pub_sub.py): An example of how to create a round-robin group chat among three agents using publish-subscribe API.
+
 ## Demos
 
 We provide interactive demos that showcase applications that can be built using AGNext:

python/examples/core/two_agents_pub_sub.py → python/examples/core/two_agents_pub_sub_termination.py

@@ -1,6 +1,6 @@
 import asyncio
 from dataclasses import dataclass
-from typing import List
+from typing import Any, List
 
 from agnext.application import SingleThreadedAgentRuntime
 from agnext.components import TypeRoutedAgent, message_handler
@@ -12,7 +12,8 @@
     SystemMessage,
     UserMessage,
 )
-from agnext.core import CancellationToken
+from agnext.core import AgentId, CancellationToken
+from agnext.core.intervention import DefaultInterventionHandler
 
 
 @dataclass
@@ -21,10 +22,15 @@ class Message:
     content: str
 
 
+@dataclass
+class Termination:
+    pass
+
+
 class ChatCompletionAgent(TypeRoutedAgent):
     """An agent that uses a chat completion model to respond to messages.
     It keeps a memory of the conversation and uses it to generate responses.
-    It terminates the conversation when the termination word is mentioned."""
+    It publishes a termination message when the termination word is mentioned."""
 
     def __init__(
         self,
@@ -43,6 +49,7 @@ def __init__(
     async def handle_message(self, message: Message, cancellation_token: CancellationToken) -> None:
         self._memory.append(message)
         if self._termination_word in message.content:
+            self.publish_message(Termination())
             return
         llm_messages: List[LLMMessage] = []
         for m in self._memory[-10:]:
@@ -55,8 +62,30 @@ async def handle_message(self, message: Message, cancellation_token: Cancellatio
         self.publish_message(Message(content=response.content, source=self.metadata["name"]))
 
 
+class TerminationHandler(DefaultInterventionHandler):
+    """A handler that listens for termination messages."""
+
+    def __init__(self) -> None:
+        self._terminated = False
+
+    async def on_publish(self, message: Any, *, sender: AgentId | None) -> Any:
+        if isinstance(message, Termination):
+            self._terminated = True
+        return message
+
+    @property
+    def terminated(self) -> bool:
+        return self._terminated
+
+
 async def main() -> None:
-    runtime = SingleThreadedAgentRuntime()
+    # Create the termination handler.
+    termination_handler = TerminationHandler()
+
+    # Create the runtime with the termination handler.
+    runtime = SingleThreadedAgentRuntime(intervention_handler=termination_handler)
+
+    # Register the agents.
     jack = runtime.register_and_get(
         "Jack",
         lambda: ChatCompletionAgent(
@@ -84,8 +113,8 @@ async def main() -> None:
     message = Message(content="Can you tell me something fun about SF?", source="User")
     runtime.send_message(message, jack)
 
-    # Process messages until the agent responds.
-    while True:
+    # Process messages until termination.
+    while not termination_handler.terminated:
         await runtime.process_next()
 
 

python/examples/patterns/group_chat_pub_sub.py

@@ -0,0 +1,166 @@
+import asyncio
+from dataclasses import dataclass
+from typing import Any, List
+
+from agnext.application import SingleThreadedAgentRuntime
+from agnext.components import TypeRoutedAgent, message_handler
+from agnext.components.models import (
+    AssistantMessage,
+    ChatCompletionClient,
+    LLMMessage,
+    OpenAI,
+    SystemMessage,
+    UserMessage,
+)
+from agnext.core import AgentId, CancellationToken
+from agnext.core.intervention import DefaultInterventionHandler
+
+
+@dataclass
+class Message:
+    source: str
+    content: str
+
+
+@dataclass
+class RequestToSpeak:
+    pass
+
+
+@dataclass
+class Termination:
+    pass
+
+
+class RoundRobinGroupChatManager(TypeRoutedAgent):
+    def __init__(
+        self,
+        description: str,
+        participants: List[AgentId],
+        num_rounds: int,
+    ) -> None:
+        super().__init__(description)
+        self._participants = participants
+        self._num_rounds = num_rounds
+        self._round_count = 0
+
+    @message_handler
+    async def handle_message(self, message: Message, cancellation_token: CancellationToken) -> None:
+        # Select the next speaker in a round-robin fashion
+        speaker = self._participants[self._round_count % len(self._participants)]
+        self._round_count += 1
+        if self._round_count == self._num_rounds * len(self._participants):
+            # End the conversation after the specified number of rounds.
+            self.publish_message(Termination())
+            return
+        # Send a request to speak message to the selected speaker.
+        self.send_message(RequestToSpeak(), speaker)
+
+
+class GroupChatParticipant(TypeRoutedAgent):
+    def __init__(
+        self,
+        description: str,
+        system_messages: List[SystemMessage],
+        model_client: ChatCompletionClient,
+    ) -> None:
+        super().__init__(description)
+        self._system_messages = system_messages
+        self._model_client = model_client
+        self._memory: List[Message] = []
+
+    @message_handler
+    async def handle_message(self, message: Message, cancellation_token: CancellationToken) -> None:
+        self._memory.append(message)
+
+    @message_handler
+    async def handle_request_to_speak(self, message: RequestToSpeak, cancellation_token: CancellationToken) -> None:
+        # Generate a response to the last message in the memory
+        if not self._memory:
+            return
+        llm_messages: List[LLMMessage] = []
+        for m in self._memory[-10:]:
+            if m.source == self.metadata["name"]:
+                llm_messages.append(AssistantMessage(content=m.content, source=self.metadata["name"]))
+            else:
+                llm_messages.append(UserMessage(content=m.content, source=m.source))
+        response = await self._model_client.create(self._system_messages + llm_messages)
+        assert isinstance(response.content, str)
+        speach = Message(content=response.content, source=self.metadata["name"])
+        self._memory.append(speach)
+        self.publish_message(speach)
+
+
+class TerminationHandler(DefaultInterventionHandler):
+    """A handler that listens for termination messages."""
+
+    def __init__(self) -> None:
+        self._terminated = False
+
+    async def on_publish(self, message: Any, *, sender: AgentId | None) -> Any:
+        if isinstance(message, Termination):
+            self._terminated = True
+        return message
+
+    @property
+    def terminated(self) -> bool:
+        return self._terminated
+
+
+async def main() -> None:
+    # Create the termination handler.
+    termination_handler = TerminationHandler()
+
+    # Create the runtime.
+    runtime = SingleThreadedAgentRuntime(intervention_handler=termination_handler)
+
+    # Register the participants.
+    agent1 = runtime.register_and_get(
+        "DataScientist",
+        lambda: GroupChatParticipant(
+            description="A data scientist",
+            system_messages=[SystemMessage("You are a data scientist.")],
+            model_client=OpenAI(model="gpt-3.5-turbo"),
+        ),
+    )
+    agent2 = runtime.register_and_get(
+        "Engineer",
+        lambda: GroupChatParticipant(
+            description="An engineer",
+            system_messages=[SystemMessage("You are an engineer.")],
+            model_client=OpenAI(model="gpt-3.5-turbo"),
+        ),
+    )
+    agent3 = runtime.register_and_get(
+        "Artist",
+        lambda: GroupChatParticipant(
+            description="An artist",
+            system_messages=[SystemMessage("You are an artist.")],
+            model_client=OpenAI(model="gpt-3.5-turbo"),
+        ),
+    )
+
+    # Register the group chat manager.
+    runtime.register(
+        "GroupChatManager",
+        lambda: RoundRobinGroupChatManager(
+            description="A group chat manager",
+            participants=[agent1, agent2, agent3],
+            num_rounds=3,
+        ),
+    )
+
+    # Start the conversation.
+    runtime.publish_message(Message(content="Hello, everyone!", source="Moderator"), namespace="default")
+
+    # Run the runtime until termination.
+    while not termination_handler.terminated:
+        await runtime.process_next()
+
+
+if __name__ == "__main__":
+    import logging
+
+    logging.basicConfig(level=logging.WARNING)
+    logging.getLogger("agnext").setLevel(logging.DEBUG)
+    asyncio.run(main())