Events for handoffs

autogen/agentchat/group/events/handoff_events.py

@@ -0,0 +1,86 @@
+# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Callable
+from typing import Any
+
+from autogen.agentchat.agent import Agent
+from autogen.agentchat.group.targets.transition_target import TransitionTarget
+from autogen.events.base_event import BaseEvent, wrap_event
+from autogen.formatting_utils import colored
+
+
+@wrap_event
+class AfterWorksTransitionEvent(BaseEvent):
+    """Event for after works handoffs"""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    source_agent: Agent
+    transition_target: TransitionTarget
+
+    def __init__(self, source_agent: Agent, transition_target: TransitionTarget):
+        super().__init__(source_agent=source_agent, transition_target=transition_target)
+
+    def print(self, f: Callable[..., Any] | None = None) -> None:
+        f = f or print
+        super().print(f)
+
+        f(
+            colored(
+                f"***** AfterWork handoff ({self.source_agent.name if hasattr(self.source_agent, 'name') else self.source_agent}): {self.transition_target.display_name()} *****",
+                "blue",
+            ),
+            flush=True,
+        )
+
+
+@wrap_event
+class OnContextConditionTransitionEvent(BaseEvent):
+    """Event for OnContextCondition handoffs"""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    source_agent: Agent
+    transition_target: TransitionTarget
+
+    def __init__(self, source_agent: Agent, transition_target: TransitionTarget):
+        super().__init__(source_agent=source_agent, transition_target=transition_target)
+
+    def print(self, f: Callable[..., Any] | None = None) -> None:
+        f = f or print
+        super().print(f)
+
+        f(
+            colored(
+                f"***** OnContextCondition handoff ({self.source_agent.name if hasattr(self.source_agent, 'name') else self.source_agent}): {self.transition_target.display_name()} *****",
+                "blue",
+            ),
+            flush=True,
+        )
+
+
+@wrap_event
+class OnConditionLLMTransitionEvent(BaseEvent):
+    """Event for LLM-based OnCondition handoffs"""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    source_agent: Agent
+    transition_target: TransitionTarget
+
+    def __init__(self, source_agent: Agent, transition_target: TransitionTarget):
+        super().__init__(source_agent=source_agent, transition_target=transition_target)
+
+    def print(self, f: Callable[..., Any] | None = None) -> None:
+        f = f or print
+        super().print(f)
+
+        f(
+            colored(
+                f"***** LLM-based OnCondition handoff ({self.source_agent.name if hasattr(self.source_agent, 'name') else self.source_agent}): {self.transition_target.display_name()} *****",
+                "blue",
+            ),
+            flush=True,
+        )

autogen/agentchat/group/events/reply_result_events.py

@@ -0,0 +1,36 @@
+# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Callable
+from typing import Any
+
+from autogen.agentchat.agent import Agent
+from autogen.agentchat.group.targets.transition_target import TransitionTarget
+from autogen.events.base_event import BaseEvent, wrap_event
+from autogen.formatting_utils import colored
+
+
+@wrap_event
+class ReplyResultTransitionEvent(BaseEvent):
+    """Event for reply result transitions"""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    source_agent: Agent
+    transition_target: TransitionTarget
+
+    def __init__(self, source_agent: Agent, transition_target: TransitionTarget):
+        super().__init__(source_agent=source_agent, transition_target=transition_target)
+
+    def print(self, f: Callable[..., Any] | None = None) -> None:
+        f = f or print
+        super().print(f)
+
+        f(
+            colored(
+                f"***** ReplyResult transition ({self.source_agent.name if hasattr(self.source_agent, 'name') else self.source_agent}): {self.transition_target.display_name()} *****",
+                "blue",
+            ),
+            flush=True,
+        )

autogen/agentchat/group/group_tool_executor.py

@@ -7,6 +7,10 @@
 from copy import deepcopy
 from typing import Annotated, Any
 
+from autogen.agentchat.group.events.handoff_events import OnConditionLLMTransitionEvent
+from autogen.agentchat.group.events.reply_result_events import ReplyResultTransitionEvent
+from autogen.io.base import IOStream
+
 from ...oai import OpenAIWrapper
 from ...tools import Depends, Tool
 from ...tools.dependency_injection import inject_params, on
@@ -32,6 +36,8 @@ def __init__(self) -> None:
 
         # Store the next target from a tool call
         self._group_next_target: TransitionTarget | None = None
+        # Group manager will be set by link_agents_to_group_manager
+        self._group_manager: Any = None
 
         # Primary tool reply function for handling the tool reply and the ReplyResult and TransitionTarget returns
         self.register_reply([Agent, None], self._generate_group_tool_reply, remove_other_reply_funcs=True)
@@ -126,6 +132,90 @@ def register_agents_functions(self, agents: list[ConversableAgent], context_vari
             for tool in agent.tools:
                 self.register_for_execution(serialize=False, silent_override=True)(tool)
 
+    def function_is_agent_llm_handoff(self, agent_name: str, function_name: str) -> bool:
+        """Determines if a function name is an LLM handoff.
+
+        Args:
+            agent_name (str): The name of the agent the conditions against
+            function_name (str): The function name to check.
+
+        Returns:
+            bool: True if the function is an LLM handoff, False otherwise.
+        """
+        agent = self._group_manager.groupchat.agent_by_name(agent_name)
+        if agent is None:
+            return False
+        # Check if agent is a ConversableAgent with handoffs
+        if not isinstance(agent, ConversableAgent) or not hasattr(agent, "handoffs"):
+            return False
+        return any(on_condition.llm_function_name == function_name for on_condition in agent.handoffs.llm_conditions)
+
+    def get_sender_agent_for_message(self, message: dict[str, Any]) -> Agent | None:
+        """Gets the sender agent from the message.
+
+        Args:
+            message: The message containing the tool call and source agent
+
+        Returns:
+            The sender agent, or None if not found
+        """
+        if "name" in message and self._group_manager:
+            agent = self._group_manager.groupchat.agent_by_name(message.get("name"))
+            return agent  # type: ignore[no-any-return]
+        return None
+
+    def is_handoff_function(self, message: dict[str, Any]) -> bool:
+        """Checks if the tool call is a handoff function.
+
+        Args:
+            message: The message containing the tool call and source agent
+        """
+        if "name" in message:
+            agent_name = message.get("name")
+
+            if agent_name and "tool_calls" in message and self._group_manager:
+                for tool_call in message["tool_calls"]:
+                    if "function" in tool_call and "name" in tool_call["function"]:
+                        function_name = tool_call["function"]["name"]
+                        if self.function_is_agent_llm_handoff(agent_name, function_name):
+                            return True
+        return False
+
+    def _send_llm_handoff_event(self, message: dict[str, Any], transition_target: TransitionTarget) -> None:
+        """Send an LLM OnCondition handoff event.
+
+        Args:
+            message: The message containing the tool call and source agent
+            transition_target: The target to transition to
+        """
+        if self.is_handoff_function(message):
+            source_agent = self.get_sender_agent_for_message(message)
+            if source_agent:
+                iostream = IOStream.get_default()
+                iostream.send(
+                    OnConditionLLMTransitionEvent(
+                        source_agent=source_agent,
+                        transition_target=transition_target,
+                    )
+                )
+
+    def _send_reply_result_handoff_event(self, message: dict[str, Any], transition_target: TransitionTarget) -> None:
+        """Send a ReplyResult handoff event.
+
+        Args:
+            message: The message containing the tool call and source agent
+            transition_target: The target to transition to
+        """
+        source_agent = self.get_sender_agent_for_message(message)
+        if source_agent:
+            iostream = IOStream.get_default()
+            iostream.send(
+                ReplyResultTransitionEvent(
+                    source_agent=source_agent,
+                    transition_target=transition_target,
+                )
+            )
+
     def _generate_group_tool_reply(
         self,
         agent: ConversableAgent,
@@ -172,13 +262,15 @@ def _generate_group_tool_reply(
                 for tool_response in tool_message["tool_responses"]:
                     content = tool_response.get("content")
 
-                    # Tool Call returns that are a target are either a ReplyResult or a TransitionTarget are the next agent
+                    # Tool Call returns that are a target are either a ReplyResult or a TransitionTarget
                     if isinstance(content, ReplyResult):
                         if content.context_variables and content.context_variables.to_dict() != {}:
                             agent.context_variables.update(content.context_variables.to_dict())
                         if content.target is not None:
+                            self._send_reply_result_handoff_event(message_copy, content.target)
                             next_target = content.target
                     elif isinstance(content, TransitionTarget):
+                        self._send_llm_handoff_event(message_copy, content)
                         next_target = content
 
                     # Serialize the content to a string

autogen/agentchat/group/group_utils.py

@@ -8,6 +8,9 @@
 from types import MethodType
 from typing import TYPE_CHECKING, Any, Optional
 
+from autogen.agentchat.group.events.handoff_events import AfterWorksTransitionEvent, OnContextConditionTransitionEvent
+from autogen.io.base import IOStream
+
 from ..agent import Agent
 from ..groupchat import GroupChat, GroupChatManager
 from .context_variables import ContextVariables
@@ -114,12 +117,17 @@ def _evaluate_after_works_conditions(
             after_work_condition.condition is None or after_work_condition.condition.evaluate(agent.context_variables)
         ):
             # Condition matched, resolve and return
-            return after_work_condition.target.resolve(
+            after_works_speaker = after_work_condition.target.resolve(
                 groupchat,
                 agent,
                 user_agent,
             ).get_speaker_selection_result(groupchat)
 
+            iostream = IOStream.get_default()
+            iostream.send(AfterWorksTransitionEvent(source_agent=agent, transition_target=after_work_condition.target))
+
+            return after_works_speaker
+
     return None
 
 
@@ -142,6 +150,9 @@ def _run_oncontextconditions(
 
             transfer_name = on_condition.target.display_name()
 
+            iostream = IOStream.get_default()
+            iostream.send(OnContextConditionTransitionEvent(source_agent=agent, transition_target=on_condition.target))
+
             return True, "[Handing off to " + transfer_name + "]"
 
     return False, None