Executive Summary: Effective multi-agent systems depend on well-designed communication patterns that enable agents to collaborate, share context, and coordinate actions. This comprehensive guide explores AutoGen’s communication mechanisms, from two-agent conversations and group chats to nested conversations and sequential workflows. After implementing complex agent orchestration for enterprise applications, I’ve found that communication pattern selection significantly impacts system reliability, performance, and maintainability. Organizations should design communication flows based on task complexity, required agent interactions, and human oversight needs while implementing proper message routing, context management, and error handling from the start.
Two-Agent Conversation Patterns
The simplest AutoGen pattern involves two agents exchanging messages until a termination condition is met. This pattern suits tasks where a single assistant agent can complete work with feedback from a user proxy. The initiating agent sends a message, the recipient responds, and the conversation continues until one agent signals completion or the maximum turn limit is reached.
Configure conversation flow through max_turns, termination messages, and auto-reply settings. Lower max_turns values prevent runaway conversations but may truncate complex tasks. Termination messages (“TERMINATE”, “TASK_COMPLETE”) provide explicit completion signals. Auto-reply settings control whether agents respond automatically or wait for human input between turns.
Message content shapes agent behavior significantly. Include clear task descriptions, expected output formats, and success criteria in initial messages. Provide context about previous work or constraints. Well-structured initial messages reduce back-and-forth clarification and improve task completion rates.
Group Chat Orchestration
Group chats enable multiple agents to collaborate on complex tasks requiring diverse expertise. A GroupChatManager coordinates speaker selection, ensuring appropriate agents contribute at each conversation turn. This pattern excels for tasks like code review (developer + reviewer + tester), research synthesis (researcher + analyst + writer), or system design (architect + developer + security expert).
Speaker selection methods determine conversation flow. “round_robin” cycles through agents sequentially—predictable but may not match task needs. “auto” uses an LLM to select the most appropriate next speaker based on conversation context—flexible but adds latency and cost. “random” provides variety but lacks strategic coordination. Custom selection functions enable domain-specific routing logic.
Group chat configuration requires careful tuning. Set max_round based on task complexity—too low truncates work, too high wastes resources. Configure allow_repeat_speaker based on whether consecutive contributions from the same agent make sense. Define clear agent descriptions to help the auto-selector make appropriate choices.
Python Implementation: Advanced Communication Patterns
Here’s a comprehensive implementation demonstrating advanced AutoGen communication patterns:
"""Microsoft AutoGen - Advanced Communication Patterns"""
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from autogen.agentchat.contrib.retrieve_assistant_agent import RetrieveAssistantAgent
from typing import Optional, Dict, Any, List, Callable, Union
import logging
from dataclasses import dataclass
from enum import Enum
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SpeakerSelectionMethod(Enum):
"""Speaker selection methods for group chat."""
ROUND_ROBIN = "round_robin"
AUTO = "auto"
RANDOM = "random"
CUSTOM = "custom"
@dataclass
class ConversationConfig:
"""Configuration for conversation patterns."""
max_turns: int = 10
max_rounds: int = 20
termination_msg: str = "TERMINATE"
human_input_mode: str = "NEVER"
speaker_selection: SpeakerSelectionMethod = SpeakerSelectionMethod.AUTO
class CommunicationPatterns:
"""Advanced communication patterns for AutoGen agents."""
def __init__(self, llm_config: Dict[str, Any]):
self.llm_config = llm_config
self.agents: Dict[str, autogen.Agent] = {}
# ==================== Two-Agent Patterns ====================
def create_feedback_loop(
self,
assistant_name: str,
assistant_prompt: str,
max_iterations: int = 5
) -> tuple[AssistantAgent, UserProxyAgent]:
"""Create a feedback loop pattern for iterative refinement."""
assistant = AssistantAgent(
name=assistant_name,
system_message=f"""{assistant_prompt}
After each iteration, evaluate your work against the requirements.
If improvements are needed, explain what you'll change and provide updated output.
When the work meets all requirements, say TERMINATE.""",
llm_config=self.llm_config,
)
critic = UserProxyAgent(
name="critic",
system_message="""You are a critical reviewer. Evaluate the assistant's work against requirements.
Provide specific, actionable feedback for improvements.
If the work is satisfactory, approve it.""",
human_input_mode="NEVER",
max_consecutive_auto_reply=max_iterations,
is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
)
self.agents[assistant_name] = assistant
self.agents["critic"] = critic
return assistant, critic
def create_chain_of_thought(
self,
stages: List[Dict[str, str]]
) -> List[AssistantAgent]:
"""Create a chain-of-thought pattern with sequential agents."""
agents = []
for i, stage in enumerate(stages):
agent = AssistantAgent(
name=stage["name"],
system_message=f"""{stage["prompt"]}
You are stage {i+1} of {len(stages)} in a reasoning chain.
Build upon the previous stage's output.
Provide clear, structured output for the next stage.
When your analysis is complete, say STAGE_COMPLETE.""",
llm_config=self.llm_config,
)
agents.append(agent)
self.agents[stage["name"]] = agent
return agents
# ==================== Group Chat Patterns ====================
def create_debate_pattern(
self,
topic: str,
perspectives: List[Dict[str, str]]
) -> tuple[GroupChat, GroupChatManager]:
"""Create a debate pattern with opposing viewpoints."""
agents = []
# Create moderator
moderator = AssistantAgent(
name="moderator",
system_message=f"""You are a debate moderator for the topic: {topic}
Your responsibilities:
1. Ensure all perspectives are heard fairly
2. Ask clarifying questions
3. Summarize key points from each side
4. Guide the discussion toward resolution
5. When consensus is reached or debate is exhausted, provide final summary and say TERMINATE""",
llm_config=self.llm_config,
)
agents.append(moderator)
self.agents["moderator"] = moderator
# Create debaters
for perspective in perspectives:
debater = AssistantAgent(
name=perspective["name"],
system_message=f"""You are arguing for: {perspective["position"]}
Your role:
1. Present strong arguments for your position
2. Respond to counterarguments constructively
3. Acknowledge valid points from other perspectives
4. Support claims with evidence and reasoning
Be persuasive but respectful. Focus on the strongest arguments.""",
llm_config=self.llm_config,
)
agents.append(debater)
self.agents[perspective["name"]] = debater
# Create user proxy for initiation
user = UserProxyAgent(
name="audience",
human_input_mode="NEVER",
max_consecutive_auto_reply=0,
)
agents.append(user)
group_chat = GroupChat(
agents=agents,
messages=[],
max_round=30,
speaker_selection_method="auto",
allow_repeat_speaker=False,
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config=self.llm_config,
)
return group_chat, manager
def create_review_pipeline(
self,
reviewers: List[Dict[str, str]]
) -> tuple[GroupChat, GroupChatManager]:
"""Create a multi-stage review pipeline."""
agents = []
# Create author agent
author = AssistantAgent(
name="author",
system_message="""You are the content author.
Your responsibilities:
1. Present your work clearly
2. Respond to reviewer feedback
3. Make revisions based on valid suggestions
4. Explain your design decisions when questioned
After addressing all feedback, present the final version.""",
llm_config=self.llm_config,
)
agents.append(author)
self.agents["author"] = author
# Create reviewers
for reviewer in reviewers:
agent = AssistantAgent(
name=reviewer["name"],
system_message=f"""You are a {reviewer["specialty"]} reviewer.
Focus areas: {reviewer["focus"]}
Your responsibilities:
1. Review the work from your specialty perspective
2. Provide specific, actionable feedback
3. Prioritize issues by severity
4. Acknowledge good practices
Be thorough but constructive.""",
llm_config=self.llm_config,
)
agents.append(agent)
self.agents[reviewer["name"]] = agent
# Create coordinator
coordinator = AssistantAgent(
name="coordinator",
system_message="""You are the review coordinator.
Your responsibilities:
1. Ensure all reviewers have provided feedback
2. Consolidate feedback into actionable items
3. Track which issues have been addressed
4. When all issues are resolved, approve and say TERMINATE""",
llm_config=self.llm_config,
)
agents.append(coordinator)
self.agents["coordinator"] = coordinator
# User proxy
user = UserProxyAgent(
name="submitter",
human_input_mode="NEVER",
max_consecutive_auto_reply=0,
)
agents.append(user)
group_chat = GroupChat(
agents=agents,
messages=[],
max_round=40,
speaker_selection_method="auto",
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config=self.llm_config,
)
return group_chat, manager
# ==================== Custom Speaker Selection ====================
def create_custom_selection_chat(
self,
agents: List[autogen.Agent],
selection_func: Callable
) -> tuple[GroupChat, GroupChatManager]:
"""Create group chat with custom speaker selection."""
group_chat = GroupChat(
agents=agents,
messages=[],
max_round=30,
speaker_selection_method=selection_func,
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config=self.llm_config,
)
return group_chat, manager
@staticmethod
def priority_based_selection(
last_speaker: autogen.Agent,
groupchat: GroupChat
) -> Union[autogen.Agent, str]:
"""Custom selection based on agent priority and message content."""
messages = groupchat.messages
if not messages:
return "round_robin"
last_message = messages[-1].get("content", "").lower()
# Priority routing based on keywords
priority_map = {
"security": ["security_expert", "architect"],
"performance": ["performance_engineer", "developer"],
"code": ["developer", "code_reviewer"],
"test": ["tester", "qa_engineer"],
"design": ["architect", "designer"],
}
for keyword, preferred_agents in priority_map.items():
if keyword in last_message:
for agent_name in preferred_agents:
for agent in groupchat.agents:
if agent.name == agent_name and agent != last_speaker:
return agent
# Default to auto selection
return "auto"
# ==================== Nested Conversations ====================
def create_nested_chat(
self,
outer_agents: List[autogen.Agent],
inner_config: Dict[str, Any]
) -> tuple[GroupChat, GroupChatManager]:
"""Create nested conversation pattern."""
# Inner chat for detailed work
inner_agents = []
for agent_config in inner_config.get("agents", []):
agent = AssistantAgent(
name=agent_config["name"],
system_message=agent_config["prompt"],
llm_config=self.llm_config,
)
inner_agents.append(agent)
inner_user = UserProxyAgent(
name="inner_coordinator",
human_input_mode="NEVER",
max_consecutive_auto_reply=5,
)
inner_agents.append(inner_user)
inner_chat = GroupChat(
agents=inner_agents,
messages=[],
max_round=inner_config.get("max_round", 10),
)
inner_manager = GroupChatManager(
groupchat=inner_chat,
llm_config=self.llm_config,
)
# Outer chat with nested capability
outer_chat = GroupChat(
agents=outer_agents + [inner_manager],
messages=[],
max_round=20,
)
outer_manager = GroupChatManager(
groupchat=outer_chat,
llm_config=self.llm_config,
)
return outer_chat, outer_manager
# ==================== Message Routing Utilities ====================
class MessageRouter:
"""Utility for routing messages between agents."""
def __init__(self):
self.routes: Dict[str, List[str]] = {}
self.filters: Dict[str, Callable] = {}
def add_route(
self,
source: str,
destinations: List[str],
filter_func: Optional[Callable] = None
):
"""Add a routing rule."""
self.routes[source] = destinations
if filter_func:
self.filters[source] = filter_func
def get_destinations(
self,
source: str,
message: str
) -> List[str]:
"""Get destination agents for a message."""
if source not in self.routes:
return []
destinations = self.routes[source]
if source in self.filters:
filter_func = self.filters[source]
destinations = [d for d in destinations if filter_func(message, d)]
return destinations
# ==================== Example Usage ====================
def example_debate():
"""Example: Multi-perspective debate pattern."""
llm_config = {
"config_list": [{"model": "gpt-4", "api_key": "your-key"}],
"temperature": 0.7,
}
patterns = CommunicationPatterns(llm_config)
perspectives = [
{"name": "cloud_advocate", "position": "Cloud-native architecture is superior for modern applications"},
{"name": "onprem_advocate", "position": "On-premises infrastructure provides better control and security"},
{"name": "hybrid_advocate", "position": "Hybrid approaches offer the best of both worlds"},
]
group_chat, manager = patterns.create_debate_pattern(
topic="Best infrastructure strategy for enterprise applications",
perspectives=perspectives
)
# Start debate
user = patterns.agents.get("audience")
if user:
user.initiate_chat(
manager,
message="Let's discuss the best infrastructure strategy for a large enterprise migrating legacy applications."
)
def example_review_pipeline():
"""Example: Multi-stage code review pipeline."""
llm_config = {
"config_list": [{"model": "gpt-4", "api_key": "your-key"}],
"temperature": 0.3,
}
patterns = CommunicationPatterns(llm_config)
reviewers = [
{"name": "security_reviewer", "specialty": "security", "focus": "vulnerabilities, authentication, data protection"},
{"name": "performance_reviewer", "specialty": "performance", "focus": "efficiency, scalability, resource usage"},
{"name": "style_reviewer", "specialty": "code quality", "focus": "readability, maintainability, best practices"},
]
group_chat, manager = patterns.create_review_pipeline(reviewers)
# Submit code for review
code_submission = """
Please review this Python function:
def process_user_data(user_id, data):
conn = sqlite3.connect('users.db')
cursor = conn.cursor()
cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
user = cursor.fetchone()
if user:
cursor.execute(f"UPDATE users SET data = '{data}' WHERE id = {user_id}")
conn.commit()
return user
"""
user = UserProxyAgent(name="submitter", human_input_mode="NEVER")
user.initiate_chat(manager, message=code_submission)
if __name__ == "__main__":
print("AutoGen Communication Patterns Examples")
# example_debate()
# example_review_pipeline()Context Management and Message History
Effective context management ensures agents have relevant information without overwhelming their context windows. AutoGen maintains conversation history automatically, but long conversations may exceed model context limits. Implement summarization strategies for extended conversations—periodically condense history into summaries that preserve essential context while reducing token count.
Message filtering controls what information flows between agents. Not all agents need complete conversation history—filter messages to include only relevant exchanges. Use the last_n_messages parameter to limit context to recent interactions. Implement custom message transformers to extract and forward only pertinent information.
State management across conversations requires external storage for persistent context. Store conversation summaries, agent decisions, and task progress in databases or files. Reload relevant state when resuming conversations or starting related tasks. This pattern enables long-running workflows that span multiple sessions.
Key Takeaways and Best Practices
Communication pattern selection significantly impacts multi-agent system effectiveness. Use two-agent patterns for focused tasks with clear feedback loops. Deploy group chats when diverse expertise must collaborate on complex problems. Implement custom speaker selection for domain-specific routing requirements.
The Python examples provided here establish patterns for sophisticated agent communication. Design termination conditions carefully to prevent infinite loops while allowing sufficient iteration for quality output. In the next article, we’ll explore code generation agents and automated development workflows using these communication foundations.
Discover more from Code, Cloud & Context
Subscribe to get the latest posts sent to your email.