How to control termination conditions in AutoGen

Multi-agent loops fail silently when they run one turn too long or stop before the task is done. Controlling AutoGen termination conditions is the difference between a reproducible pipeline and a token-burning roulette wheel. The framework gives you both built-in guards and escape hatches for custom logic; use them deliberately or you will debug hangs at 2 a.m.

Step 1: Install and import the AgentChat API

AutoGen 0.4 reorganized the package into autogen-core, autogen-agentchat, and autogen-ext. The high-level team and condition classes live in autogen-agentchat. Install the agent chat layer and the OpenAI model adapter:

pip install autogen-agentchat autogen-ext openai

Import the pieces needed for a round-robin team:

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

Create a model client. If you point it at an OpenAI-compatible gateway such as n4n.ai, its automatic fallback when a provider is degraded prevents transient 429s from masquerading as task completion—your termination condition sees a clean response, not a dropped connection.

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini", api_key="YOUR_KEY")

Step 2: Apply a built-in MaxMessageTermination

The simplest AutoGen termination conditions are hard caps. MaxMessageTermination stops the team after N agent messages regardless of content. This is your circuit breaker against runaway loops.

termination = MaxMessageTermination(max_messages=5)

agent = AssistantAgent("assistant", model_client=model_client)
team = RoundRobinGroupChat([agent], termination_condition=termination)

async def run():
    result = await team.run_stream(task="Count to 10.")
    async for msg in result:
        print(msg)

The team emits at most five assistant messages then halts. The count includes all agent outputs, not just the initiating agent. Use this when you need a predictable upper bound for cost or latency in a batch job. Without it, a vague prompt can spin indefinitely.

One caveat: MaxMessageTermination is evaluated after each message is added to the transcript. If you stream, the termination check still happens post-message, so you won’t get a partial message cut off mid-token.

Step 3: Stop on a keyword with TextMentionTermination

Often you want an agent to signal completion by emitting a sentinel string. TextMentionTermination watches every message for that substring and terminates on match.

termination = TextMentionTermination("TERMINATE")

agent = AssistantAgent(
    "assistant",
    model_client=model_client,
    system_message="When done, reply exactly: TASK COMPLETE. TERMINATE",
)
team = RoundRobinGroupChat([agent], termination_condition=termination)

If the model complies, the loop ends early even if the message cap is higher. The match is case-sensitive and operates on the raw string content. For multi-agent setups, any agent’s message containing the text triggers termination—there is no source filtering built in.

In practice, models miss exact sentinels. Couple this with MaxMessageTermination via a custom wrapper (next step) so a stubborn model can’t loop forever when it forgets to say the keyword.

Step 4: Combine conditions with a custom wrapper

The RoundRobinGroupChat constructor accepts exactly one termination condition. To enforce both a keyword and a ceiling, wrap them in a composite. The base class TerminationCondition defines a synchronous should_terminate(self, messages) method, so composition is straightforward.

from autogen_agentchat.base import TerminationCondition
from typing import Sequence
from autogen_agentchat.messages import AgentMessage

class AndTermination(TerminationCondition):
    def __init__(self, *conditions: TerminationCondition):
        self.conditions = conditions

    def should_terminate(self, messages: Sequence[AgentMessage]) -> bool:
        return all(c.should_terminate(messages) for c in self.conditions)

combined = AndTermination(
    MaxMessageTermination(max_messages=10),
    TextMentionTermination("TERMINATE"),
)

Now the team stops at message 10 or when the keyword appears, whichever comes first. Swap all for any to get OR semantics—terminate if either condition is met. This pattern is the cleanest way to layer AutoGen termination conditions without rewriting the built-ins.

Step 5: Write a domain-specific custom termination condition

Built-ins miss structured outputs. Suppose your agent returns JSON and you want to stop when "status":"complete" appears. Subclass TerminationCondition and inspect content directly:

import json

class JsonStatusTermination(TerminationCondition):
    def should_terminate(self, messages: Sequence[AgentMessage]) -> bool:
        for msg in messages:
            content = getattr(msg, "content", "")
            if not isinstance(content, str):
                continue
            try:
                data = json.loads(content)
            except json.JSONDecodeError:
                continue
            if isinstance(data, dict) and data.get("status") == "complete":
                return True
        return False

Attach it to the team exactly like the built-ins:

team = RoundRobinGroupChat([agent], termination_condition=JsonStatusTermination())

This pattern is how you make AutoGen termination conditions match your actual success criteria instead of guessing from prose. You can also track running totals (e.g., sum of extracted numbers) by storing state on the condition instance, but keep the method cheap—it runs every turn.

Step 6: Control termination in legacy GroupChat (v0.2)

Plenty of production code still uses the older autogen package. There, termination is set on GroupChat via max_round and a termination_msg lambda:

from autogen import GroupChat, GroupChatManager, AssistantAgent, UserProxyAgent

assistant = AssistantAgent("assistant", llm_config={"model": "gpt-4o-mini"})
user = UserProxyAgent("user", human_input_mode="NEVER")

groupchat = GroupChat(
    agents=[assistant, user],
    messages=[],
    max_round=8,
    termination_msg=lambda x: "APPROVE" in x.get("content", ""),
)
manager = GroupChatManager(groupchat, llm_config={"model": "gpt-4o-mini"})

user.initiate_chat(manager, message="Generate a report.")

max_round is the message cap; termination_msg receives the last message dict and returns True to stop. The lambda can inspect any field. This is the precursor to the v0.4 condition classes—same intent, less composable, and harder to unit test because the logic is inline.

If you maintain this code, extract the lambda into a named function so you can test it against sample message dicts before deploying.

Step 7: Verify your termination logic

Don’t trust a condition you haven’t tested. The should_terminate method is pure given a message list, so unit test it directly with constructed messages:

from autogen_agentchat.messages import AssistantMessage

def test_json_status_termination():
    cond = JsonStatusTermination()
    msgs = [AssistantMessage(source="assistant", content='{"status":"complete"}')]
    assert cond.should_terminate(msgs) is True

    msgs_fail = [AssistantMessage(source="assistant", content='{"status":"running"}')]
    assert cond.should_terminate(msgs_fail) is False

Run with pytest. For integration confidence, stream the team run and assert the message count never exceeds your cap:

async def test_team_respects_cap():
    team = RoundRobinGroupChat([agent], termination_condition=MaxMessageTermination(3))
    count = 0
    async for _ in team.run_stream(task="Loop"):
        count += 1
    assert count <= 3

If these pass, your AutoGen termination conditions are enforcing the bounds you designed. Add a log line inside should_terminate during local runs to see which condition fired—this saves time when a team stops unexpectedly.

Step 8: Operational notes

Termination conditions are evaluated after every message, so keep should_terminate O(n) or better. Parsing huge transcripts on each turn adds latency that compounds in long sessions. For stateful checks, persist a counter or flag on the condition object rather than re-scanning all messages.

When you swap models behind a gateway, the termination logic stays identical—only the model client changes. That separation is why investing in precise AutoGen termination conditions pays off across model upgrades and provider outages. Set a cap always, add a semantic trigger when you can, and test the boundary.