AutoGen (Microsoft) Integration for Multi-Agent Systems

Microsoft AutoGen Integration for Multi-Agent Systems

AutoGen is a Microsoft Research framework for building multi-agent systems with an emphasis on conversational patterns between agents. Key concept: agents communicate in dialogues, each can initiate and respond. AutoGen v0.4 (AgentChat) is redesigned compared to v0.2 — new asynchronous design, typed messages, runtime abstraction for distributed execution.

AutoGen AgentChat: Basic Patterns

import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat, SelectorGroupChat, MagenticOneGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4o")

# Simple dialogue of two agents
assistant = AssistantAgent(
    name="assistant",
    model_client=model_client,
    system_message="You are a helpful assistant. Solve tasks sequentially.",
)

code_executor = AssistantAgent(
    name="code_executor",
    model_client=model_client,
    system_message="You are a Python developer. Write clean, working code.",
)

# Termination condition: when someone writes TERMINATE
termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(20)

team = RoundRobinGroupChat(
    participants=[assistant, code_executor],
    termination_condition=termination,
)

async def run():
    result = await team.run(task="Write a script for parsing CSV and calculating column averages")
    print(result.messages[-1].content)

asyncio.run(run())

SelectorGroupChat: LLM Routing

from autogen_agentchat.teams import SelectorGroupChat

researcher = AssistantAgent(
    name="researcher",
    model_client=model_client,
    system_message="You are a researcher. You find facts and data.",
)

analyst = AssistantAgent(
    name="analyst",
    model_client=model_client,
    system_message="You are an analyst. You interpret data and draw conclusions.",
)

critic = AssistantAgent(
    name="critic",
    model_client=model_client,
    system_message="You are a critic. You identify weaknesses in arguments.",
)

# SelectorGroupChat: LLM chooses the next participant
selector_team = SelectorGroupChat(
    participants=[researcher, analyst, critic],
    model_client=model_client,
    termination_condition=TextMentionTermination("DONE") | MaxMessageTermination(15),
    selector_prompt="""Choose the next conversation participant.
Available: {participants}
History: {history}
Return only the participant name.""",
)

Custom Agents with Tools

from autogen_agentchat.agents import AssistantAgent
from autogen_core.tools import FunctionTool

async def query_database(query: str, table: str) -> str:
    """Execute SQL query to analytics database"""
    result = await db_pool.fetch(query)
    return str(result[:100])  # Limit output

async def send_email(to: str, subject: str, body: str) -> str:
    """Send email notification"""
    await email_service.send(to=to, subject=subject, body=body)
    return f"Email sent to {to}"

db_tool = FunctionTool(query_database, description="SQL query to analytics DB")
email_tool = FunctionTool(send_email, description="Send email notifications")

data_agent = AssistantAgent(
    name="data_agent",
    model_client=model_client,
    tools=[db_tool],
    system_message="Analyze data through SQL queries. Always use SELECT only.",
    reflect_on_tool_use=True,  # Agent analyzes tool result
)

notification_agent = AssistantAgent(
    name="notification_agent",
    model_client=model_client,
    tools=[email_tool],
    system_message="Send notifications based on analysis results.",
)

AutoGen Core: Low-Level API

from autogen_core import SingleThreadedAgentRuntime, RoutedAgent, message_handler
from autogen_core import TopicId, TypeSubscription
from dataclasses import dataclass

@dataclass
class AnalysisRequest:
    query: str
    requester_id: str

@dataclass
class AnalysisResult:
    result: str
    confidence: float

class AnalystAgent(RoutedAgent):
    def __init__(self, model_client):
        super().__init__("Financial Analyst Agent")
        self._model_client = model_client

    @message_handler
    async def handle_request(self, message: AnalysisRequest, ctx) -> AnalysisResult:
        # Process analysis request
        response = await self._model_client.create(
            messages=[{"role": "user", "content": message.query}]
        )
        return AnalysisResult(
            result=response.content,
            confidence=0.85,
        )

# Distributed runtime
runtime = SingleThreadedAgentRuntime()
await AnalystAgent.register(runtime, "analyst", lambda: AnalystAgent(model_client))

runtime.start()
result = await runtime.send_message(
    AnalysisRequest(query="Analyze P&L for Q1 2025", requester_id="user_1"),
    recipient=AgentId("analyst", "default"),
)

MagenticOne: Autonomous Web Tasks

from autogen_ext.teams.magentic_one import MagenticOne
from autogen_ext.models.openai import OpenAIChatCompletionClient

# MagenticOne — specialized team for web tasks
# Includes: Orchestrator, WebSurfer, FileSurfer, Coder, ComputerTerminal
magentic = MagenticOne(
    client=OpenAIChatCompletionClient(model="gpt-4o"),
)

result = await magentic.run(task="""
Find the latest 5 publications on quantum computing on arxiv.org,
extract authors and key conclusions, save to papers.csv
""")

Practical Case: Code Review System

Task: automate code review for a team of 12 developers. Pull request required 2–4 hours of reviewer wait time.

Agent Team:

1. Security Reviewer — analyze vulnerabilities (SQL injection, XSS, secrets in code)
2. Performance Analyst — identify N+1, inefficient algorithms, unnecessary queries
3. Style Checker — code style guide compliance, naming, documentation
4. Test Coverage Agent — analyze test coverage, suggest cases
5. Summary Agent — summary report, prioritize comments

Pattern: SelectorGroupChat, each agent speaks in their area, Summary Agent forms conclusion when all have spoken.

termination = TextMentionTermination("REVIEW_COMPLETE") | MaxMessageTermination(25)

review_team = SelectorGroupChat(
    participants=[security_reviewer, perf_analyst, style_checker, test_agent, summary_agent],
    model_client=model_client,
    termination_condition=termination,
)

pr_content = load_pull_request(pr_id=1234)
result = await review_team.run(task=f"Review the following PR:\n{pr_content}")

Results:

• Initial review time: 2–4 hours → 4 minutes
• Security issues found before merge: +340% (previously missed due to fatigue)
• Developers rated comment quality: 4.2/5.0
• False positives: 12% — requires fine-tuning prompts

Timeline

• Prototype of 2-agent dialogue: 1–2 days
• SelectorGroupChat with 4–5 agents: 1 week
• Custom tools + CI/CD integration: 2–3 weeks
• AutoGen Core with distributed runtime: 3–4 weeks