Setting termination conditions in AutoGen group chats

AutoGen group chat termination conditions control when a multi-agent conversation stops, preventing infinite loops and runaway costs. This guide walks through every termination mechanism AutoGen offers — from simple round limits to custom logic — with runnable code you can drop into a script and verify immediately.

Step 1: Understand the termination condition types

AutoGen’s GroupChat class accepts three complementary ways to end a conversation:

max_rounds — Hard cap on speaker turns. One round equals one agent speaking once.
termination_msg — A string or callable that inspects the last message. If it matches (or returns True), the chat stops.
Custom speaker selection — The speaker_selection_method can return None to halt, though this is less common.

You can combine all three. The chat stops when any condition triggers. This matters: a custom function that never returns True will still yield to max_rounds, which acts as a safety net.

Step 2: Set up a minimal group chat

Create a fresh virtual environment and install AutoGen:

python -m venv .venv && source .venv/bin/activate
pip install pyautogen==0.2.35

Now write a baseline script with two agents and no explicit termination beyond the default max_rounds=10:

# baseline.py
import autogen
from autogen import Agent, GroupChat, GroupChatManager

config_list = [{"model": "gpt-4o-mini", "api_key": "YOUR_KEY"}]

assistant = Agent(
    name="assistant",
    llm_config={"config_list": config_list},
    system_message="You are a helpful assistant. Keep replies under 50 words.",
)

user_proxy = Agent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config=False,
    llm_config=False,
    system_message="You ask questions and summarize answers. Keep replies under 30 words.",
)

groupchat = GroupChat(
    agents=[user_proxy, assistant],
    messages=[],
    max_rounds=10,
)

manager = GroupChatManager(groupchat=groupchat, llm_config={"config_list": config_list})

user_proxy.initiate_chat(manager, message="What is the capital of France?")
print(f"Total rounds: {len(groupchat.messages)}")

Run it:

python baseline.py

Verify success: The script prints Total rounds: 3 (user → assistant → user summary) and exits cleanly. The default max_rounds=10 was never hit.

Step 3: Implement max_rounds as a safety net

max_rounds is the simplest termination condition. Set it low during development to catch runaway conversations early:

# max_rounds_demo.py
groupchat = GroupChat(
    agents=[user_proxy, assistant],
    messages=[],
    max_rounds=3,  # Hard stop after 3 speaker turns
)

With max_rounds=3, the same question produces:

Total rounds: 3

The conversation stops mid-flow — the assistant answers, but the user_proxy never gets to summarize. This is intentional: max_rounds counts speaker turns, not full cycles. Use it as a budget guardrail, not a precision tool.

Verify success: Run the script and confirm the round count matches your max_rounds value exactly. Check that no exception is raised — the manager simply stops selecting speakers.

Step 4: Add a termination_msg string match

The termination_msg parameter accepts a string. When the last message’s content contains that substring (case-sensitive), the chat ends after that message is added:

# string_termination.py
groupchat = GroupChat(
    agents=[user_proxy, assistant],
    messages=[],
    max_rounds=20,
    termination_msg="TERMINATE",
)

Modify the assistant’s system message to emit the keyword when done:

assistant = Agent(
    name="assistant",
    llm_config={"config_list": config_list},
    system_message=(
        "You are a helpful assistant. When the user asks for a fact, "
        "answer in one sentence then append ' TERMINATE' exactly."
    ),
)

Run it:

python string_termination.py

Output shows the assistant’s reply ends with TERMINATE, and the chat stops immediately — the user_proxy never speaks again.

Verify success: Check the final message in groupchat.messages[-1]["content"] contains TERMINATE. Confirm the total rounds equal 2 (user → assistant).

Caveat: substring matching

termination_msg="DONE" matches "TASK DONE" and "DONE WITH THIS". If you need exact matches or structured signals, use a callable instead (Step 5).

Step 5: Write a custom termination callable

A callable termination_msg receives the last message dict and returns True to stop. This lets you inspect role, content, tool calls, or any metadata:

# callable_termination.py
def stop_on_json_done(last_msg: dict) -> bool:
    """Stop when assistant emits a JSON object with status: complete."""
    if last_msg.get("role") != "assistant":
        return False
    content = last_msg.get("content", "")
    try:
        import json
        data = json.loads(content)
        return data.get("status") == "complete"
    except json.JSONDecodeError:
        return False

groupchat = GroupChat(
    agents=[user_proxy, assistant],
    messages=[],
    max_rounds=15,
    termination_msg=stop_on_json_done,
)

Update the assistant to emit structured output:

assistant = Agent(
    name="assistant",
    llm_config={"config_list": config_list},
    system_message=(
        "Answer the user's question. Then output ONLY a JSON object: "
        '{"status": "complete", "answer": "<your answer>"}'
    ),
)

Run it. The chat stops after the assistant’s JSON message. The user_proxy never sees it.

Verify success: Add a debug print inside the callable:

def stop_on_json_done(last_msg: dict) -> bool:
    print(f"[termination check] role={last_msg.get('role')} content={last_msg.get('content')[:80]}")
    ...

You’ll see the check fire after every message. Confirm it returns True exactly once, on the assistant’s JSON turn.

Step 6: Combine multiple conditions

Real systems layer conditions: a semantic “done” signal plus a hard round cap plus a budget check. Here’s a production-style pattern:

# combined_termination.py
import os
from dataclasses import dataclass

@dataclass
class ChatBudget:
    max_rounds: int = 20
    max_tokens: int = 8000
    used_tokens: int = 0

    def check(self, last_msg: dict) -> bool:
        # Token accounting (rough estimate)
        content = last_msg.get("content", "")
        self.used_tokens += len(content) // 4
        if self.used_tokens >= self.max_tokens:
            print(f"[budget] Token limit reached: {self.used_tokens}")
            return True
        return False

budget = ChatBudget(max_rounds=15, max_tokens=4000)

def combined_termination(last_msg: dict) -> bool:
    # 1. Semantic done signal
    if last_msg.get("role") == "assistant":
        content = last_msg.get("content", "")
        if content.strip().endswith("<<DONE>>"):
            print("[termination] Semantic done signal detected")
            return True
    # 2. Budget check
    if budget.check(last_msg):
        return True
    return False

groupchat = GroupChat(
    agents=[user_proxy, assistant],
    messages=[],
    max_rounds=budget.max_rounds,
    termination_msg=combined_termination,
)

The assistant system message now ends with <<DONE>>. The budget tracker approximates tokens per message (4 chars ≈ 1 token). Whichever condition fires first wins.

Verify success: Run with a long prompt that would exceed the token budget before <<DONE>> appears. Confirm the budget path triggers. Then run a short prompt and confirm the semantic path triggers. In both cases, len(groupchat.messages) <= budget.max_rounds.

Step 7: Handle termination in speaker selection

Advanced control: a custom speaker_selection_method can return None to stop the chat. This is useful when the next speaker depends on runtime state that the termination callable can’t see (e.g., a tool result stored in agent memory).

# speaker_selection_termination.py
def select_speaker(last_speaker: Agent, groupchat: GroupChat) -> Agent | None:
    messages = groupchat.messages
    if not messages:
        return groupchat.agents[0]  # Start with first agent

    last_msg = messages[-1]
    # Stop if last message was a tool result with error
    if last_msg.get("role") == "tool" and "error" in last_msg.get("content", "").lower():
        print("[speaker selection] Tool error detected, stopping")
        return None

    # Round-robin fallback
    agents = groupchat.agents
    last_idx = agents.index(last_speaker)
    return agents[(last_idx + 1) % len(agents)]

groupchat = GroupChat(
    agents=[user_proxy, assistant],
    messages=[],
    max_rounds=10,
    speaker_selection_method=select_speaker,
)

Verify success: Inject a tool error message manually in a test:

groupchat.messages.append({
    "role": "tool",
    "content": "Error: API rate limit exceeded",
    "name": "some_tool",
})
# Then call select_speaker manually and assert it returns None

Step 8: Test and verify termination behavior systematically

Don’t rely on manual runs. Write a test suite that exercises each condition:

# test_termination.py
import pytest
from autogen import GroupChat, Agent

def make_test_chat(termination_msg, max_rounds=5):
    a1 = Agent(name="a1", llm_config=False, human_input_mode="NEVER")
    a2 = Agent(name="a2", llm_config=False, human_input_mode="NEVER")
    return GroupChat(
        agents=[a1, a2],
        messages=[],
        max_rounds=max_rounds,
        termination_msg=termination_msg,
    )

def test_max_rounds_stops_exactly_at_limit():
    chat = make_test_chat(termination_msg=None, max_rounds=3)
    # Simulate 3 rounds
    for i in range(3):
        chat.messages.append({"role": "assistant", "content": f"msg {i}"})
        # Manually invoke the internal check
        from autogen.groupchat import _should_terminate
        if _should_terminate(chat, chat.messages[-1]):
            break
    assert len(chat.messages) == 3

def test_string_termination_triggers_on_substring():
    chat = make_test_chat(termination_msg="STOP_HERE", max_rounds=10)
    chat.messages.append({"role": "assistant", "content": "working"})
    chat.messages.append({"role": "assistant", "content": "almost STOP_HERE done"})
    from autogen.groupchat import _should_terminate
    assert _should_terminate(chat, chat.messages[-1]) is True

def test_callable_termination_receives_last_message():
    captured = {}
    def capture(last_msg):
        captured["last"] = last_msg
        return False
    chat = make_test_chat(termination_msg=capture, max_rounds=10)
    test_msg = {"role": "user", "content": "hello"}
    chat.messages.append(test_msg)
    from autogen.groupchat import _should_terminate
    _should_terminate(chat, test_msg)
    assert captured["last"] is test_msg

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

Run with python test_termination.py. All three tests should pass.

Verify success: The tests confirm:

max_rounds stops at exactly N messages
String termination matches substrings
Callable termination receives the correct message dict

Step 9: Debug common termination failures

Symptom	Likely cause	Fix
Chat never stops	`termination_msg` callable returns `False` always; `max_rounds` too high	Add logging inside callable; lower `max_rounds`
Chat stops too early	`termination_msg` string matches a common word (“done”, “complete”)	Use a unique sentinel like `<<TASK_COMPLETE>>` or a callable
Termination runs but agents keep speaking	You’re checking `groupchat.messages` after `initiate_chat` returns, but the manager may have queued extra calls	Inspect messages inside a custom speaker selection or termination callable
Token budget never triggers	Rough char/4 estimate is too loose	Use `tiktoken` for accurate counts: `len(enc.encode(content))`

Add this debug helper to any script:

def debug_termination(chat: GroupChat, last_msg: dict) -> bool:
    print(f"[DEBUG] Round {len(chat.messages)} | Role: {last_msg.get('role')} | Content: {last_msg.get('content')[:100]}")
    return False  # Never actually terminate, just log

groupchat = GroupChat(
    ...,
    termination_msg=debug_termination,  # Temporary
)

Step 10: Integrate with an inference gateway (optional)

If you route through a gateway that exposes per-token usage in response headers, you can implement precise budget termination without local estimation. For example, n4n.ai returns x-usage-prompt-tokens and x-usage-completion-tokens on every response. A termination callable can accumulate those instead of estimating:

# Requires a wrapper that captures gateway headers
class GatewayBudget:
    def __init__(self, limit: int):
        self.limit = limit
        self.used = 0

    def check(self, response_headers: dict) -> bool:
        prompt = int(response_headers.get("x-usage-prompt-tokens", 0))
        completion = int(response_headers.get("x-usage-completion-tokens", 0))
        self.used += prompt + completion
        return self.used >= self.limit

Wire this into your agent’s llm_config via a custom client that stores the last response headers where your termination callable can read them. This eliminates localStorage them.

Quick reference: termination condition precedence

Condition	Evaluated	Stops before next speaker?
`speaker_selection_method` returns `None`	Before each turn	Yes
`termination_msg` callable returns `True`	After each message	Yes
`termination_msg` string matches	After each message	Yes
`max_rounds` reached	Before each turn	Yes

All are independent — any True stops the chat. Order of evaluation is roughly: speaker selection → max_rounds → termination_msg. But don’t rely on ordering; make conditions idempotent.

Final verification checklist

Before shipping a group chat to production:

max_rounds set to a defensible upper bound (cost * time)
Semantic termination signal is unique and unambiguous
Callable termination handles None/missing keys gracefully
Unit tests cover each condition in isolation
Integration test runs a full conversation to completion
Logging captures termination reason for observability

You now have a complete, tested termination strategy. The chat stops when you want it to, not when the model decides to keep talking.