Stop Conditions for specific agents only

[Leon0402]

What feature would you like to be added?

It would be nice if TextMentionTermination("...") in v4 could be defined for specific agents i.e. it should only stop if the specific agent sends / receives (tbd) the mentioned text.

Why is this needed?

More often than not the prompt contains instructions for the agent how it can stop the agent. The TextMentionTermination is currently triggered by the prompt already, which is not the intention. While a straightforward solution would be ignoring the prompt, I would suggest something more flexible, where one can specify the agent(s) that needs to send the stop condition.

[ekzhu]

Usually the prompt is set in system message so it won't show up in the message stream. Though I agree adding an source field can be useful for allowing only one agent to trigger the termination.

Happy to accept a PR for this.

[Leon0402]

Really? I heard that system message is mostly used for generic instructions. But for instance few shot examples are part of the prompt. I couldn't find official information of openai for this though. In my experiments both seems to work okay.

[ekzhu]

Depending on how you are using the agent I guess. A system message can contain instruction regarding termination -- it has been used a lot in the past in many AutoGen examples.

[Leon0402]

I will probably do a PR once my other stuff is merged, but if someone already needs it:

class TextMentionTermination(TerminationCondition):
    def __init__(self, text: str, sources: list[str] | None = None) -> None:
        self._text = text
        self._terminated = False
        self._sources = sources

    @property
    def terminated(self) -> bool:
        return self._terminated

    async def __call__(self, messages: list[AgentMessage]) -> StopMessage | None:
        if self._terminated:
            raise TerminatedException("Termination condition has already been reached")
        for message in messages:
            if self._sources is not None and message.source not in self._sources:
                continue

            if isinstance(message, TextMessage | StopMessage) and self._text in message.content:
                self._terminated = True
                return StopMessage(content=f"Text '{self._text}' mentioned", source="TextMentionTermination")
            elif isinstance(message, MultiModalMessage):
                for item in message.content:
                    if isinstance(item, str) and self._text in item:
                        self._terminated = True
                        return StopMessage(content=f"Text '{self._text}' mentioned", source="TextMentionTermination")
        return None

    async def reset(self) -> None:
        self._terminated = False