Posted inAI

Building a Multi-Agent AI System with Microsoft AutoGen v0.4: Orchestrate Specialized Agents for Data Analysis and Code Generation

AI tutorial - IT technology blog
AI tutorial - IT technology blog

Context & Why Multi-Agent AI Systems Matter

Single-agent systems hit a wall fast when tasks get complex. Ask one LLM to plan a data pipeline, write pandas code, review that code for bugs, and then optimize it — the context window fills up, the model loses track, and output quality drops. Multi-agent systems exist precisely to break that bottleneck.

Microsoft AutoGen v0.4 restructures how AI agents collaborate. Unlike earlier versions, v0.4 introduces a clean event-driven architecture with a proper AgentChat high-level API. Each agent has a specific role and system prompt. A team coordinator decides who speaks next. The result mirrors how a real engineering team works: one person plans, another codes, a third reviews.

Once you understand how to wire agents together, the single-prompt ceiling stops being a constraint. You start thinking in workflows instead of prompts. That mental shift is what makes the pattern worth learning for anyone building AI-powered automation.

What follows is a concrete walkthrough: a three-agent team built for data analysis — a Planner, a Data Analyst, and a Code Reviewer. By the end, you’ll have a working system you can adapt for your own projects.

Installing AutoGen v0.4

Prerequisites

Before starting, make sure you have:

  • Python 3.10 or higher
  • An OpenAI API key (or any OpenAI-compatible endpoint like Azure OpenAI or LM Studio)
  • A terminal and basic familiarity with pip

Create a virtual environment first — this keeps dependencies isolated and avoids version conflicts:

python -m venv autogen-env
source autogen-env/bin/activate  # Windows: autogen-env\Scripts\activate

Installing the Packages

AutoGen v0.4 splits into modular packages. You need two core ones:

pip install autogen-agentchat autogen-ext[openai]

Verify the installation worked:

python -c "import autogen_agentchat; print(autogen_agentchat.__version__)"

You should see a version like 0.4.x. If you get an import error, double-check your virtual environment is activated.

Setting Up Your API Key

Store your API key as an environment variable — never hardcode it in your scripts:

# Linux/macOS
export OPENAI_API_KEY="sk-your-key-here"

# Windows PowerShell
$env:OPENAI_API_KEY="sk-your-key-here"

Configuring Your Agent Team

Name the file multi_agent.py. The full implementation below is around 70 lines — short enough to read in one sitting, but complete enough to run against a real dataset.

Step 1: Set Up the Model Client

The model client is what every agent uses to call the LLM. AutoGen v0.4 abstracts this cleanly so you define it once and share it across all agents:

import asyncio
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Shared model client — all agents use the same config
model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
)

Using gpt-4o-mini keeps costs low while testing. At roughly $0.15 per million input tokens, you can run dozens of test workflows before spending a dollar. Swap to gpt-4o when you need higher accuracy for production tasks.

Step 2: Create Specialized Agents

Each agent gets a focused system_message that defines its role. The tighter the role definition, the better the output quality:

# Agent 1: Breaks down the task into clear steps
planner_agent = AssistantAgent(
    name="Planner",
    model_client=model_client,
    system_message="""You are a project planner.
When given a data analysis task, break it into 3-5 concrete steps.
List each step clearly. Do not write any code — only outline the plan.
End your response with: 'Plan ready.'""",
)

# Agent 2: Writes Python code based on the plan
analyst_agent = AssistantAgent(
    name="DataAnalyst",
    model_client=model_client,
    system_message="""You are a data analyst who writes Python code.
You receive a plan and implement each step using pandas, numpy, or matplotlib.
Always include comments explaining what each code block does.
End your response with: 'Code ready.'""",
)

# Agent 3: Reviews code quality and suggests improvements
reviewer_agent = AssistantAgent(
    name="CodeReviewer",
    model_client=model_client,
    system_message="""You are a senior code reviewer.
Review the Python code provided for:
- Correctness and logic errors
- Code clarity and readability
- Edge cases not handled
Provide specific, actionable feedback.
If the code looks good, say: 'APPROVED' to end the session.""",
)

Step 3: Build the Team

RoundRobinGroupChat passes the conversation in a fixed order — Planner → DataAnalyst → CodeReviewer — then loops unless a termination condition fires:

# Stop when the reviewer says APPROVED
termination = TextMentionTermination("APPROVED")

# Build the team — agents speak in order
team = RoundRobinGroupChat(
    participants=[planner_agent, analyst_agent, reviewer_agent],
    termination_condition=termination,
    max_turns=9,  # Safety cap: max 3 rounds of 3 agents
)

The max_turns cap prevents runaway conversations if agents never reach the termination keyword. Nine turns gives three full Planner → Analyst → Reviewer cycles, which is usually enough for this workflow.

Step 4: Run the Team on a Real Task

async def main():
    task = """
    Analyze a CSV dataset of sales transactions with columns:
    date, product_name, quantity, unit_price, region.

    Calculate:
    1. Total revenue by region
    2. Top 5 best-selling products by quantity
    3. Monthly revenue trend for the last 6 months
    4. Generate a printed summary report with these statistics.

    Assume the CSV file is named 'sales_data.csv' in the current directory.
    """

    # Console streams each agent's response to your terminal in real time
    await Console(team.run_stream(task=task))

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

Run it from your terminal:

python multi_agent.py

Verification & Monitoring

Reading the Console Output

The Console wrapper streams each agent’s messages as they arrive. You’ll see labeled sections like this:

---------- Planner ----------
Step 1: Load the CSV file using pandas read_csv()...
Step 2: Add a revenue column (quantity * unit_price)...
Step 3: Group by region and sum revenue...
...
Plan ready.

---------- DataAnalyst ----------
import pandas as pd

# Load and prepare data
df = pd.read_csv('sales_data.csv')
df['revenue'] = df['quantity'] * df['unit_price']
...
Code ready.

---------- CodeReviewer ----------
The logic is correct. Suggest adding: if df.empty: raise ValueError.
APPROVED

Each section clearly shows which agent is speaking. If an agent produces bad output, the label tells you exactly which system message to tune.

Capturing Results Programmatically

Printing to the console is fine for development. When you move to production, capture the full message history in a variable so you can log it, store it, or trigger downstream actions based on the outcome:

async def main():
    task = "Your task here..."
    result = await team.run(task=task)

    for message in result.messages:
        print(f"[{message.source}] {message.content[:300]}")

    print(f"\nStop reason: {result.stop_reason}")
    print(f"Total messages: {len(result.messages)}")

The result.stop_reason field tells you whether the run ended cleanly (termination condition met) or hit the max_turns cap — useful for alerting when something went wrong.

Common Issues and Fixes

  • Agents loop without stopping: The termination keyword isn’t appearing. Check agent system messages — instruct agents explicitly to say the exact keyword at the right moment.
  • Rate limit errors: Add a delay between test runs, reduce max_turns, or switch to a higher-tier API plan.
  • Agent ignores its role: System messages need to be directive. Instead of “You can write code,” write “You must write Python code and nothing else.”
  • Output quality is inconsistent: Try upgrading from gpt-4o-mini to gpt-4o, especially for the CodeReviewer agent which requires more nuanced judgment.

Scaling Up: Dynamic Agent Selection

Once your three-agent team works reliably, try SelectorGroupChat — it uses a model to decide which agent speaks next rather than following a fixed order:

from autogen_agentchat.teams import SelectorGroupChat

dynamic_team = SelectorGroupChat(
    participants=[planner_agent, analyst_agent, reviewer_agent],
    model_client=model_client,
    termination_condition=termination,
)

This works well when your agents have overlapping responsibilities, or when which agent should speak depends on the context of the previous message. For straightforward sequential workflows like this data analysis pipeline, stick with RoundRobinGroupChat — it’s predictable and easier to debug.

At this point, you have a working multi-agent system that plans, codes, and reviews automatically. The natural next step is giving agents real tools — file I/O, web search, code execution — using AutoGen’s tool-calling support. That’s the point where the architecture stops being a toy and starts handling real production workloads.

Share:
Tags: