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