Building reliable multi-agent systems means knowing when to stop. This autogen agent team termination condition tutorial walks through the exact hooks AutoGen exposes to halt agent chatter before it burns tokens or loops forever. We’ll stand up a small researcher/writer team, then layer termination conditions from naive to production-grade.
Prerequisites
- Python 3.10 or newer
autogen-agentchat,autogen-ext, andopenaipackages installed (pip install autogen-agentchat autogen-ext openai)- A key for an OpenAI-compatible endpoint. The snippets point at
https://api.n4n.ai/v1, which fronts 240+ models and handles fallback, but any base URL works. - Familiarity with
asyncioand basic AutoGen agent construction.
Model client setup
AutoGen’s OpenAIChatCompletionClient speaks the standard OpenAI shape. Point it at your gateway:
from autogen_ext.models.openai import OpenAIChatCompletionClient
model_client = OpenAIChatCompletionClient(
model="openai/gpt-4o-mini",
api_key="YOUR_KEY",
base_url="https://api.n4n.ai/v1",
)If you run a local model or another vendor, swap base_url and model. Keep the client reusable; agents are cheap to construct but the client holds connection pools.
A team with a hard message cap
The simplest termination is MaxMessageTermination. It counts agent messages and stops the team after N. No semantic checking—just a circuit breaker.
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination
researcher = AssistantAgent(
"researcher",
model_client=model_client,
system_message="You gather concise facts. Reply in 1-2 sentences.",
)
writer = AssistantAgent(
"writer",
model_client=model_client,
system_message="You turn facts into a single tight paragraph.",
)
team = RoundRobinGroupChat(
[researcher, writer],
termination_condition=MaxMessageTermination(max_messages=4),
)
import asyncio
async def run():
async for msg in team.run_stream(task="Summarize why termination conditions matter."):
print(f"{msg.source}: {msg.content}")
asyncio.run(run())Expected output (truncated):
researcher: Termination conditions prevent runaway token spend and infinite agent loops.
writer: They act as circuit breakers, ensuring a multi-agent chat stops when its goal is met or a safe limit is hit.
researcher: Without them, two agents can politely ping-pong forever.
writer: That wastes API budget and masks failures as "still running."After the fourth message the team returns control. The task may be unfinished, but the loop is bounded.
Stop on explicit signal
A message cap is blind. For task-driven stops, use TextMentionTermination. Instruct one agent to emit a sentinel when done.
from autogen_agentchat.conditions import TextMentionTermination
researcher = AssistantAgent(
"researcher",
model_client=model_client,
system_message="Research briefly. When the writer confirms, reply with exactly TERMINATE.",
)
writer = AssistantAgent(
"writer",
model_client=model_client,
system_message="Write one sentence. If it covers the topic, reply with exactly TERMINATE.",
)
team = RoundRobinGroupChat(
[researcher, writer],
termination_condition=TextMentionTermination("TERMINATE"),
)Run the same run() loop. Output ends when an agent prints TERMINATE (the string is stripped from the delivered message by the condition). This is the core pattern in most autogen agent team termination condition tutorial examples: a cheap, deterministic stop word.
Compose conditions with OR / AND
Real teams need both: stop on completion or after a safety cap. OrTerminationCondition and AndTerminationCondition compose existing checks.
from autogen_agentchat.conditions import OrTerminationCondition
safety = MaxMessageTermination(max_messages=10)
signal = TextMentionTermination("TERMINATE")
cond = OrTerminationCondition([safety, signal])
team = RoundRobinGroupChat([researcher, writer], termination_condition=cond)Use AndTerminationCondition when you want both a signal and a minimum number of exchanges (rare, but useful for forced review cycles).
Custom predicate termination
When a keyword is too loose, subclass TerminationCondition. The protocol is a single async __call__ receiving the message history. Below we stop when the last message contains a JSON status field.
from autogen_agentchat.base import TerminationCondition
class JsonStatusDone(TerminationCondition):
async def __call__(self, messages) -> bool:
if not messages:
return False
last = messages[-1]
content = getattr(last, "content", "")
return '"status":"done"' in content.lower()
custom_team = RoundRobinGroupChat(
[researcher, writer],
termination_condition=OrTerminationCondition([JsonStatusDone(), MaxMessageTermination(8)]),
)This pattern shines when agents emit structured output (tool calls, JSON) and you want to terminate on a parsed state, not raw text. In this autogen agent team termination condition tutorial we keep the class minimal, but you can track speaker roles, token counts, or external state inside the instance.
Unit testing your condition
Because the condition is just an async callable over messages, you can test it without any model calls:
import pytest
@pytest.mark.asyncio
async def test_json_done():
cond = JsonStatusDone()
fake_msg = type("M", (), {"content": '{"status":"done"}'})()
assert await cond([fake_msg]) is True
fake_msg2 = type("M", (), {"content": "still working"})()
assert await cond([fake_msg2]) is FalseToken usage termination
For cost control, TokenUsageTermination halts after a cumulative token budget.
from autogen_agentchat.conditions import TokenUsageTermination
budget = TokenUsageTermination(max_total_tokens=5000)Wire it into an Or with your signal condition. Because n4n.ai and similar gateways meter per-token usage upstream, you can also enforce budgets at the proxy layer, but in-agent termination avoids the extra round trip.
Inspecting why the team stopped
The non-streaming team.run(task) returns a result object that references the condition which fired. Use it in logs:
result = await team.run(task="Explain AutoGen termination conditions.")
print(f"Stopped because: {result.termination_condition}")When streaming, you don’t get that object inline, so keep a reference to your condition instance and inspect its state after the loop exits.
Common pitfalls
- No cap at all. A
RoundRobinGroupChatwithout a termination condition runs until an agent raises or you kill the process. Always ship aMaxMessageTerminationas a backstop. - Sentinel collisions. If “TERMINATE” might appear in normal text, use a rare token or structured JSON. We prefer the JSON predicate above for any agent that emits code.
- Streaming and early exit.
run_streamyields messages as they arrive; your loop should not assume the final message is the terminal one. Check the condition reference if you need to log why it stopped. - Forgetting async. Custom conditions must be
async def __call__. A sync method will break the team’s event loop.
Putting it together
A production-grade research team typically looks like:
from autogen_agentchat.conditions import (
MaxMessageTermination,
TextMentionTermination,
OrTerminationCondition,
)
def build_team():
researcher = AssistantAgent(
"researcher",
model_client=model_client,
system_message="Find facts. End your final answer with DONE.",
)
writer = AssistantAgent(
"writer",
model_client=model_client,
system_message="Synthesize. End with DONE when acceptable.",
)
cond = OrTerminationCondition([
TextMentionTermination("DONE"),
MaxMessageTermination(12),
])
return RoundRobinGroupChat([researcher, writer], termination_condition=cond)
async def main():
team = build_team()
async for msg in team.run_stream(task="Explain AutoGen termination conditions."):
print(msg.source, "->", msg.content)
asyncio.run(main())That’s the full arc of this autogen agent team termination condition tutorial: cap, signal, compose, customize. Pick the loosest condition that guarantees stop, then tighten with semantic checks.