AutoGen: Microsoft's Multi-Agent Conversation Framework

AutoGen is an open-source framework from Microsoft Research that enables the development of LLM-based applications using a multi-agent conversation architecture. With AutoGen, you can create multiple AI agents that communicate, collaborate, and solve complex tasks automatically.

In this tutorial, we will learn how to use AutoGen from installation, creating various types of agents, building multi-agent conversations, to building a virtual software development team consisting of a planner, coder, and tester.

Why AutoGen?

Before we begin, let's understand why AutoGen has become a popular choice for building multi-agent systems:

• Multi-Agent Conversation: Agents can communicate with each other naturally
• Customizable: Each agent can be configured with different behaviors, tools, and LLMs
• Human-in-the-Loop: Supports human intervention in conversation flows
• Code Execution: Agents can write and execute code automatically
• Teachable Agents: Agents can learn from previous interactions
• Nested Chats: Supports hierarchical conversations for complex workflows

Installation

Prerequisites

Make sure you have Python 3.8 or later installed on your system.

Basic Installation

pip install pyautogen

Installation with Additional Features

# With Docker support for code execution
pip install pyautogen[docker]

With teachable agents support
pip install pyautogen[teachable]

Full installation
pip install pyautogen[docker,teachable]

LLM Configuration

AutoGen requires access to an LLM. Create a configuration file OAICONFIGLIST:

[
    {
        "model": "gpt-4",
        "apikey": "sk-your-openai-api-key"

    },
    {
        "model": "gpt-3.5-turbo",
        "apikey": "sk-your-openai-api-key"
    }
]

Or use environment variables:

import os
os.environ["OPENAIAPIKEY"] = "sk-your-openai-api-key"

Core Agent Concepts

ConversableAgent

import autogen

configlist = autogen.configlistfromjson("OAICONFIGLIST")

llmconfig = {

    "configlist": configlist,

    "temperature": 0,
}

Create a basic ConversableAgent
agent = autogen.ConversableAgent(
    name="basicagent",
    llmconfig=llmconfig,
    systemmessage="You are a helpful assistant.",

    humaninputmode="NEVER",

)

Important humaninputmode parameters:

assistant = autogen.AssistantAgent(
    name="assistant",
    llmconfig=llmconfig,

    systemmessage="""You are an AI assistant that helps complete tasks.
    Provide clear and structured solutions.
    Reply with 'TERMINATE' when the task is complete."""
)

UserProxyAgent

userproxy = autogen.UserProxyAgent(
    name="userproxy",
    humaninputmode="TERMINATE",
    maxconsecutiveautoreply=10,

    isterminationmsg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),

    codeexecutionconfig={

        "workdir": "coding",
        "usedocker": False,  # Set True for security

    },
)

import autogen

configlist = autogen.configlistfromjson("OAICONFIGLIST")


assistant = autogen.AssistantAgent(
    name="assistant",
    llmconfig={"configlist": configlist},
    systemmessage="You are an expert Python programmer. Write clean, well-documented code."

)

userproxy = autogen.UserProxyAgent(
    name="userproxy",

    humaninputmode="NEVER",

    maxconsecutiveautoreply=5,
    isterminationmsg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    codeexecutionconfig={"workdir": "coding", "usedocker": False},
)

Start the conversation
userproxy.initiatechat(
    assistant,
    message="Create a Python function that calculates the Fibonacci sequence both recursively and iteratively, then compare their performance."
)

Group Chat: Multi-Agent Conversations

Group Chat allows more than two agents to communicate in a single session.

Creating a GroupChat

import autogen

configlist = autogen.configlistfromjson("OAICONFIGLIST")
llmconfig = {"configlist": configlist, "temperature": 0}


Planner Agent
planner = autogen.AssistantAgent(
    name="planner",
    llmconfig=llmconfig,

    systemmessage="""You are a project planner. Your responsibilities:
Analyze project requirements
Create step-by-step implementation plans
Ensure all aspects are covered
    Do not write code, focus on planning."""

)

Critic Agent
critic = autogen.AssistantAgent(
    name="critic",
    llmconfig=llmconfig,
    systemmessage="""You are a code reviewer. Your responsibilities:

Review written code
Suggest improvements
Ensure best practices are followed
    Provide constructive feedback."""

)

Coder Agent
coder = autogen.AssistantAgent(
    name="coder",
    llmconfig=llmconfig,

    systemmessage="""You are a senior programmer. Your responsibilities:
Write code based on the plan
Implement features according to specifications
Fix code based on reviews"""
)


User Proxy
userproxy = autogen.UserProxyAgent(

    name="admin",
    humaninputmode="NEVER",

    maxconsecutiveautoreply=10,
    codeexecutionconfig={"workdir": "groupcoding", "usedocker": False},

)

# Create GroupChat
groupchat = autogen.GroupChat(
    agents=[userproxy, planner, coder, critic],
    messages=[],
    maxround=20,

    speakerselectionmethod="auto",  # LLM selects the next speaker

)

Create Manager
manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llmconfig=llmconfig,

)

Start group chat
userproxy.initiatechat(

    manager,
    message="Build a simple REST API using FastAPI for book management (CRUD operations)."
)

# Round-robin: takes turns
groupchatrr = autogen.GroupChat(
    agents=[userproxy, planner, coder, critic],

    messages=[],
    maxround=20,
    speakerselectionmethod="roundrobin",

)

Manual: user selects
groupchatmanual = autogen.GroupChat(
    agents=[userproxy, planner, coder, critic],

    messages=[],
    maxround=20,
    speakerselectionmethod="manual",
)

Custom function
def customspeakerselection(lastspeaker, groupchat):

    if lastspeaker is userproxy:

        return planner
    elif lastspeaker is planner:
        return coder
    elif lastspeaker is coder:

        return critic
    else:
        return userproxy

groupchatcustom = autogen.GroupChat(

    agents=[userproxy, planner, coder, critic],
    messages=[],
    maxround=20,

    speakerselectionmethod=customspeakerselection,

)

userproxy = autogen.UserProxyAgent(
    name="executor",
    codeexecutionconfig={
        "workdir": "workspace",

        "usedocker": False,
        "timeout": 60,
        "lastnmessages": 3,
    },
)

Docker Execution

userproxy = autogen.UserProxyAgent(
    name="executor",
    codeexecutionconfig={

        "workdir": "workspace",
        "usedocker": "python:3.11-slim",

        "timeout": 120,
    },
)

from autogen.coding import LocalCommandLineCodeExecutor

executor = LocalCommandLineCodeExecutor(
    timeout=60,
    workdir="myworkspace",

)

userproxy = autogen.UserProxyAgent(
    name="executor",
    codeexecutionconfig={"executor": executor},
)

Tool Use (Function Calling)

Agents can use tools/functions to accomplish tasks.

Registering Tools

from typing import Annotated

Define functions
def getweather(

    city: Annotated[str, "Name of the city to check weather"]
) -> str:
    """Get weather information for a specific city."""
    weatherdata = {
        "New York": "Sunny, 28°C",
        "London": "Cloudy, 18°C",
        "Tokyo": "Light rain, 22°C",
    }
    return weatherdata.get(city, f"Weather data for {city} is not available")


def calculatebmi(
    weight: Annotated[float, "Body weight in kg"],
    height: Annotated[float, "Height in meters"]
) -> str:
    """Calculate Body Mass Index (BMI)."""
    bmi = weight / (height  2)

    if bmi < 18.5:
        category = "Underweight"
    elif bmi < 25:
        category = "Normal"
    elif bmi < 30:
        category = "Overweight"
    else:
        category = "Obese"
    return f"BMI: {bmi:.1f} - Category: {category}"

Register with agents
assistant = autogen.AssistantAgent(
    name="assistant",
    llmconfig=llmconfig,
)

userproxy = autogen.UserProxyAgent(

    name="userproxy",
    humaninputmode="NEVER",
    codeexecutionconfig=False,  # Disable code execution
)

Register tools
autogen.registerfunction(

    getweather,
    caller=assistant,
    executor=userproxy,

    name="getweather",
    description="Get weather information for a specific city"
)

autogen.registerfunction(

    calculatebmi,
    caller=assistant,
    executor=userproxy,

    name="calculatebmi",
    description="Calculate Body Mass Index"
)

Use tools
userproxy.initiatechat(
    assistant,
    message="What's the weather like in Tokyo today? And please calculate my BMI: weight 70kg, height 1.75m"
)

# Always request human approval
agentalways = autogen.UserProxyAgent(

    name="humanalways",
    humaninputmode="ALWAYS",
    codeexecutionconfig={"workdir": "coding"},

)

Request input on termination
agentterminate = autogen.UserProxyAgent(
    name="humanterminate",

    humaninputmode="TERMINATE",

    maxconsecutiveautoreply=5,
    isterminationmsg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

No human intervention
agentnever = autogen.UserProxyAgent(

    name="humannever",
    humaninputmode="NEVER",
    maxconsecutiveautoreply=10,

)

def istaskcomplete(msg):
    content = msg.get("content", "")
    if content:
        return "TASKCOMPLETE" in content or "TERMINATE" in content
    return False

userproxy = autogen.UserProxyAgent(

    name="user",
    humaninputmode="TERMINATE",

    isterminationmsg=istaskcomplete,

    maxconsecutiveautoreply=15,
)

import autogen

configlist = autogen.configlistfromjson("OAICONFIGLIST")
llmconfig = {"configlist": configlist}


Main agent
writer = autogen.AssistantAgent(
    name="writer",
    llmconfig=llmconfig,

    systemmessage="You are a technical writer. Write content based on the given outline."
)

Reviewer agent for nested chat
reviewer = autogen.AssistantAgent(
    name="reviewer",
    llmconfig=llmconfig,
    systemmessage="""You are an editor. Review the writing and provide feedback.

    Format: [APPROVED] if it's good, or [REVISION NEEDED] with detailed corrections."""
)

SEO agent for nested chat
seooptimizer = autogen.AssistantAgent(
    name="seooptimizer",

    llmconfig=llmconfig,

    systemmessage="You are an SEO specialist. Optimize content for search engines."
)

userproxy = autogen.UserProxyAgent(

    name="user",
    humaninputmode="NEVER",

    maxconsecutiveautoreply=3,
    codeexecutionconfig=False,
)

Configure nested chats
nestedchatconfig = [
    {
        "recipient": reviewer,
        "message": "Review the following writing and provide feedback:\n\n",
        "summarymethod": "lastmsg",
        "maxturns": 2,

    },
    {
        "recipient": seooptimizer,
        "message": "Optimize the following content for SEO:\n\n",
        "summarymethod": "lastmsg",
        "maxturns": 2,

    },
]

Register nested chats
userproxy.registernestedchats(
    nestedchatconfig,
    trigger=writer,
)

Start conversation
userproxy.initiatechat(
    writer,
    message="Write a short article about the benefits of using Python for data science."
)

from autogen.agentchat.contrib.capabilities.teachability import Teachability
from autogen import ConversableAgent

configlist = autogen.configlistfromjson("OAICONFIGLIST")
llmconfig = {"configlist": configlist}


Create a teachable agent
teachableagent = ConversableAgent(
    name="teachableassistant",

    llmconfig=llmconfig,

    systemmessage="You are an assistant that can learn from conversations.",
)

Add teachability capability
teachability = Teachability(
    resetdb=False,  # Set True to reset memory

    pathtodbdir="./teachabledb",

    verbosity=0,
    recallthreshold=1.5,
)

teachability.addtoagent(teachableagent)


User proxy
user = autogen.UserProxyAgent(
    name="user",
    humaninputmode="ALWAYS",

    codeexecutionconfig=False,

)

First conversation - teaching information
user.initiatechat(
    teachableagent,

    message="Remember that our internal service API keys use the prefix 'INT-' followed by 12 alphanumeric characters."
)

Second conversation - agent recalls information
user.initiatechat(
    teachableagent,

    message="What's the format for our internal service API keys?"
)

import autogen

configlist = autogen.configlistfromjson("OAICONFIGLIST")

llmconfig = {"configlist": configlist, "temperature": 0}

1. Project Planner
planner = autogen.AssistantAgent(
    name="planner",
    llmconfig=llmconfig,
    systemmessage="""You are a Senior Project Planner. Your responsibilities:

Analyze feature requirements from the user
Break them down into structured tasks
Create technical specifications for each task
Determine implementation order

    Output format:
    ## Requirements Analysis
    [description]

    ## Task Breakdown
Task 1: [description]
Task 2: [description]
    

    ## Technical Specifications
    [technical details]

    After the plan is approved, hand off to the coder for implementation.
    Do not write any code."""
)

2. Senior Developer
coder = autogen.AssistantAgent(
    name="coder",
    llmconfig=llmconfig,

    systemmessage="""You are a Senior Python Developer. Your responsibilities:
Implement code based on the planner's plan
Write clean, documented code following best practices
Use type hints and docstrings
Fix code based on tester feedback

    Always write code in complete, runnable Python blocks.
    Include all necessary imports."""
)

3. QA Tester
tester = autogen.AssistantAgent(
    name="tester",
    llmconfig=llmconfig,
    systemmessage="""You are a Senior QA Engineer. Your responsibilities:

Review code written by the coder
Write unit tests using pytest
Identify bugs, edge cases, and potential issues
Ensure code meets the planner's specifications

    Feedback format:
    ## Code Review
    [review findings]

    ## Unit Tests
    [test code]

    ## Issues Found
    [list of issues]

    If all tests pass and there are no critical issues, reply with 'APPROVED - TERMINATE'"""
)

4. User Proxy (code executor)
executor = autogen.UserProxyAgent(
    name="executor",
    humaninputmode="NEVER",

    maxconsecutiveautoreply=15,
    isterminationmsg=lambda x: "TERMINATE" in (x.get("content", "") or ""),
    codeexecutionconfig={
        "workdir": "devproject",
        "usedocker": False,

        "timeout": 120,
    },
)

# Define conversation flow
def devteamspeakerselection(lastspeaker, groupchat):

    messages = groupchat.messages

    if lastspeaker is executor:
        # After user gives a task, planner plans
        if len(messages) <= 2:
            return planner
        # After code execution, tester reviews
        return tester
    elif lastspeaker is planner:

        # After planning, coder implements
        return coder
    elif lastspeaker is coder:
        # After coding, executor runs the code
        return executor
    elif lastspeaker is tester:

        # If there are issues, go back to coder
        lastmsg = messages[-1].get("content", "")
        if "APPROVED" in lastmsg:

            return executor  # Done
        return coder  # Fix issues

    return planner

Create group chat
devgroupchat = autogen.GroupChat(
    agents=[executor, planner, coder, tester],
    messages=[],
    maxround=25,

    speakerselectionmethod=devteamspeakerselection,
)

devmanager = autogen.GroupChatManager(

    groupchat=devgroupchat,
    llmconfig=llmconfig,
)

# Assign a task to the team
task = """
Build a Task Manager module with the following features:
TaskManager class that can add, remove, and display tasks
Each task has: id, title, description, priority (low/medium/high), status (todo/inprogress/done), createdat
Filter tasks by priority and status
Sort tasks by priority or createdat

Export tasks to JSON format
Input validation (title cannot be empty, priority must be valid, etc.)
"""

executor.initiatechat(
    devmanager,

    message=task,
)

import json
from datetime import datetime
from typing import Optional
from enum import Enum
from dataclasses import dataclass, field, asdict

class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class Status(Enum):
    TODO = "todo"
    INPROGRESS = "inprogress"

    DONE = "done"

@dataclass
class Task:
    id: int
    title: str
    description: str
    priority: Priority
    status: Status = Status.TODO
    createdat: datetime = field(defaultfactory=datetime.now)


    def todict(self):
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "priority": self.priority.value,
            "status": self.status.value,
            "createdat": self.createdat.isoformat(),
        }

class TaskManager:
    def init(self):
        self.tasks: dict[int, Task] = {}

        self.nextid: int = 1


    def addtask(
        self,
        title: str,
        description: str = "",
        priority: str = "medium"
    ) -> Task:
        if not title or not title.strip():
            raise ValueError("Title cannot be empty")

        try:
            pri = Priority(priority.lower())
        except ValueError:
            raise ValueError(
                f"Invalid priority: {priority}. "
                f"Use: {[p.value for p in Priority]}"
            )

        task = Task(
            id=self.nextid,
            title=title.strip(),
            description=description.strip(),
            priority=pri,
        )
        self.tasks[self.nextid] = task

        self.nextid += 1

        return task

    def removetask(self, taskid: int) -> bool:

        if taskid not in self.tasks:

            raise KeyError(f"Task with ID {taskid} not found")
        del self.tasks[taskid]
        return True

    def gettask(self, taskid: int) -> Task:
        if taskid not in self.tasks:
            raise KeyError(f"Task with ID {taskid} not found")

        return self.tasks[taskid]


    def updatestatus(self, taskid: int, status: str) -> Task:

        task = self.gettask(taskid)

        try:
            task.status = Status(status.lower())
        except ValueError:
            raise ValueError(
                f"Invalid status: {status}. "
                f"Use: {[s.value for s in Status]}"
            )
        return task

    def listtasks(
        self,
        priority: Optional[str] = None,
        status: Optional[str] = None,
    ) -> list[Task]:
        tasks = list(self.tasks.values())


        if priority:
            try:
                pri = Priority(priority.lower())
                tasks = [t for t in tasks if t.priority == pri]
            except ValueError:
                raise ValueError(f"Invalid priority: {priority}")

        if status:
            try:
                st = Status(status.lower())
                tasks = [t for t in tasks if t.status == st]
            except ValueError:
                raise ValueError(f"Invalid status: {status}")

        return tasks

    def sorttasks(
        self,
        by: str = "createdat",

        reverse: bool = False
    ) -> list[Task]:
        tasks = list(self.tasks.values())
        priorityorder = {Priority.HIGH: 3, Priority.MEDIUM: 2, Priority.LOW: 1}


        if by == "priority":
            tasks.sort(key=lambda t: priorityorder[t.priority], reverse=reverse)
        elif by == "createdat":

            tasks.sort(key=lambda t: t.createdat, reverse=reverse)
        else:
            raise ValueError(f"Sort by '{by}' not supported. Use: priority, createdat")


        return tasks

    def exportjson(self, filepath: Optional[str] = None) -> str:
        data = [task.todict() for task in self.tasks.values()]
        jsonstr = json.dumps(data, indent=2, ensureascii=False)

        if filepath:
            with open(filepath, "w", encoding="utf-8") as f:
                f.write(jsonstr)


        return jsonstr

Usage example
if name == "main":
    tm = TaskManager()

    # Add tasks
    tm.addtask("Setup database", "Configure PostgreSQL", "high")

    tm.addtask("Create API endpoint", "REST API for user management", "medium")
    tm.addtask("Write documentation", "API documentation", "low")


    # Filter and sort
    highpriority = tm.listtasks(priority="high")

    sortedtasks = tm.sorttasks(by="priority", reverse=True)


    # Export
    print(tm.exportjson())

Tips and Best Practices

1. Clear System Messages

Write specific and structured system messages for each agent. The clearer the instructions, the better the results.

2. Limit Auto-Replies

Always set maxconsecutiveautoreply to prevent infinite loops.

userproxy = autogen.UserProxyAgent(
    name="user",
    maxconsecutiveautoreply=10,  # Limit to 10 iterations

)

codeexecutionconfig={
    "usedocker": "python:3.11-slim",
    "timeout": 120,
}

4. Logging for Debugging

import autogen

Enable logging
autogen.runtimelogging.start(loggertype="file", config={"filename": "autogen.log"})

Run conversations
...

Stop logging
autogen.runtimelogging.stop()

# Use caching to avoid repeated API calls
with autogen.Cache.disk(cacheseed=42) as cache:
    userproxy.initiatechat(
        assistant,
        message="Write a hello world program",
        cache=cache,
    )

Conclusion

AutoGen from Microsoft provides a powerful framework for building multi-agent conversation systems. With features like group chat, code execution, tool use, and teachable agents, you can build complex and collaborative AI applications.

Keys to success with AutoGen:

• Define agent roles clearly through system messages
• Choose the right speaker selection method
• Leverage code execution for automatic verification
• Use human-in-the-loop for quality control
• Implement nested chats for hierarchical workflows

With these concepts mastered, you are ready to build multi-agent systems that can solve complex tasks collaboratively, from software development to data analysis and business automation.