How to Build a Multi-Agent System with AutoGen (Group Chat)

pip install pyautogen openai python-dotenv

import os
from dotenv import load_dotenv
import autogen

load_dotenv()

# LLM configuration
llm_config = {
    "model": "gpt-4o",
    "api_key": os.environ["OPENAI_API_KEY"],
    "temperature": 0.1,
}

# Slightly higher temperature for creative agents
creative_llm_config = {
    "model": "gpt-4o",
    "api_key": os.environ["OPENAI_API_KEY"],
    "temperature": 0.4,
}

researcher = autogen.AssistantAgent(
    name="Researcher",
    system_message="""You are a technical researcher. Your job:
1. Analyze the user's request to identify what information is needed.
2. Provide accurate, well-sourced information relevant to the task.
3. Structure your findings clearly with headers and bullet points.
4. After providing findings, explicitly state: 'Research complete. Ready for implementation.'

Do NOT write code. That is the Coder's job.
Do NOT critique work. That is the Critic's job.
""",
    llm_config=llm_config,
)

coder = autogen.AssistantAgent(
    name="Coder",
    system_message="""You are an expert Python developer. Your job:
1. Wait for the Researcher to provide background information.
2. Write clean, working Python code based on the research and requirements.
3. Include docstrings and inline comments.
4. Wrap ALL executable code in ```python code blocks.
5. After writing code, state: 'Code complete. Ready for review.'

Do NOT conduct research. Do NOT provide critique beyond code quality.
""",
    llm_config=llm_config,
)

critic = autogen.AssistantAgent(
    name="Critic",
    system_message="""You are a technical reviewer. Your job:
1. Wait for the Coder to produce code.
2. Review the code for: correctness, edge cases, security issues, performance.
3. Provide specific, actionable feedback with line references where possible.
4. If the code is acceptable, state exactly: 'APPROVED. TASK COMPLETE'
5. If revisions are needed, list them clearly and address them to the Coder.

Be direct. Do not pad feedback with compliments.
""",
    llm_config=llm_config,
)

user_proxy = autogen.UserProxyAgent(
    name="UserProxy",
    human_input_mode="NEVER",  # Fully automated
    max_consecutive_auto_reply=1,
    is_termination_msg=lambda msg: "TASK COMPLETE" in msg.get("content", ""),
    code_execution_config={
        "work_dir": "agent_workspace",
        "use_docker": False,  # Set True in production for safety
    },
    system_message="You initiate tasks and execute code when required.",
)

# Create the group chat with all agents
groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, coder, critic],
    messages=[],
    max_round=20,  # Absolute maximum — prevents infinite loops
    speaker_selection_method="auto",  # Let the manager decide
    allow_repeat_speaker=False,  # Prevents agents from talking twice in a row
)

# Create the manager that coordinates the conversation
manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
    system_message="""You coordinate a team of three specialists: Researcher, Coder, and Critic.
    
    Typical workflow:
    1. Researcher gathers information first.
    2. Coder implements based on research.
    3. Critic reviews the code.
    4. If revisions needed, Coder revises, then Critic re-reviews.
    
    Enforce this order. If an agent speaks out of turn, redirect appropriately.
    Stop the conversation when Critic says 'APPROVED. TASK COMPLETE'.
    """,
)

# Start the group chat with a task
result = user_proxy.initiate_chat(
    manager,
    message="""Build a Python function that:
    1. Takes a list of URLs as input
    2. Fetches the content of each URL concurrently using asyncio
    3. Returns a dictionary mapping each URL to its response status code and content length
    4. Handles errors gracefully (timeouts, connection failures)
    Include proper type hints and a usage example.
    """,
)

UserProxy → Manager:
"Build a Python function that fetches URLs concurrently..."

Manager → Researcher:
"Researcher, please analyze the requirements and provide relevant background."

Researcher → All:
"For concurrent URL fetching in Python, the best approach is asyncio + aiohttp...
[detailed research on async patterns, error handling, type hints]
Research complete. Ready for implementation."

Manager → Coder:
"Coder, please implement based on the Researcher's findings."

Coder → All:
"```python
import asyncio
import aiohttp
from typing import Dict, Tuple

async def fetch_url(session: aiohttp.ClientSession, url: str) -> Tuple[int, int]:
    ...

Code complete. Ready for review."

Critic → All:
"Revisions look good. Timeout set correctly, exceptions properly scoped.
APPROVED. TASK COMPLETE"

[Chat terminates]

def custom_speaker_selection(last_speaker, groupchat):
    """Custom speaker selection based on workflow state."""
    messages = groupchat.messages
    
    if len(messages) == 0:
        return researcher  # Always start with researcher
    
    last_msg = messages[-1]["content"].lower()
    last_name = last_speaker.name
    
    # After research is done, go to coder
    if last_name == "Researcher" and "research complete" in last_msg:
        return coder
    
    # After code is done, go to critic
    if last_name == "Coder" and "code complete" in last_msg:
        return critic
    
    # After critique, go back to coder if revisions needed
    if last_name == "Critic" and "revisions needed" in last_msg:
        return coder
    
    # Default: let auto decide
    return "auto"

groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, coder, critic],
    messages=[],
    max_round=20,
    speaker_selection_method=custom_speaker_selection,
)

is_termination_msg=lambda msg: "TASK COMPLETE" in msg.get("content", "")

import logging

logging.basicConfig(level=logging.INFO)

try:
    result = user_proxy.initiate_chat(
        manager,
        message="Your task here",
        max_turns=20,
    )
    
    # Check if we hit max_round without terminating
    if len(groupchat.messages) >= groupchat.max_round:
        logging.warning("GroupChat hit max_round limit without clean termination")
        # Extract the last Coder output as fallback
        for msg in reversed(groupchat.messages):
            if msg["name"] == "Coder" and "```python" in msg["content"]:
                print("Fallback: using last Coder output")
                print(msg["content"])
                break

except Exception as e:
    logging.error(f"GroupChat failed: {e}")
    raise

import os
import autogen
from dotenv import load_dotenv

load_dotenv()

llm_config = {"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}

researcher = autogen.AssistantAgent(
    name="Researcher",
    system_message="Analyze requirements and provide technical background. End with 'Research complete. Ready for implementation.'",
    llm_config=llm_config,
)

coder = autogen.AssistantAgent(
    name="Coder",
    system_message="Implement Python code based on research. End with 'Code complete. Ready for review.'",
    llm_config=llm_config,
)

critic = autogen.AssistantAgent(
    name="Critic",
    system_message="Review code for correctness and quality. If acceptable, say exactly: 'APPROVED. TASK COMPLETE'",
    llm_config=llm_config,
)

user_proxy = autogen.UserProxyAgent(
    name="UserProxy",
    human_input_mode="NEVER",
    is_termination_msg=lambda msg: "TASK COMPLETE" in msg.get("content", ""),
    code_execution_config={"work_dir": "workspace", "use_docker": False},
)

groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, coder, critic],
    messages=[],
    max_round=15,
)

manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

user_proxy.initiate_chat(
    manager,
    message="Write a Python class for a rate-limited HTTP client with exponential backoff."
)