Agent name termination

python/packages/autogen-agentchat/src/autogen_agentchat/task/__init__.py

@@ -6,6 +6,7 @@
     TextMentionTermination,
     TimeoutTermination,
     TokenUsageTermination,
+    SourceMatchTermination
 )
 
 __all__ = [
@@ -15,5 +16,6 @@
     "TokenUsageTermination",
     "HandoffTermination",
     "TimeoutTermination",
+    "SourceMatchTermination",
     "Console",
 ]

python/packages/autogen-agentchat/src/autogen_agentchat/task/_terminations.py

@@ -1,5 +1,5 @@
 import time
-from typing import Sequence
+from typing import Sequence, List
 
 from ..base import TerminatedException, TerminationCondition
 from ..messages import AgentMessage, HandoffMessage, MultiModalMessage, StopMessage, TextMessage
@@ -208,3 +208,36 @@ async def __call__(self, messages: Sequence[AgentMessage]) -> StopMessage | None
     async def reset(self) -> None:
         self._start_time = time.monotonic()
         self._terminated = False
+
+
+class SourceMatchTermination(TerminationCondition):
+    """Terminate the conversation after a specific source responds.
+
+    Args:
+        sources (List[str]): List of source names to terminate the conversation.
+
+    Raises:
+        TerminatedException: If the termination condition has already been reached.
+    """
+
+    def __init__(self, sources: List[str]) -> None:
+        self._sources = sources
+        self._terminated = False
+
+    @property
+    def terminated(self) -> bool:
+        return self._terminated
+
+    async def __call__(self, messages: Sequence[AgentMessage]) -> StopMessage | None:
+        if self._terminated:
+            raise TerminatedException("Termination condition has already been reached")
+        if not messages:
+            return None
+        for message in messages:
+            if message.source in self._sources:
+                self._terminated = True
+                return StopMessage(content=f"'{message.source}' answered", source="SourceMatchTermination")
+        return None
+
+    async def reset(self) -> None:
+        self._terminated = False

python/packages/autogen-agentchat/tests/test_termination_condition.py

@@ -1,6 +1,7 @@
 import asyncio
 
 import pytest
+from autogen_agentchat.base import TerminatedException
 from autogen_agentchat.messages import HandoffMessage, StopMessage, TextMessage
 from autogen_agentchat.task import (
     HandoffTermination,
@@ -9,6 +10,7 @@
     TextMentionTermination,
     TimeoutTermination,
     TokenUsageTermination,
+    SourceMatchTermination,
 )
 from autogen_core.components.models import RequestUsage
 
@@ -226,3 +228,28 @@ async def test_timeout_termination() -> None:
     assert await termination([TextMessage(content="Hello", source="user")]) is None
     await asyncio.sleep(0.2)
     assert await termination([TextMessage(content="World", source="user")]) is not None
+
+
+@pytest.mark.asyncio
+async def test_agent_name_termination() -> None:
+    termination = SourceMatchTermination(sources=["Assistant"])
+    assert await termination([]) is None
+
+    continue_messages = [
+        TextMessage(content="Hello", source="agent"),
+        TextMessage(content="Hello", source="user")
+    ]
+    assert await termination(continue_messages) is None
+
+    terminate_messages = [
+        TextMessage(content="Hello", source="agent"),
+        TextMessage(content="Hello", source="Assistant"),
+        TextMessage(content="Hello", source="user")
+    ]
+    result = await termination(terminate_messages)
+    assert isinstance(result, StopMessage)
+    assert termination.terminated
+    with pytest.raises(TerminatedException):
+        await termination([])
+    await termination.reset()
+    assert not termination.terminated