Nested conversations in AutoGen: a practical guide

Nested conversations in AutoGen let you decompose complex tasks into isolated sub-dialogues that run independently before returning results to a parent conversation. This pattern is essential when you need parallel fact-finding, specialized reasoning loops, or want to prevent context pollution between unrelated subtasks. This autogen nested conversations guide walks through the patterns that work in production, the APIs you’ll actually use, and the failure modes that bite teams the first time they try this.

Understanding the nested conversation model

AutoGen’s conversation model is fundamentally a sequence of messages between agents. A nested conversation is simply a conversation that gets spawned from within another conversation — typically triggered by a function call, a specific agent’s response, or an explicit handoff. The child conversation maintains its own message history, agent roster, and termination conditions, then returns a summary or structured result to the parent.

The key distinction from a flat group chat: context isolation. In a standard group chat, every agent sees every message. In a nested conversation, the child’s internal deliberations stay private unless explicitly surfaced. This matters for token budgets, reasoning clarity, and preventing cross-contamination of intermediate hypotheses.

# Conceptual model: parent spawns child, gets result back
parent_chat = autogen.GroupChat(agents=[manager, researcher, writer])
# Inside researcher's turn, it might spawn:
child_result = await run_nested_conversation(
    task="Compare three vector DBs for RAG",
    agents=[analyst, engineer],
    max_rounds=6
)
# researcher receives child_result and continues in parent

When to reach for nesting

Not every multi-agent workflow needs nesting. Use it when:

• Subtasks require different agent specializations that don’t belong in the main conversation (e.g., a code review agent that shouldn’t see product strategy discussions)
• You need parallel exploration — spawn multiple nested conversations simultaneously for independent research tracks
• Context window pressure is real — child conversations can summarize aggressively without losing parent context
• Failure isolation matters — a hallucinating child agent shouldn’t derail the parent’s reasoning

Avoid nesting when the task is naturally linear, when agents need full shared context, or when the overhead of managing child lifecycles exceeds the benefit.

The core pattern: function-tool spawning

The most reliable way to spawn nested conversations in AutoGen 0.2+ is registering a function tool on the parent agent that internally runs a complete conversation. This keeps the parent’s LLM in control of when to delegate, while the function handles the mechanics.

import autogen
from typing import List, Dict, Any
import asyncio

async def run_research_subtask(topic: str, questions: List[str]) -> Dict[str, Any]:
    """Spawn a focused research conversation and return structured findings."""
    
    researcher = autogen.AssistantAgent(
        name="researcher",
        system_message="You answer research questions concisely. Cite sources.",
        llm_config={"model": "gpt-4o-mini"}
    )
    
    critic = autogen.AssistantAgent(
        name="critic",
        system_message="You verify claims and flag unsupported assertions.",
        llm_config={"model": "gpt-4o-mini"}
    )
    
    # Build the child conversation
    child_chat = autogen.GroupChat(
        agents=[researcher, critic],
        messages=[],
        max_round=len(questions) * 2
    )
    
    manager = autogen.GroupChatManager(
        groupchat=child_chat,
        llm_config={"model": "gpt-4o"}
    )
    
    # Prime with the research brief
    init_msg = f"Research topic: {topic}\nQuestions:\n" + "\n".join(f"- {q}" for q in questions)
    
    # Run to completion
    result = await manager.a_run(init_msg)
    
    # Extract structured summary from final messages
    summary = extract_findings(child_chat.messages)
    return {"topic": topic, "findings": summary, "rounds": len(child_chat.messages)}

def extract_findings(messages: List[Dict]) -> List[Dict]:
    """Parse the child conversation into structured output."""
    findings = []
    for msg in messages:
        if msg.get("name") == "critic" and "verified" in msg.get("content", "").lower():
            findings.append({"claim": msg["content"], "status": "verified"})
        elif msg.get("name") == "researcher":
            findings.append({"claim": msg["content"], "status": "proposed"})
    return findings

# Register on parent agent
planner = autogen.AssistantAgent(
    name="planner",
    system_message="You break down complex topics and delegate research via run_research_subtask.",
    llm_config={"model": "gpt-4o"}
)

planner.register_for_llm(name="run_research_subtask", description="Delegate research to a nested conversation")(run_research_subtask)

This pattern gives you explicit control over the child’s agent composition, termination logic, and result extraction. The parent LLM decides what to research; the function decides how.

Managing context and state

The biggest trap in nested conversations is state leakage. Here’s what you need to handle explicitly:

Message history isolation

By default, each GroupChat maintains its own messages list. But if you pass agent instances that have been used elsewhere, their internal chat_history (the per-agent message cache) may carry baggage. Always instantiate fresh agents for each child conversation, or call agent.clear_history() before spawning.

# Bad: reusing agents leaks context
shared_researcher = autogen.AssistantAgent(name="researcher", ...)
# ... used in parent ...
child_chat = GroupChat(agents=[shared_researcher, critic])  # researcher carries parent history

# Good: fresh instances per spawn
def make_researcher():
    return autogen.AssistantAgent(name="researcher", system_message=..., llm_config=...)

child_chat = GroupChat(agents=[make_researcher(), make_critic()])

Token budget accounting

Nested conversations consume tokens independently. If you’re operating near context limits, you need to account for both parent and child usage. A practical pattern: have the child return a token count alongside its result, and let the parent decide whether to continue or summarize.

async def run_budgeted_research(topic: str, max_tokens: int = 8000) -> Dict:
    child_result = await run_research_subtask(topic, QUESTIONS)
    
    # Estimate tokens (rough: 4 chars ≈ 1 token for English)
    child_tokens = sum(len(str(m.get("content", ""))) for m in child_chat.messages) // 4
    
    if child_tokens > max_tokens:
        # Trigger summarization pass
        child_result["findings"] = await summarize_findings(child_result["findings"], max_tokens // 2)
        child_tokens = estimate_tokens(child_result["findings"])
    
    return {**child_result, "tokens_used": child_tokens}

Result serialization

Don’t return raw message lists. Define a structured schema for child results and enforce it. This makes the parent’s job predictable and lets you version the interface.

from pydantic import BaseModel
from typing import Literal

class ResearchFinding(BaseModel):
    claim: str
    evidence: str
    confidence: Literal["high", "medium", "low"]
    source_agent: str

class ResearchResult(BaseModel):
    topic: str
    findings: List[ResearchFinding]
    rounds: int
    tokens_estimated: int
    truncated: bool = False

Parallel nested conversations

One of the strongest use cases for nesting is fan-out: spawning multiple independent child conversations simultaneously. AutoGen’s GroupChatManager.a_run is async, so you can use asyncio.gather for true parallelism.

async def parallel_research(topics: List[str]) -> List[ResearchResult]:
    """Run multiple research conversations in parallel."""
    
    async def research_one(topic: str) -> ResearchResult:
        return await run_research_subtask(topic, DEFAULT_QUESTIONS)
    
    # Limit concurrency to avoid rate limits
    semaphore = asyncio.Semaphore(3)
    
    async def bounded(topic: str) -> ResearchResult:
        async with semaphore:
            return await research_one(topic)
    
    results = await asyncio.gather(*[bounded(t) for t in topics])
    return results

Pitfall: If you’re using a shared LLM endpoint with rate limits, unbounded parallelism will get you 429ed. Always use a semaphore or a dedicated rate limiter. If you’re routing through a gateway like n4n.ai, you get automatic fallback across providers, but you still need client-side concurrency control to avoid overwhelming any single upstream.

Handoff patterns: explicit vs. implicit

Two schools of thought on how the parent triggers nesting:

Explicit handoff (recommended for clarity)

The parent agent emits a structured tool call. The function runs the child conversation synchronously (or async) and returns a result. The parent sees the result as a tool response and continues.

# Parent system message excerpt
"""
When you need focused research, call run_research_subtask with:
- topic: the research question
- questions: specific sub-questions to answer
The tool returns structured findings you can cite directly.
"""

Pros: Visible in conversation history, debuggable, LLM controls delegation. Cons: Parent must wait for child to complete (sequential unless you design async tools).

Implicit handoff (via speaker selection)

In a GroupChat, you can designate a “router” agent whose job is to select the next speaker. That router could dynamically add a “nested conversation agent” to the chat, which internally runs a sub-dialogue before yielding back.

class NestedConversationAgent(autogen.AssistantAgent):
    def __init__(self, subtask_factory, **kwargs):
        super().__init__(**kwargs)
        self.subtask_factory = subtask_factory
    
    async def generate_reply(self, messages, sender, config):
        # Run the nested conversation
        result = await self.subtask_factory(messages[-1]["content"])
        # Return summary as this agent's reply
        return f"Subtask complete: {result.summary}"

Pros: Feels like a natural participant in the group chat. Cons: Harder to debug, context pollution risk, termination logic gets fuzzy.

Recommendation: Start with explicit tool-based handoffs. They map to how engineers think about function calls and are easier to instrument.

Termination and failure handling

Child conversations need their own termination conditions. Don’t rely on the parent’s max_round — the child might need more or fewer rounds.

def make_child_chat(agents, task_type: str) -> autogen.GroupChat:
    termination_configs = {
        "research": {"max_round": 8, "termination_msg": "RESEARCH_COMPLETE"},
        "code_review": {"max_round": 5, "termination_msg": "REVIEW_DONE"},
        "fact_check": {"max_round": 3, "termination_msg": "VERIFIED"},
    }
    cfg = termination_configs.get(task_type, {"max_round": 6})
    
    return autogen.GroupChat(
        agents=agents,
        messages=[],
        max_round=cfg["max_round"],
        # Custom termination: look for keyword in last message
        speaker_selection_method="auto"
    )

Handle child failures gracefully. If a child conversation hits its round limit without producing a usable result, return a structured error the parent can reason about — not an exception that crashes the parent.

async def safe_run_child(chat: autogen.GroupChat, init_msg: str) -> Dict:
    try:
        manager = autogen.GroupChatManager(groupchat=chat, llm_config=LLM_CONFIG)
        await manager.a_run(init_msg)
        
        # Check if we got a real result
        last_msg = chat.messages[-1] if chat.messages else {}
        if "COMPLETE" not in last_msg.get("content", ""):
            return {"status": "incomplete", "messages": chat.messages, "error": "No termination signal"}
        
        return {"status": "success", "result": extract_result(chat.messages)}
    except Exception as e:
        return {"status": "error", "error": str(e), "messages": chat.messages}

Debugging nested conversations

When things go wrong, you need visibility into the child conversation without drowning in logs. Three practical techniques:

1. Structured logging with correlation IDs

import uuid
import logging

logger = logging.getLogger("autogen.nested")

async def run_traced_child(task_id: str, topic: str) -> Dict:
    trace_id = f"{task_id}-{uuid.uuid4().hex[:8]}"
    logger.info(f"[{trace_id}] Starting child conversation for: {topic}")
    
    # Inject trace_id into agent names for log correlation
    researcher = autogen.AssistantAgent(name=f"researcher-{trace_id}", ...)
    
    result = await run_child_conversation(researcher, topic)
    
    logger.info(f"[{trace_id}] Completed with {len(result.get('findings', []))} findings")
    return result

2. Conversation replay utility

Build a small script that takes a saved message list and re-renders it with timestamps, token counts, and agent roles. Invaluable for post-mortems.

def replay_conversation(messages: List[Dict], show_tokens: bool = True):
    for i, msg in enumerate(messages):
        role = msg.get("name", msg.get("role", "unknown"))
        content = msg.get("content", "")[:200]
        tokens = len(content) // 4 if show_tokens else "?"
        print(f"[{i:03d}] {role:20s} ({tokens:>4}t) {content}")

3. Interactive debugger breakpoint

In development, drop into a REPL at child conversation boundaries:

# In your child conversation factory
import ipdb; ipdb.set_trace()  # or breakpoint() in Python 3.7+
# Inspect chat.messages, agent states, token counts before returning

Common pitfalls and how to avoid them

Testing strategies

Unit test the child conversation factory in isolation. Mock the LLM responses to verify:

• The child terminates correctly on your termination signal
• Result extraction produces valid schema
• Token estimation is in the right ballpark
• Error paths return structured errors, not exceptions

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_research_child_terminates():
    with patch("autogen.GroupChatManager.a_run", new_callable=AsyncMock) as mock_run:
        mock_run.return_value = None
        # Inject messages that include termination signal
        chat = make_child_chat([make_researcher(), make_critic()], "research")
        chat.messages = [
            {"role": "user", "content": "Research X"},
            {"name": "researcher", "content": "Found A, B, C"},
            {"name": "critic", "content": "VERIFIED: A, B, C. RESEARCH_COMPLETE"}
        ]
        
        result = await safe_run_child(chat, "Research X")
        
        assert result["status"] == "success"
        assert len(result["result"]["findings"]) == 3

Integration test the parent→child handoff with a real (cheap) model. Verify the parent actually calls the tool with sensible arguments and incorporates the result.

Scaling considerations

As your nested conversation topology grows, you’ll hit operational limits:

• Observability: You need distributed tracing across parent/child boundaries. OpenTelemetry spans with parent_id linking child to parent.
• Cost attribution: Tag each conversation with a project_id and conversation_type so you can break down spend by workflow.
• Rate limit coordination: If multiple parent conversations spawn children simultaneously, you need a global rate limiter, not per-process semaphores.
• State persistence: For long-running workflows, serialize child results to a database (not just in-memory) so you can resume after crashes.

Putting it together: a complete example

Here’s a minimal but complete pattern you can adapt:

# nested_conversations.py
import autogen
import asyncio
from pydantic import BaseModel
from typing import List, Literal
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

LLM_CONFIG = {"model": "gpt-4o-mini", "temperature": 0.3}

class Finding(BaseModel):
    claim: str
    confidence: Literal["high", "medium", "low"]

class ResearchResult(BaseModel):
    topic: str
    findings: List[Finding]
    tokens_estimated: int

async def research_subtask(topic: str, questions: List[str]) -> ResearchResult:
    """Run a focused research conversation."""
    trace_id = f"research-{hash(topic) % 10000:04d}"
    logger.info(f"[{trace_id}] Spawning research for: {topic}")
    
    researcher = autogen.AssistantAgent(
        name=f"researcher-{trace_id}",
        system_message="Answer questions concisely. End with RESEARCH_COMPLETE.",
        llm_config=LLM_CONFIG
    )
    critic = autogen.AssistantAgent(
        name=f"critic-{trace_id}",
        system_message="Verify claims. Flag uncertainty. End with VERIFIED or NEEDS_MORE.",
        llm_config=LLM_CONFIG
    )
    
    chat = autogen.GroupChat(
        agents=[researcher, critic],
        messages=[],
        max_round=6
    )
    manager = autogen.GroupChatManager(groupchat=chat, llm_config=LLM_CONFIG)
    
    prompt = f"Topic: {topic}\nQuestions:\n" + "\n".join(f"- {q}" for q in questions)
    await manager.a_run(prompt)
    
    # Extract findings from critic's verified claims
    findings = []
    for msg in chat.messages:
        if msg.get("name", "").startswith("critic") and "VERIFIED" in msg.get("content", ""):
            for line in msg["content"].split("\n"):
                if line.strip().startswith("-"):
                    findings.append(Finding(claim=line.strip()[1:].strip(), confidence="high"))
    
    tokens = sum(len(str(m.get("content", ""))) for m in chat.messages) // 4
    logger.info(f"[{trace_id}] Done: {len(findings)} findings, ~{tokens} tokens")
    
    return ResearchResult(topic=topic, findings=findings, tokens_estimated=tokens)

# Parent agent with tool
planner = autogen.AssistantAgent(
    name="planner",
    system_message="""You decompose complex topics into research questions.
Use the research_subtask tool to delegate. Synthesize results into a final answer.""",
    llm_config={"model": "gpt-4o", "temperature": 0.2}
)

planner.register_for_llm(
    name="research_subtask",
    description="Delegate focused research to a nested conversation"
)(research_subtask)

# Run
async def main():
    user_proxy = autogen.UserProxyAgent(
        name="user",
        human_input_mode="NEVER",
        code_execution_config=False
    )
    
    chat = autogen.GroupChat(agents=[user_proxy, planner], messages=[], max_round=10)
    manager = autogen.GroupChatManager(groupchat=chat, llm_config={"model": "gpt-4o"})
    
    await manager.a_run("Compare PostgreSQL, MongoDB, and Redis for a session store. Consider latency, consistency, and operational complexity.")

if __name__ == "__main__":
    asyncio.run(main())

This pattern — explicit tool delegation, fresh agent instances, structured results, token accounting, and traceable logging — covers 90% of production nested conversation needs. Start here, measure where the bottlenecks are (usually token usage or rate limits), and extend only when you have data.