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

Інтеграція Microsoft AutoGen для мультиагентних систем

AutoGen — фреймворк Microsoft Research для побудови мультиагентних систем з акцентом на розмовні шаблони між агентами. Ключова концепція: агенти спілкуються в діалогах, кожен може ініціювати та відповідати. AutoGen v0.4 (AgentChat) перероблено порівняно з v0.2 — новий асинхронний дизайн, типізовані повідомлення, 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}
Верни тільки ім'я учасника.""",
)

Користувацькі агенти з інструментами

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

Результати:

• Час отримання первинного review: 2–4 години → 4 хвилини
• Виявлені проблеми безпеки до merge: +340% (раніше пропускалися через втому)
• Розробники оцінили якість замічань: 4.2/5.0
• Помилкові спрацювання: 12% — вимагає тонкого налаштування промптів

Строки

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