Port changes from agexplore (#47)

Author: jackgerrits

src/agnext/application/_single_threaded_agent_runtime.py

@@ -1,12 +1,16 @@
 import asyncio
+import logging
 from asyncio import Future
+from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Any, Awaitable, Dict, List, Mapping, Set
 
 from ..core import Agent, AgentRuntime, CancellationToken
 from ..core.exceptions import MessageDroppedException
 from ..core.intervention import DropMessage, InterventionHandler
 
+logger = logging.getLogger("agnext")
+
 
 @dataclass(kw_only=True)
 class PublishMessageEnvelope:
@@ -58,6 +62,14 @@ def add_agent(self, agent: Agent) -> None:
             self._per_type_subscribers[message_type].append(agent)
         self._agents.add(agent)
 
+    @property
+    def agents(self) -> Sequence[Agent]:
+        return list(self._agents)
+
+    @property
+    def unprocessed_messages(self) -> Sequence[PublishMessageEnvelope | SendMessageEnvelope | ResponseMessageEnvelope]:
+        return self._message_queue
+
     # Returns the response of the message
     def send_message(
         self,
@@ -123,6 +135,10 @@ async def _process_send(self, message_envelope: SendMessageEnvelope) -> None:
         assert recipient in self._agents
 
         try:
+            sender_name = message_envelope.sender.name if message_envelope.sender is not None else "Unknown"
+            logger.info(
+                f"Calling message handler for {recipient.name} with message type {type(message_envelope.message).__name__} from {sender_name}"
+            )
             response = await recipient.on_message(
                 message_envelope.message,
                 cancellation_token=message_envelope.cancellation_token,
@@ -145,6 +161,11 @@ async def _process_publish(self, message_envelope: PublishMessageEnvelope) -> No
         for agent in self._per_type_subscribers.get(type(message_envelope.message), []):  # type: ignore
             if message_envelope.sender is not None and agent.name == message_envelope.sender.name:
                 continue
+
+            logger.info(
+                f"Calling message handler for {agent.name} with message type {type(message_envelope.message).__name__} published by {message_envelope.sender.name if message_envelope.sender is not None else 'Unknown'}"
+            )
+
             future = agent.on_message(
                 message_envelope.message,
                 cancellation_token=message_envelope.cancellation_token,
@@ -154,12 +175,16 @@ async def _process_publish(self, message_envelope: PublishMessageEnvelope) -> No
         try:
             _all_responses = await asyncio.gather(*responses)
         except BaseException:
-            # TODO log error
+            logger.error("Error processing publish message", exc_info=True)
             return
 
         # TODO if responses are given for a publish
 
     async def _process_response(self, message_envelope: ResponseMessageEnvelope) -> None:
+        recipient_name = message_envelope.recipient.name if message_envelope.recipient is not None else "Unknown"
+        logger.info(
+            f"Resolving response for recipient {recipient_name} from {message_envelope.sender.name} with message type {type(message_envelope.message).__name__}"
+        )
         message_envelope.future.set_result(message_envelope.message)
 
     async def process_next(self) -> None:

src/agnext/chat/patterns/orchestrator_chat.py

@@ -30,6 +30,10 @@ def __init__(
         self._max_stalled_turns_before_retry = max_stalled_turns_before_retry
         self._max_retry_attempts_before_educated_guess = max_retry_attempts
 
+    @property
+    def children(self) -> Sequence[str]:
+        return [agent.name for agent in self._specialists] + [self._orchestrator.name] + [self._planner.name]
+
     @message_handler(TextMessage)
     async def on_text_message(
         self,

src/agnext/core/__init__.py

@@ -3,8 +3,9 @@
 """
 
 from ._agent import Agent
+from ._agent_props import AgentChildren
 from ._agent_runtime import AgentRuntime
 from ._base_agent import BaseAgent
 from ._cancellation_token import CancellationToken
 
-__all__ = ["Agent", "AgentRuntime", "BaseAgent", "CancellationToken"]
+__all__ = ["Agent", "AgentRuntime", "BaseAgent", "CancellationToken", "AgentChildren"]

src/agnext/core/_agent_props.py

@@ -0,0 +1,9 @@
+from typing import Protocol, Sequence, runtime_checkable
+
+
+@runtime_checkable
+class AgentChildren(Protocol):
+    @property
+    def children(self) -> Sequence[str]:
+        """Names of the children of the agent."""
+        ...

src/agnext/core/intervention.py

@@ -1,4 +1,4 @@
-from typing import Any, Awaitable, Callable, Protocol, Sequence, final
+from typing import Any, Awaitable, Callable, Protocol, final
 
 from agnext.core import Agent
 
@@ -21,9 +21,6 @@ class InterventionHandler(Protocol):
     async def on_send(self, message: Any, *, sender: Agent | None, recipient: Agent) -> Any | type[DropMessage]: ...
     async def on_publish(self, message: Any, *, sender: Agent | None) -> Any | type[DropMessage]: ...
     async def on_response(self, message: Any, *, sender: Agent, recipient: Agent | None) -> Any | type[DropMessage]: ...
-    async def on_publish_response(
-        self, message: Sequence[Any], *, recipient: Agent | None
-    ) -> Sequence[Any] | type[DropMessage]: ...
 
 
 class DefaultInterventionHandler(InterventionHandler):
@@ -35,8 +32,3 @@ async def on_publish(self, message: Any, *, sender: Agent | None) -> Any | type[
 
     async def on_response(self, message: Any, *, sender: Agent, recipient: Agent | None) -> Any | type[DropMessage]:
         return message
-
-    async def on_publish_response(
-        self, message: Sequence[Any], *, recipient: Agent | None
-    ) -> Sequence[Any] | type[DropMessage]:
-        return message