Learnixo

AutoGen Essentials · Lesson 10 of 11

Termination Conditions

Why Termination Matters

An AutoGen conversation without a reliable termination condition is a liability. In development, it means you wait too long and burn API tokens. In production, it means an agent loop runs until it hits your rate limit or costs hundreds of dollars.

AutoGen provides four termination mechanisms. You should always use at least two as defence in depth.

Termination mechanisms (use multiple in production):
─────────────────────────────────────────────────────────────
1. TERMINATE keyword  → agent signals "I'm done" in message content
2. max_turns          → hard ceiling on initiate_chat call
3. is_termination_msg → custom function: any condition you want
4. max_consecutive_auto_reply → agent-level safety limit
─────────────────────────────────────────────────────────────

Mechanism 1: The TERMINATE Keyword Convention

The simplest and most common pattern. The assistant is instructed to include "TERMINATE" in its final message, and the user proxy checks for it.

Setting Up the Convention

Python
import autogen
import os

llm_config = {
    "config_list": [{"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}],
    "temperature": 0,
}

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    system_message="""You are a helpful assistant.
    Complete the given task fully.
    When you are confident the task is done, end your message with: TERMINATE

    Only say TERMINATE when the task is genuinely complete. Do not say it prematurely.
    """,
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
    code_execution_config={"work_dir": "workspace", "use_docker": False},
)

How It Works

When is_termination_msg(message) returns True, the conversation stops immediately — no further messages are sent or processed.

Python
# Simple check: case-sensitive exact match
is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", "")

# Case-insensitive (safer against model variation)
is_termination_msg=lambda msg: "terminate" in msg.get("content", "").lower()

# Only terminate if TERMINATE is at the end (reduce false positives)
is_termination_msg=lambda msg: msg.get("content", "").strip().endswith("TERMINATE")

Handling None Content

LLM responses can occasionally return None content (e.g., if the model calls a function and returns no text). Always guard against this:

Python
def is_done(msg: dict) -> bool:
    """Safe termination check that handles None content."""
    content = msg.get("content")
    if content is None:
        return False
    if not isinstance(content, str):
        return False
    return "TERMINATE" in content

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    is_termination_msg=is_done,
    ...
)

Mechanism 2: max_turns on initiate_chat

max_turns is a parameter on the initiate_chat call. It limits the total number of back-and-forth turns regardless of message content.

Python
user_proxy.initiate_chat(
    assistant,
    message="Explain quantum computing in 3 paragraphs.",
    max_turns=4,    # conversation ends after 4 turns, period
)

Turn counting: one turn = one message from each agent in the pair. With max_turns=4, you get up to 4 user_proxy messages and 4 assistant messages.

Important distinction:

| Setting | Where | What it counts | |---|---|---| | max_turns | initiate_chat() parameter | Total turns in this specific chat | | max_consecutive_auto_reply | UserProxyAgent constructor | Consecutive automated replies from the proxy |

Python
# max_consecutive_auto_reply resets when human input is provided (TERMINATE or ALWAYS mode)
# max_turns is absolute — it counts regardless of human intervention

Mechanism 3: Custom is_termination_msg Function

The is_termination_msg parameter accepts any callable that takes a message dict and returns a boolean. This unlocks sophisticated termination logic.

Example 1: Multiple Trigger Words

Python
DONE_SIGNALS = {"TERMINATE", "DONE", "COMPLETE", "FINISHED", "LGTM"}

def multi_signal_termination(msg: dict) -> bool:
    """Terminate on any of several completion signals."""
    content = msg.get("content")
    if not isinstance(content, str):
        return False
    content_upper = content.upper()
    return any(signal in content_upper for signal in DONE_SIGNALS)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    is_termination_msg=multi_signal_termination,
    ...
)

Example 2: Termination Based on Code Execution Success

Python
def terminate_on_success(msg: dict) -> bool:
    """
    Terminate only when:
    1. The assistant says TERMINATE, AND
    2. The most recent code execution succeeded (exitcode: 0)
    """
    content = msg.get("content", "")
    if not isinstance(content, str):
        return False

    has_terminate = "TERMINATE" in content
    has_success = "exitcode: 0" in content

    # Only terminate if TERMINATE is present
    # If it's in an execution result message, ignore (those are from user_proxy)
    role = msg.get("role", "")
    name = msg.get("name", "")

    if name == "assistant" and has_terminate:
        return True

    return False

Example 3: Terminate After Achieving a Specific Goal

Python
def goal_achieved_termination(msg: dict) -> bool:
    """
    Terminate when the agent explicitly says the goal is achieved
    in a structured format.
    """
    import json
    content = msg.get("content", "")
    if not isinstance(content, str):
        return False

    # Look for structured completion signal
    if "STATUS: COMPLETE" in content:
        return True

    # Also accept the conventional TERMINATE
    if "TERMINATE" in content:
        return True

    return False


# In the system message, instruct the agent to use the format:
assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    system_message="""Complete the given task.
    When done, end your message with exactly:
    STATUS: COMPLETE
    TERMINATE
    """,
)

Example 4: Terminate Based on Message Count

Python
class CountingTerminator:
    """Terminate after the assistant has sent N messages."""

    def __init__(self, max_assistant_messages: int):
        self.max_messages = max_assistant_messages
        self.count = 0

    def __call__(self, msg: dict) -> bool:
        if msg.get("name") == "assistant":
            self.count += 1
        if "TERMINATE" in msg.get("content", ""):
            return True
        return self.count >= self.max_messages


# Use it
terminator = CountingTerminator(max_assistant_messages=5)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    is_termination_msg=terminator,   # callable object works the same as lambda
    ...
)

Mechanism 4: max_consecutive_auto_reply

This is the agent-level safety net. If is_termination_msg never triggers (because the LLM forgot to say TERMINATE, or there is a bug in your termination logic), max_consecutive_auto_reply kicks in.

Python
user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,   # safety: stop after 10 automated replies
    is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
    ...
)

When max_consecutive_auto_reply is reached:

  • In NEVER mode: the conversation ends silently
  • In TERMINATE mode: the human is prompted to continue or stop
  • In ALWAYS mode: the human is prompted (as usual)

Recommended values:

| Workflow type | Recommended max_consecutive_auto_reply | |---|---| | Simple Q&A (no code) | 3 to 5 | | Code generation with tests | 8 to 12 | | Complex multi-step research | 15 to 20 | | Group chat (per agent) | 10 to 15 |

Timeout Handling

AutoGen does not natively support wall-clock timeouts on the entire conversation — only on individual code executions. To add a conversation-level timeout, wrap initiate_chat in a thread with a timeout:

Python
import threading
import autogen

def run_with_timeout(user_proxy, assistant, message: str, timeout_seconds: int = 120):
    """Run an AutoGen conversation with a wall-clock timeout."""
    result = {"completed": False, "error": None}

    def run():
        try:
            user_proxy.initiate_chat(assistant, message=message)
            result["completed"] = True
        except Exception as e:
            result["error"] = str(e)

    thread = threading.Thread(target=run, daemon=True)
    thread.start()
    thread.join(timeout=timeout_seconds)

    if thread.is_alive():
        print(f"[TIMEOUT] Conversation exceeded {timeout_seconds}s limit.")
        # The thread is daemon=True, so it will be killed when main thread ends
        return None, False

    if result["error"]:
        print(f"[ERROR] {result['error']}")
        return None, False

    return user_proxy.chat_messages.get(assistant, []), True


# Usage
messages, completed = run_with_timeout(
    user_proxy, assistant,
    message="Write a comprehensive analysis of distributed systems.",
    timeout_seconds=60,
)

if completed:
    print(f"Conversation completed with {len(messages)} messages.")
else:
    print("Conversation timed out — partial results may be available.")

For async environments (v0.4), use asyncio.wait_for:

Python
import asyncio
from autogen_agentchat.teams import RoundRobinGroupChat

async def run_with_async_timeout(team, task: str, timeout: float = 60.0):
    try:
        result = await asyncio.wait_for(
            team.run(task=task),
            timeout=timeout,
        )
        return result
    except asyncio.TimeoutError:
        print(f"Conversation timed out after {timeout}s")
        return None

Best Practice: Defence in Depth

Never rely on a single termination mechanism. Here is the production-recommended pattern:

Python
import autogen
import os

llm_config = {
    "config_list": [{"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}],
    "temperature": 0,
}

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    system_message="""You are a Python expert.
    Complete the task, test your code, then end with TERMINATE.
    """,
)


def safe_termination(msg: dict) -> bool:
    """
    Defence-in-depth termination check.
    Handles None content, non-string content, and multiple trigger words.
    """
    content = msg.get("content")
    if not isinstance(content, str):
        return False
    return any(signal in content for signal in ("TERMINATE", "DONE", "COMPLETE"))


# Layer 1: reliable is_termination_msg
# Layer 2: max_consecutive_auto_reply (agent-level safety net)
user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=12,      # Layer 2: agent safety net
    is_termination_msg=safe_termination,  # Layer 1: keyword check
    code_execution_config={
        "work_dir": "workspace",
        "use_docker": False,
        "timeout": 30,                   # Layer 3: per-execution timeout
    },
)

# Layer 4: initiate_chat max_turns (absolute ceiling)
user_proxy.initiate_chat(
    assistant,
    message="Write a merge sort function with tests.",
    max_turns=20,                        # Layer 4: hard ceiling on this chat
)

Common Termination Bugs and Fixes

Bug 1: LLM sometimes forgets to say TERMINATE

The model may occasionally complete a task without including the keyword.

Fix: Add max_consecutive_auto_reply as a backup:

Python
max_consecutive_auto_reply=10  # catches cases where TERMINATE is missing

Bug 2: TERMINATE appears in the middle of a message (false positive)

If the assistant writes something like "I need to TERMINATE the loop variable", the conversation ends prematurely.

Fix: Check that TERMINATE appears at the end:

Python
is_termination_msg=lambda msg: (
    isinstance(msg.get("content"), str)
    and msg["content"].strip().endswith("TERMINATE")
)

Bug 3: Content is a list (function call results)

Tool call responses in some AutoGen versions return content as a list of dicts, not a string.

Fix:

Python
def robust_termination(msg: dict) -> bool:
    content = msg.get("content")
    if isinstance(content, list):
        # Join all text parts
        text_parts = [
            part.get("text", "") for part in content
            if isinstance(part, dict)
        ]
        content = " ".join(text_parts)
    if not isinstance(content, str):
        return False
    return "TERMINATE" in content

Bug 4: Conversation loops after TERMINATE

If both agents have is_termination_msg set, make sure only the right one is checking.

Fix: Only the UserProxyAgent (the one that called initiate_chat) needs is_termination_msg. The AssistantAgent does not need it.

Summary

  • Use at least two termination mechanisms in every AutoGen workflow
  • The TERMINATE keyword is the primary signal — always instruct your agent to use it
  • is_termination_msg is a custom function: check for None/non-string content defensively
  • max_consecutive_auto_reply is the agent-level safety net
  • max_turns on initiate_chat is the hard ceiling
  • Code execution timeouts (timeout in code_execution_config) prevent runaway processes
  • For wall-clock conversation timeouts, use threading.Thread with join(timeout=N)

Next: the final lesson — a structured interview comparing AutoGen and LangGraph, covering real design decisions you will face in production.

Previous

GroupChatManager: Selecting the Next Speaker

Next Lesson

Interview: AutoGen vs LangGraph — When Would You Choose?