Интеграция AutoGen (Microsoft) для мульти-агентных систем

Интеграция Microsoft AutoGen для мульти-агентных систем

AutoGen — фреймворк Microsoft Research для построения мульти-агентных систем с акцентом на conversational patterns между агентами. Ключевая концепция: агенты общаются в диалогах, каждый может инициировать и отвечать. AutoGen v0.4 (AgentChat) переработан в сравнении с v0.2 — новый асинхронный дизайн, typed messages, runtime-абстракция для распределённого выполнения.

AutoGen AgentChat: базовые паттерны

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")

# Простой диалог двух агентов
assistant = AssistantAgent(
    name="assistant",
    model_client=model_client,
    system_message="Ты — полезный ассистент. Решай задачи последовательно.",
)

code_executor = AssistantAgent(
    name="code_executor",
    model_client=model_client,
    system_message="Ты — Python-разработчик. Пиши чистый, работающий код.",
)

# Условие завершения: когда кто-то напишет TERMINATE
termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(20)

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

async def run():
    result = await team.run(task="Напиши скрипт для парсинга CSV и вычисления средних значений по колонкам")
    print(result.messages[-1].content)

asyncio.run(run())

SelectorGroupChat: LLM-маршрутизация

from autogen_agentchat.teams import SelectorGroupChat

researcher = AssistantAgent(
    name="researcher",
    model_client=model_client,
    system_message="Ты исследователь. Находишь факты и данные.",
)

analyst = AssistantAgent(
    name="analyst",
    model_client=model_client,
    system_message="Ты аналитик. Интерпретируешь данные и строишь выводы.",
)

critic = AssistantAgent(
    name="critic",
    model_client=model_client,
    system_message="Ты критик. Выявляешь слабые места в аргументах.",
)

# SelectorGroupChat: LLM сам выбирает следующего участника
selector_team = SelectorGroupChat(
    participants=[researcher, analyst, critic],
    model_client=model_client,
    termination_condition=TextMentionTermination("DONE") | MaxMessageTermination(15),
    selector_prompt="""Выбери следующего участника беседы.
Доступные: {participants}
История: {history}
Верни только имя участника.""",
)

Custom Agents с инструментами

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

async def query_database(query: str, table: str) -> str:
    """Выполнить SQL-запрос к базе данных аналитики"""
    result = await db_pool.fetch(query)
    return str(result[:100])  # Ограничиваем вывод

async def send_email(to: str, subject: str, body: str) -> str:
    """Отправить email уведомление"""
    await email_service.send(to=to, subject=subject, body=body)
    return f"Email отправлен на {to}"

db_tool = FunctionTool(query_database, description="SQL-запрос к analytics DB")
email_tool = FunctionTool(send_email, description="Отправка email уведомлений")

data_agent = AssistantAgent(
    name="data_agent",
    model_client=model_client,
    tools=[db_tool],
    system_message="Анализируй данные через SQL-запросы. Всегда используй только SELECT.",
    reflect_on_tool_use=True,  # Агент анализирует результат инструмента
)

notification_agent = AssistantAgent(
    name="notification_agent",
    model_client=model_client,
    tools=[email_tool],
    system_message="Отправляй уведомления по результатам анализа.",
)

AutoGen Core: низкоуровневый 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:
        # Обработка запроса на анализ
        response = await self._model_client.create(
            messages=[{"role": "user", "content": message.query}]
        )
        return AnalysisResult(
            result=response.content,
            confidence=0.85,
        )

# Распределённый runtime
runtime = SingleThreadedAgentRuntime()
await AnalystAgent.register(runtime, "analyst", lambda: AnalystAgent(model_client))

runtime.start()
result = await runtime.send_message(
    AnalysisRequest(query="Анализ P&L за Q1 2025", requester_id="user_1"),
    recipient=AgentId("analyst", "default"),
)

MagenticOne: автономные веб-задачи

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

# MagenticOne — специализированная команда для веб-задач
# Включает: Orchestrator, WebSurfer, FileSurfer, Coder, ComputerTerminal
magentic = MagenticOne(
    client=OpenAIChatCompletionClient(model="gpt-4o"),
)

result = await magentic.run(task="""
Найди последние 5 публикаций о квантовых вычислениях на arxiv.org,
извлеки авторов и ключевые выводы, сохрани в papers.csv
""")

Практический кейс: система code review

Задача: автоматизация code review для команды из 12 разработчиков. Pull request требовал 2–4 часа ожидания ревьюера.

Команда агентов:

1. Security Reviewer — анализ уязвимостей (SQL injection, XSS, secrets в коде)
2. Performance Analyst — выявление N+1, неэффективных алгоритмов, лишних запросов
3. Style Checker — соответствие code style guide, именованиям, документации
4. Test Coverage Agent — анализ покрытия тестами, предложение кейсов
5. Summary Agent — сводный отчёт, приоритизация замечаний

Паттерн: SelectorGroupChat, каждый агент высказывается по своей области, Summary Agent формирует итог когда все высказались.

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 следующего PR:\n{pr_content}")

Результаты:

• Время получения первичного ревью: 2–4 часа → 4 минуты
• Выявленных проблем безопасности до мёрджа: +340% (ранее пропускались при усталости)
• Разработчики оценили качество замечаний: 4.2/5.0
• Ложные срабатывания (false positives): 12% — требует тонкой настройки промптов

Сроки

• Прототип 2-агентного диалога: 1–2 дня
• SelectorGroupChat с 4–5 агентами: 1 неделя
• Кастомные инструменты + интеграция с CI/CD: 2–3 недели
• AutoGen Core с распределённым runtime: 3–4 недели