Framework Deep Dive: CrewAI - Role-Based Multi-Agent Orchestration
An in-depth exploration of CrewAI's role-based architecture, crew orchestration patterns, task delegation, and production best practices for building collaborative AI agent teams
Microsoft’s AutoGen represents a fundamentally different approach to AI agents. While frameworks like LangChain focus on chains and tools, AutoGen centers on conversation as the primary coordination mechanism. Agents communicate through messages, collaborate on complex tasks, and naturally divide work—much like a team of specialists working together. (For a comparison with other frameworks, see our Complete Guide to AI Agent Frameworks.)
This deep dive explores AutoGen’s architecture, examines its multi-agent patterns, and provides practical guidance for building sophisticated agent systems.
AutoGen is built on a key insight: the most natural way for agents to coordinate is through conversation. Rather than explicit orchestration logic or rigid pipelines, agents simply talk to each other, negotiating who should handle which parts of a task.
This conversational approach offers several advantages:
Natural Task Division: Agents can dynamically decide who handles what based on the conversation context, without predefined routing rules.
Flexible Collaboration: The same agents can work together in different configurations depending on the task, adapting their collaboration patterns on the fly.
Human Integration: Since the coordination mechanism is conversation, humans can participate naturally—providing guidance, reviewing outputs, or stepping in when needed.
Transparency: All coordination happens through messages, making it easy to understand how agents reached their decisions and debug issues.
AutoGen 0.4 introduced a completely redesigned architecture with clear separation of concerns:
The foundation provides event-driven, distributed agent infrastructure:
from autogen_core import MessageContext, RoutedAgent, message_handler
from dataclasses import dataclass
@dataclass
class TextMessage:
content: str
source: str
class CustomAgent(RoutedAgent):
@message_handler
async def handle_message(
self,
message: TextMessage,
ctx: MessageContext
) -> TextMessage:
# Process message and respond
response = f"Received: {message.content}"
return TextMessage(content=response, source=self.id.type)
The Core layer enables distributed agents that can run across different processes or machines, communicating through gRPC.
Built on Core, AgentChat provides high-level abstractions for common multi-agent patterns:
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
# Create a model client
model_client = OpenAIChatCompletionClient(model="gpt-4o")
# Create an agent
assistant = AssistantAgent(
name="Assistant",
model_client=model_client,
system_message="You are a helpful AI assistant."
)
Most applications will use AgentChat, dropping to Core only when needing custom agent types or distributed deployment.
Extensions provide integrations with external services:
The workhorse of AutoGen, AssistantAgent wraps an LLM with optional tools:
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
# Define tools as async functions
async def search_database(query: str) -> str:
"""Search the product database for matching items."""
# Your implementation
return f"Found 3 products matching '{query}'"
async def calculate_shipping(
weight: float,
destination: str
) -> str:
"""Calculate shipping cost based on weight and destination."""
base_rate = 5.0
per_kg = 2.5
cost = base_rate + (weight * per_kg)
return f"Shipping to {destination}: ${cost:.2f}"
# Create agent with tools
model_client = OpenAIChatCompletionClient(model="gpt-4o")
sales_agent = AssistantAgent(
name="SalesAssistant",
model_client=model_client,
tools=[search_database, calculate_shipping],
system_message="""You are a sales assistant.
Help customers find products and calculate shipping costs.
Always provide accurate pricing information."""
)
Represents human participants in the conversation:
from autogen_agentchat.agents import UserProxyAgent
# Human-in-the-loop agent
human = UserProxyAgent(
name="User",
description="A human user who provides requirements and feedback"
)
UserProxyAgent can request input from actual humans, enabling hybrid human-AI workflows.
Executes code generated by other agents:
from autogen_agentchat.agents import CodeExecutorAgent
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
# Create code executor (Docker recommended for safety)
code_executor = DockerCommandLineCodeExecutor(
image="python:3.11-slim",
timeout=60
)
executor_agent = CodeExecutorAgent(
name="CodeRunner",
code_executor=code_executor,
description="Executes Python code and returns results"
)
AutoGen’s power shines in multi-agent teams. Several built-in patterns cover common use cases. For an architectural overview of multi-agent patterns across frameworks, see Multi-Agent Collaboration Patterns.
Agents take turns in a fixed order:
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
async def main():
model_client = OpenAIChatCompletionClient(model="gpt-4o")
# Create specialized agents
writer = AssistantAgent(
name="Writer",
model_client=model_client,
system_message="""You are a technical writer.
Write clear, concise content."""
)
editor = AssistantAgent(
name="Editor",
model_client=model_client,
system_message="""You are an editor.
Review content for clarity and accuracy.
Say TERMINATE when satisfied."""
)
# Create team with termination condition
termination = TextMentionTermination("TERMINATE")
team = RoundRobinGroupChat(
[writer, editor],
termination_condition=termination,
max_turns=6
)
# Run the team
result = await team.run(
task="Write a brief explanation of microservices architecture."
)
print(result.messages[-1].content)
asyncio.run(main())
RoundRobinGroupChat works well for structured workflows where each agent has a clear role in sequence—like writer-editor or coder-reviewer pairs.
An LLM dynamically selects the next speaker based on context:
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination
async def main():
model_client = OpenAIChatCompletionClient(model="gpt-4o")
# Create specialized agents
researcher = AssistantAgent(
name="Researcher",
model_client=model_client,
system_message="You research and gather information.",
description="Expert at finding and synthesizing information"
)
analyst = AssistantAgent(
name="Analyst",
model_client=model_client,
system_message="You analyze data and draw conclusions.",
description="Expert at data analysis and interpretation"
)
presenter = AssistantAgent(
name="Presenter",
model_client=model_client,
system_message="""You create clear presentations.
Say TERMINATE when the presentation is complete.""",
description="Expert at presenting findings clearly"
)
# Selector chooses next speaker based on context
team = SelectorGroupChat(
[researcher, analyst, presenter],
model_client=model_client, # Model that selects speakers
termination_condition=TextMentionTermination("TERMINATE")
)
result = await team.run(
task="Research AI agent frameworks and present key findings."
)
The selector LLM reads the conversation and each agent’s description to decide who should speak next. This enables adaptive workflows that respond to the task’s needs.
Agents hand off tasks directly to each other without a central orchestrator:
from autogen_agentchat.teams import Swarm
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import HandoffMessage
async def main():
model_client = OpenAIChatCompletionClient(model="gpt-4o")
# First-line support agent
support = AssistantAgent(
name="Support",
model_client=model_client,
handoffs=["Billing", "Technical"],
system_message="""You are first-line support.
For billing questions, handoff to Billing.
For technical issues, handoff to Technical."""
)
billing = AssistantAgent(
name="Billing",
model_client=model_client,
handoffs=["Support"],
system_message="""You handle billing questions.
Handoff back to Support if issue is resolved."""
)
technical = AssistantAgent(
name="Technical",
model_client=model_client,
handoffs=["Support"],
system_message="""You handle technical issues.
Handoff back to Support if issue is resolved."""
)
# Swarm starts with first agent
team = Swarm(
[support, billing, technical],
termination_condition=TextMentionTermination("RESOLVED")
)
result = await team.run(
task="I'm having trouble with my subscription billing."
)
Swarm is ideal for customer service, triage systems, or any workflow where the path through specialists depends on the specific issue.
Microsoft’s Magentic-One is a sophisticated multi-agent system designed for complex web and file-based tasks:
from autogen_agentchat.teams import MagenticOneGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
async def main():
model_client = OpenAIChatCompletionClient(model="gpt-4o")
# MagenticOne includes specialized agents:
# - Orchestrator (manages the workflow)
# - WebSurfer (browses the web)
# - FileSurfer (navigates local files)
# - Coder (writes and debugs code)
# - ComputerTerminal (executes commands)
team = MagenticOneGroupChat(
model_client=model_client
)
result = await team.run(
task="Research the top 3 AI frameworks and save a comparison to report.md"
)
MagenticOne excels at open-ended tasks requiring multiple capabilities—web research, file manipulation, and code execution working together.
AutoGen treats code execution as a first-class concern with secure, configurable executors.
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_agentchat.agents import AssistantAgent, CodeExecutorAgent
# Secure Docker-based execution
docker_executor = DockerCommandLineCodeExecutor(
image="python:3.11-slim",
timeout=120,
work_dir="/workspace"
)
# Coder agent generates code
coder = AssistantAgent(
name="Coder",
model_client=model_client,
system_message="""You write Python code to solve problems.
Always include proper error handling.
Format code in markdown code blocks."""
)
# Executor runs the code safely
executor = CodeExecutorAgent(
name="Executor",
code_executor=docker_executor
)
# Team them together
team = RoundRobinGroupChat(
[coder, executor],
max_turns=10
)
Docker isolation protects your system from potentially harmful generated code.
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
# Only use in development with trusted code
local_executor = LocalCommandLineCodeExecutor(
timeout=60,
work_dir="./scratch"
)
AutoGen handles conversation history automatically within a session. For persistent memory across sessions:
from autogen_agentchat.agents import AssistantAgent
# Agents maintain conversation context within a team run
# For long conversations, consider summarization
summarizer = AssistantAgent(
name="Summarizer",
model_client=model_client,
system_message="""Periodically summarize the conversation
to keep context manageable. Focus on key decisions and
outstanding action items."""
)
For advanced memory patterns like vector stores or external databases, integrate through custom tools.
Always define clear termination conditions to prevent infinite loops:
from autogen_agentchat.conditions import (
TextMentionTermination,
MaxMessageTermination,
TimeoutTermination,
HandoffTermination
)
# Combine conditions with OR
combined = (
TextMentionTermination("TERMINATE") |
MaxMessageTermination(20) |
TimeoutTermination(300) # 5 minutes
)
team = SelectorGroupChat(
agents,
termination_condition=combined
)
Wrap team execution with proper error handling:
import asyncio
from autogen_agentchat.base import TaskResult
async def run_with_retry(
team,
task: str,
max_retries: int = 3
) -> TaskResult:
for attempt in range(max_retries):
try:
result = await asyncio.wait_for(
team.run(task=task),
timeout=600 # 10 minute timeout
)
return result
except asyncio.TimeoutError:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt) # Exponential backoff
raise RuntimeError("Max retries exceeded")
Enable tracing for debugging and monitoring:
import os
# Enable LangSmith tracing (AutoGen supports it)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
# Or use AutoGen's built-in logging
import logging
logging.basicConfig(level=logging.INFO)
For better UX, stream responses as they’re generated:
from autogen_agentchat.ui import Console
async def main():
team = RoundRobinGroupChat([agent1, agent2])
# Console helper streams output
await Console(team.run_stream(task="Your task here"))
AutoGen excels in scenarios requiring:
Multi-Agent Collaboration: Tasks that naturally decompose into specialist roles—research, analysis, writing, review.
Dynamic Workflows: When the path through agents depends on the task content, not just predefined routes.
Human-in-the-Loop: Workflows where humans need to participate, review, or guide agent decisions.
Code Generation and Execution: The built-in code executor integration handles this common need securely.
Complex Problem Solving: Open-ended tasks requiring multiple capabilities working together.
Consider alternatives when:
AutoGen brings a unique perspective to AI agents: coordination through conversation. This natural approach enables sophisticated multi-agent systems where specialists collaborate dynamically, humans participate seamlessly, and complex tasks decompose organically.
The framework’s layered architecture—Core for advanced scenarios, AgentChat for rapid development—means you can start simple and scale up. Team patterns like SelectorGroupChat and Swarm cover common collaboration needs, while MagenticOne demonstrates what’s possible with specialized agents working together.
For teams building complex AI systems that require multiple perspectives or capabilities, AutoGen provides a thoughtful, well-designed foundation that makes multi-agent development accessible without sacrificing power.
This post is part of our Framework Deep Dive series, exploring the architectures and patterns of major AI agent frameworks. Next up: CrewAI Deep Dive.
An in-depth exploration of CrewAI's role-based architecture, crew orchestration patterns, task delegation, and production best practices for building collaborative AI agent teams
An in-depth exploration of LangChain's architecture, components, and best practices for building production-ready AI agents
A comprehensive comparison of Microsoft Semantic Kernel and LangChain for building AI agents, covering architecture, enterprise features, integration patterns, and when to use each framework