AutoGen: 58K+ Stars — 멀티 에이전트 프레임워크 심층 분석: CrewAI, LangGraph와의 비교 2026

# 가상 환경 생성
python -m venv .venv
source .venv/bin/activate

# AgentChat + OpenAI 확장 설치
pip install -U "autogen-agentchat" "autogen-ext[openai]"

pip install -U "autogen-agentchat" "autogen-ext[openai,azure,docker,mcp]"

import autogen_agentchat
print(autogen_agentchat.__version__)

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main() -> None:
    agent = AssistantAgent(
        name="assistant",
        model_client=OpenAIChatCompletionClient(
            model="gpt-4o",
            api_key="YOUR_API_KEY"
        ),
        system_message="You are a helpful assistant."
    )
    result = await agent.run(task="Say 'Hello World!'")
    print(result.messages[-1].content)

asyncio.run(main())

export OPENAI_API_KEY="sk-..."
python hello_agent.py

# 공식 이미지 가져오기
docker pull mcr.microsoft.com/autogen/python:latest

# API 키로 실행
docker run -it \
  -e OPENAI_API_KEY="$OPENAI_API_KEY" \
  -v "$(pwd)/workspace:/workspace" \
  mcr.microsoft.com/autogen/python:latest

from autogen_ext.models.openai import OpenAIChatCompletionClient

# OpenAI 직접 연결
openai_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key="sk-..."
)

# Azure OpenAI
azure_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    base_url="https://YOUR_RESOURCE.openai.azure.com/openai/deployments/YOUR_DEPLOYMENT",
    api_key="YOUR_AZURE_KEY",
    api_version="2024-12-01-preview"
)

from autogen_ext.models.openai import OpenAIChatCompletionClient

local_client = OpenAIChatCompletionClient(
    model="llama3.1:8b",
    base_url="http://localhost:11434/v1",
    api_key="placeholder",
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": False,
        "family": "unknown"
    }
)

from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_agentchat.agents import CodeExecutorAgent

# Docker 기반 코드 실행기 생성
executor = DockerCommandLineCodeExecutor(
    image="python:3.12-slim",
    work_dir="./coding_workspace",
    timeout=60,
    stop_container=True
)

code_agent = CodeExecutorAgent(
    name="code_executor",
    code_executor=executor
)

# 마켓플레이스에서 설치 ("AutoGen" 검색)
# 또는 CLI를 통해
code --install-extension microsoft.autogen

from autogen_ext.tools.mcp import McpWorkbench

workbench = McpWorkbench(
    server_params={"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "."]}
)

# MCP 서버의 도구가 에이전트가 사용할 수 있게 됩니다

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

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # 전문 에이전트 정의
    researcher = AssistantAgent(
        name="researcher",
        model_client=model_client,
        system_message="You are a research analyst. Gather facts and data."
    )

    writer = AssistantAgent(
        name="writer",
        model_client=model_client,
        system_message="You are a technical writer. Create clear documentation."
    )

    reviewer = AssistantAgent(
        name="reviewer",
        model_client=model_client,
        system_message="You are an editor. Review content for accuracy and clarity. "
                       "Respond with 'APPROVED' when the content is good."
    )

    # 종료: 20개 메시지 후 또는 검토자가 승인할 때 중지
    termination = MaxMessageTermination(max_messages=20) | TextMentionTermination("APPROVED")

    # 라운드 로빈 그룹 챗
    team = RoundRobinGroupChat(
        participants=[researcher, writer, reviewer],
        termination_condition=termination
    )

    result = await team.run(task="Write a one-paragraph summary of quantum computing.")
    for msg in result.messages:
        print(f"[{msg.source}]: {msg.content[:100]}...")

asyncio.run(main())

from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

# SelectorGroupChat은 LLM을 사용하여 다음에 발언할 에이전트를 결정합니다
team = SelectorGroupChat(
    participants=[researcher, writer, reviewer],
    model_client=model_client,
    termination_condition=MaxMessageTermination(max_messages=15),
    allow_repeated_speaker=False  # 동일한 에이전트가 연속으로 발언하는 것을 방지
)

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

def search_knowledge_base(query: str) -> str:
    """Search internal knowledge base."""
    # 검색 로직
    return f"Results for '{query}': ..."

search_tool = FunctionTool(search_knowledge_base, description="Search company KB")

agent = AssistantAgent(
    name="kb_assistant",
    model_client=model_client,
    tools=[search_tool],
    system_message="Use the search_knowledge_base tool to answer questions."
)

from autogen_agentchat.teams import GroupChat
from autogen_core import CancellationToken

# 대화 상태 직렬화
state = await team.save_state()

# Redis / 데이터베이스에 저장
import json
with open("team_state.json", "w") as f:
    json.dump(state, f)

# 나중에: 복원하고 재개
with open("team_state.json") as f:
    state = json.load(f)
await team.load_state(state)
result = await team.run(task="Continue from where we left off.")

from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
import tempfile

# 신뢰할 수 없는 코드에는 항상 Docker 사용
with tempfile.TemporaryDirectory() as work_dir:
    executor = DockerCommandLineCodeExecutor(
        image="python:3.12-slim",
        work_dir=work_dir,
        timeout=30,
        bind_mounts={"src": "/safe", "target": "/workspace"}
    )

    code_executor = CodeExecutorAgent(
        name="sandbox",
        code_executor=executor
    )
    # 에이전트는 컨테이너 납부에서 모든 코드를 실행합니다

from autogen_core import TRACE_LOGGER_NAME
import logging

# AutoGen 납부 추적 활성화
logging.getLogger(TRACE_LOGGER_NAME).setLevel(logging.DEBUG)

# OTLP 수집기와 통합
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

tracer = trace.get_tracer("autogen.production")

async for message in team.run_stream(task="Explain Kubernetes"):
    if message.source == "assistant":
        print(message.content, end="", flush=True)

# 진행 중인 모든 메시지 출력
team = RoundRobinGroupChat(
    participants=[agent1, agent2],
    termination_condition=termination
)
result = await team.run(task="Debug task", max_turns=10)
for msg in result.messages:
    print(f"{msg.source} -> {msg.content[:200]}")