AutoGen — Microsoft Multi-Agent Framework Complete Guide (2026)

pip install pyautogen
# For code execution support
pip install pyautogen[jupyter-executor]

import autogen
 
# LLM configuration
config_list = [
    {
        "model": "gpt-4o",
        "api_key": "your-openai-api-key",
        "temperature": 0.1,
    }
]
 
llm_config = {
    "config_list": config_list,
    "cache_seed": 42,          # Reproducible responses during development
    "timeout": 120,
    "max_tokens": 4096,
}

import autogen
 
# The AI assistant that writes solutions
assistant = autogen.AssistantAgent(
    name="PythonExpert",
    system_message="""You are an expert Python developer.
When asked to solve a problem:
1. Write clean, well-commented Python code
2. Include error handling
3. After the code, explain what it does in 2-3 sentences
Always use code blocks with the python tag.""",
    llm_config=llm_config,
)
 
# The proxy that runs code and relays results
user_proxy = autogen.UserProxyAgent(
    name="CodeRunner",
    human_input_mode="NEVER",      # Fully automated
    max_consecutive_auto_reply=5,  # Prevents infinite loops
    is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
    code_execution_config={
        "work_dir": "/tmp/autogen_workspace",
        "use_docker": False,        # Set True in production for isolation
        "timeout": 60,
    },
    llm_config=False,              # UserProxy doesn't need an LLM
)
 
# Run the conversation
user_proxy.initiate_chat(
    assistant,
    message="""Write a Python function that:
1. Takes a list of stock prices (floats)
2. Calculates: mean, standard deviation, and Sharpe ratio (assume risk-free rate of 2%)
3. Returns a dict with all three metrics
Then test it with sample data: [100, 102, 98, 105, 103, 101, 107]
Reply TERMINATE when done."""
)

def build_software_team(llm_config: dict) -> tuple:
    """Create a software development team of agents."""
 
    product_manager = autogen.AssistantAgent(
        name="ProductManager",
        system_message="""You are a product manager.
Your role: Define requirements clearly, prioritize features, and ensure the solution meets user needs.
Always start your messages with 'PM:' and end with a specific question or action item for the team.""",
        llm_config=llm_config,
    )
 
    developer = autogen.AssistantAgent(
        name="Developer",
        system_message="""You are a senior Python developer.
Your role: Write clean, efficient, production-ready Python code based on requirements.
Always start messages with 'DEV:'.
Write all code in properly tagged ```python blocks.""",
        llm_config=llm_config,
    )
 
    code_reviewer = autogen.AssistantAgent(
        name="CodeReviewer",
        system_message="""You are a code reviewer focused on quality and security.
Your role: Review code for bugs, security issues, performance problems, and style.
Always start messages with 'REVIEW:'.
Format findings as: [CRITICAL], [WARNING], or [SUGGESTION].""",
        llm_config=llm_config,
    )
 
    user_proxy = autogen.UserProxyAgent(
        name="Executor",
        human_input_mode="NEVER",
        max_consecutive_auto_reply=15,
        is_termination_msg=lambda msg: "APPROVED" in msg.get("content", ""),
        code_execution_config={
            "work_dir": "/tmp/team_workspace",
            "use_docker": False,
        },
    )
 
    groupchat = autogen.GroupChat(
        agents=[user_proxy, product_manager, developer, code_reviewer],
        messages=[],
        max_round=20,
        speaker_selection_method="auto",  # LLM selects next speaker
    )
 
    manager = autogen.GroupChatManager(
        groupchat=groupchat,
        llm_config=llm_config,
    )
 
    return user_proxy, manager
 
user_proxy, manager = build_software_team(llm_config)
 
user_proxy.initiate_chat(
    manager,
    message="""Build a URL shortener service in Python.
Requirements: store URLs in memory, generate 6-character codes, handle collisions.
The code reviewer must approve before we finish. Reply APPROVED when the final version is ready."""
)

import requests
from typing import Annotated
 
# Define tool functions with type annotations and docstrings
def search_web(
    query: Annotated[str, "The search query"],
    max_results: Annotated[int, "Maximum number of results to return"] = 5
) -> str:
    """Search the web and return results as formatted text."""
    # Mock implementation — replace with real search API
    return f"Search results for '{query}':\n1. Result 1\n2. Result 2\n3. Result 3"
 
def read_file(
    filepath: Annotated[str, "Absolute path to the file to read"]
) -> str:
    """Read and return the contents of a file."""
    try:
        with open(filepath, "r") as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File not found: {filepath}"
    except Exception as e:
        return f"Error reading file: {e}"
 
def write_file(
    filepath: Annotated[str, "Absolute path to write the file"],
    content: Annotated[str, "Content to write"]
) -> str:
    """Write content to a file, creating it if it does not exist."""
    try:
        with open(filepath, "w") as f:
            f.write(content)
        return f"Successfully wrote {len(content)} characters to {filepath}"
    except Exception as e:
        return f"Error writing file: {e}"
 
# Register tools with an agent
research_agent = autogen.AssistantAgent(
    name="Researcher",
    system_message="You research topics using available tools and summarize findings.",
    llm_config=llm_config,
)
 
executor = autogen.UserProxyAgent(
    name="ToolExecutor",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=8,
    is_termination_msg=lambda msg: "DONE" in msg.get("content", ""),
)
 
# Register tools on both agents (caller and executor)
autogen.register_function(
    search_web,
    caller=research_agent,
    executor=executor,
    name="search_web",
    description="Search the web for information on a given topic"
)

# Human approval before code execution
supervised_proxy = autogen.UserProxyAgent(
    name="HumanSupervisor",
    human_input_mode="ALWAYS",    # Prompt human for every agent response
    max_consecutive_auto_reply=0, # Never auto-reply
    code_execution_config={
        "work_dir": "/tmp/supervised",
        "use_docker": True,        # Always use Docker for human-supervised code
    }
)
 
# Or conditional human input
conditional_proxy = autogen.UserProxyAgent(
    name="ConditionalSupervisor",
    human_input_mode="TERMINATE",  # Only ask human when agent says TERMINATE
    max_consecutive_auto_reply=10,
)