Agent Orchestration with Microsoft AutoGen 0.4
Microsoft's AutoGen was largely rewritten for version 0.4, released in late 2024. The new architecture drops the conversation-centric model of earlier versions in favor of an event-driven, modular design that makes it easier to build, test, and scale multi-agent systems. If you used AutoGen before, you will find that 0.4 is essentially a new framework built on lessons from the original.
Below I will walk through the core concepts of AutoGen 0.4 and build a multi-agent system that handles research and content generation.
What changed in AutoGen 0.4
The biggest architectural shift is the introduction of the Core API and AgentChat API as separate layers. The Core API provides low-level primitives: agents, runtimes, and message passing. The AgentChat API builds on top of it with higher-level abstractions like teams, terminators, and preset agent types.
Other key changes include:
- Event-driven messaging replaces direct function calls between agents
- Runtimes manage agent lifecycle and message delivery
- Teams provide structured multi-agent orchestration patterns
- Type-safe messages replace unstructured dictionaries
Getting started
Install AutoGen 0.4 with the OpenAI extension:
pip install autogen-agentchat==0.4 autogen-ext[openai]==0.4
Building agents with AgentChat
The AgentChat API provides AssistantAgent, the workhorse for most use cases. Each agent has a system message, a model client, and optional tools.
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
model_client = OpenAIChatCompletionClient(
model="gpt-4o",
)
research_agent = AssistantAgent(
name="researcher",
model_client=model_client,
system_message=(
"You are a research specialist. Your job is to gather "
"information, analyze data, and provide detailed findings "
"on the given topic. Always cite your reasoning and be "
"thorough in your analysis."
),
)
writer_agent = AssistantAgent(
name="writer",
model_client=model_client,
system_message=(
"You are a technical writer. Take research findings and "
"transform them into clear, well-structured content. "
"Use headings, bullet points, and examples to make "
"complex topics accessible."
),
)
Each agent is self-contained with its own personality and capabilities. The model_client abstraction means you can swap between OpenAI, Azure OpenAI, or other providers without changing agent logic.
Adding tools to agents
Agents become much more powerful when they can execute tools. In AutoGen 0.4, tools are plain Python functions decorated for discovery.
import httpx
from autogen_core import CancellationToken
async def search_web(
query: str, max_results: int = 5
) -> str:
"""Search the web for information on a topic.
Args:
query: The search query string.
max_results: Maximum number of results to return.
Returns:
A formatted string of search results.
"""
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.search-provider.com/search",
params={"q": query, "limit": max_results},
)
data = response.json()
results = []
for item in data.get("results", []):
results.append(f"- {item['title']}: {item['snippet']}")
return "\n".join(results) if results else "No results found."
async def analyze_sentiment(text: str) -> str:
"""Analyze the sentiment of a given text.
Args:
text: The text to analyze.
Returns:
A sentiment analysis summary.
"""
# Simplified analysis for demonstration
positive_words = {"good", "great", "excellent", "positive", "growth"}
negative_words = {"bad", "poor", "decline", "negative", "risk"}
words = set(text.lower().split())
pos_count = len(words & positive_words)
neg_count = len(words & negative_words)
if pos_count > neg_count:
return f"Sentiment: Positive (score: {pos_count}/{pos_count + neg_count})"
elif neg_count > pos_count:
return f"Sentiment: Negative (score: {neg_count}/{pos_count + neg_count})"
return "Sentiment: Neutral"
# Create an agent with tools
research_agent = AssistantAgent(
name="researcher",
model_client=model_client,
tools=[search_web, analyze_sentiment],
system_message=(
"You are a research specialist with access to web search "
"and sentiment analysis tools. Use them when needed to "
"provide comprehensive research findings."
),
)
The function docstrings and type hints are used to generate the tool schema that the LLM sees. This is why clear, descriptive docstrings matter so much in AutoGen 0.4.
Team-based orchestration
The real power of AutoGen 0.4 lies in its team abstractions. A RoundRobinGroupChat cycles through agents in order, while a SelectorGroupChat lets the model decide which agent should respond next.
from autogen_agentchat.teams import RoundRobinGroupChat, SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
# Simple round-robin team
termination = MaxMessageTermination(max_messages=6)
round_robin_team = RoundRobinGroupChat(
participants=[research_agent, writer_agent],
termination_condition=termination,
)
For more dynamic orchestration, use SelectorGroupChat where the LLM decides which agent should act next:
critic_agent = AssistantAgent(
name="critic",
model_client=model_client,
system_message=(
"You are a content critic. Review the writer's output for "
"accuracy, clarity, and completeness. Provide specific, "
"actionable feedback. When the content meets high quality "
"standards, respond with APPROVE."
),
)
termination = TextMentionTermination("APPROVE") | MaxMessageTermination(
max_messages=10
)
selector_team = SelectorGroupChat(
participants=[research_agent, writer_agent, critic_agent],
model_client=model_client,
termination_condition=termination,
selector_prompt=(
"Select the next agent based on the conversation state:\n"
"- 'researcher' if more information is needed\n"
"- 'writer' if research is complete and content needs writing or revising\n"
"- 'critic' if content has been written and needs review\n"
"Only return the agent name."
),
)
The selector_prompt guides the model in choosing which agent should respond next. Combined with termination conditions, you get a flexible workflow that runs until the critic approves the output or a maximum message count is reached.
Running teams
Running a team is straightforward with the async API:
import asyncio
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
async def run_content_pipeline(topic: str) -> str:
result = await selector_team.run(
task=f"Research and write a comprehensive analysis on: {topic}",
cancellation_token=CancellationToken(),
)
# Extract the final message
final_output = ""
for message in result.messages:
if message.source == "writer":
final_output = message.content
return final_output
async def run_with_streaming(topic: str) -> None:
stream = selector_team.run_stream(
task=f"Research and write about: {topic}",
cancellation_token=CancellationToken(),
)
async for event in stream:
if hasattr(event, "source") and hasattr(event, "content"):
print(f"\n[{event.source}]: {event.content[:200]}...")
# Execute
asyncio.run(run_content_pipeline("The impact of edge AI on IoT applications"))
The run_stream method is particularly useful for building interactive applications where you want to show the user what each agent is doing in real time.
Custom agent types
When the built-in agents are not enough, you can create custom agents by extending BaseChatAgent:
from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.base import Response
from autogen_agentchat.messages import TextMessage
class DataAnalystAgent(BaseChatAgent):
def __init__(self, name: str, data_source: str):
super().__init__(name=name, description="Analyzes data from various sources")
self._data_source = data_source
@property
def produced_message_types(self) -> list[type]:
return [TextMessage]
async def on_messages(self, messages, cancellation_token):
last_message = messages[-1].content if messages else ""
# Custom logic: query data source, run analysis, etc.
analysis = await self._run_analysis(last_message)
return Response(
chat_message=TextMessage(
content=analysis,
source=self.name,
)
)
async def on_reset(self, cancellation_token):
pass
async def _run_analysis(self, query: str) -> str:
# Placeholder for actual data analysis logic
return f"Analysis of '{query}' from {self._data_source}: [results here]"
Custom agents give you full control over message handling, which is essential for integrating with external systems, databases, or APIs that require specific interaction patterns.
Error handling and resilience
Production multi-agent systems need solid error handling. AutoGen 0.4 supports cancellation tokens and timeout patterns:
async def run_with_timeout(
team: SelectorGroupChat,
task: str,
timeout_seconds: int = 120,
) -> str:
cancellation_token = CancellationToken()
try:
result = await asyncio.wait_for(
team.run(task=task, cancellation_token=cancellation_token),
timeout=timeout_seconds,
)
return result.messages[-1].content
except asyncio.TimeoutError:
cancellation_token.cancel()
return "Task timed out. Partial results may be available."
except Exception as e:
cancellation_token.cancel()
return f"Error during execution: {str(e)}"
Key takeaways
AutoGen 0.4 is a big improvement over its predecessors. The separation of Core and AgentChat APIs gives you the right level of abstraction for your use case. Teams with selector-based orchestration let you build dynamic workflows where the LLM decides the execution path. And the tool system is clean and Pythonic.
If you are building multi-agent systems, AutoGen 0.4 is worth evaluating. The event-driven architecture, type-safe messaging, and flexible orchestration patterns make it a solid choice for production workloads.