2026 年 AutoGen 替代方案：AutoGen 之后该何去何从

# AutoGen (before)
pip install -U "autogen-agentchat" "autogen-ext[openai]"
export OPENAI_API_KEY="..."

# Swarms API (after)
pip install aiohttp python-dotenv
export SWARMS_API_KEY="..."   # from https://swarms.world/platform/api-keys

import asyncio

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


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")

    writer = AssistantAgent(
        "writer",
        model_client=model_client,
        system_message="You write concise product announcements.",
    )
    critic = AssistantAgent(
        "critic",
        model_client=model_client,
        system_message=(
            "Review the latest draft and give specific feedback. "
            "Reply with APPROVE when the draft needs no more changes."
        ),
    )

    # Stop when the critic approves, or after the task plus 8 agent messages.
    termination = TextMentionTermination("APPROVE") | MaxMessageTermination(9)
    team = RoundRobinGroupChat([writer, critic], termination_condition=termination)

    result = await team.run(task="Write a 100-word announcement for our new usage dashboard.")
    for message in result.messages:
        print(f"{message.source}: {message.content}\n")
    print("Stopped because:", result.stop_reason)

    await model_client.close()


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}

WRITER = {
    "agent_name": "Writer",
    "description": "Writes and revises product announcements",
    "system_prompt": "You write concise product announcements.",
    "model_name": "gpt-5.4",
    "max_loops": 1,
}
CRITIC = {
    "agent_name": "Critic",
    "description": "Reviews drafts and approves them",
    "system_prompt": (
        "Review the draft and give specific feedback. "
        "Reply with APPROVE when the draft needs no more changes."
    ),
    "model_name": "claude-sonnet-5",
    "max_loops": 1,
}


async def run_agent(session, agent_config, task, history=None):
    """Run one turn on POST /v1/agent/completions and return the reply text."""
    payload = {"agent_config": agent_config, "task": task}
    if history:
        payload["history"] = history
    async with session.post(f"{BASE_URL}/v1/agent/completions", json=payload) as resp:
        if resp.status >= 400:
            body = await resp.text()
            raise RuntimeError(f"{resp.status}: {body}")
        result = await resp.json()
    # "outputs" is a list of {"role": ..., "content": ...} turns.
    turns = [
        turn
        for turn in result["outputs"]
        if str(turn.get("role", "")).lower() not in ("user", "system")
    ]
    return str(turns[-1]["content"])


async def main() -> None:
    task = "Write a 100-word announcement for our new usage dashboard."
    writer_history = []
    prompt = task

    # One session for the whole conversation; each call gets up to 120 seconds.
    timeout = aiohttp.ClientTimeout(total=120)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        for _ in range(4):  # 4 writer turns and 4 critic turns, like MaxMessageTermination(9)
            draft = await run_agent(session, WRITER, prompt, writer_history)
            print(f"Writer:\n{draft}\n")
            writer_history += [
                {"role": "user", "content": prompt},
                {"role": "assistant", "content": draft},
            ]

            review = await run_agent(session, CRITIC, f"Task: {task}\n\nDraft:\n{draft}")
            print(f"Critic:\n{review}\n")
            if "APPROVE" in review:  # the TextMentionTermination("APPROVE") check
                break
            prompt = f"Revise your draft using this feedback:\n{review}"


asyncio.run(main())

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")

    strategist = AssistantAgent(
        "product_strategist",
        model_client=model_client,
        system_message=(
            "You define positioning, target personas and differentiation. "
            "Build on what earlier speakers said."
        ),
    )
    marketer = AssistantAgent(
        "growth_marketer",
        model_client=model_client,
        system_message=(
            "You propose acquisition channels and a launch plan. "
            "Build on what earlier speakers said."
        ),
    )
    engineer = AssistantAgent(
        "sales_engineer",
        model_client=model_client,
        system_message=(
            "You cover integration requirements and a proof-of-concept plan. "
            "Build on what earlier speakers said."
        ),
    )

    # 1 task message + 3 agents x 2 rounds = 7 messages.
    team = RoundRobinGroupChat(
        [strategist, marketer, engineer],
        termination_condition=MaxMessageTermination(7),
    )

    result = await team.run(
        task=(
            "Draft a go-to-market plan for an AI code review tool "
            "for engineering teams of 20 to 100 developers."
        )
    )
    for message in result.messages:
        print(f"{message.source}:\n{message.content}\n")

    await model_client.close()


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}


async def main() -> None:
    payload = {
        "name": "GTM roundtable",
        "description": "Three specialists take turns on one go-to-market plan",
        "swarm_type": "RoundRobin",
        "task": (
            "Draft a go-to-market plan for an AI code review tool "
            "for engineering teams of 20 to 100 developers."
        ),
        "max_loops": 2,  # every agent speaks twice, in list order
        "agents": [
            {
                "agent_name": "Product Strategist",
                "description": "Positioning and target personas",
                "system_prompt": (
                    "You define positioning, target personas and differentiation. "
                    "Build on what earlier speakers said."
                ),
                "model_name": "gpt-5.4",
                "max_loops": 1,
            },
            {
                "agent_name": "Growth Marketer",
                "description": "Acquisition channels and launch plan",
                "system_prompt": (
                    "You propose acquisition channels and a launch plan. "
                    "Build on what earlier speakers said."
                ),
                "model_name": "claude-sonnet-5",
                "max_loops": 1,
            },
            {
                "agent_name": "Sales Engineer",
                "description": "Integration requirements and proof of concept",
                "system_prompt": (
                    "You cover integration requirements and a proof-of-concept plan. "
                    "Build on what earlier speakers said."
                ),
                "model_name": "gpt-5.4-mini",
                "max_loops": 1,
            },
        ],
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    for turn in result["output"]:
        print(f"{turn['role']}:\n{turn['content']}\n")
    print("Seconds:", result["execution_time"])
    print("Usage:", result["usage"])


asyncio.run(main())

import asyncio

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


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")

    planner = AssistantAgent(
        "PlanningAgent",
        description="Plans the work. Should be the first to engage when given a new task.",
        model_client=model_client,
        system_message=(
            "You break the task into subtasks and assign each one as '<agent> : <task>'. "
            "Your team: MarketResearcher, PricingAnalyst. You do not do the work yourself. "
            "When every subtask is done, summarize the findings and end with TERMINATE."
        ),
    )
    researcher = AssistantAgent(
        "MarketResearcher",
        description="Researches competitors, segments and market size.",
        model_client=model_client,
        system_message="You research competitors, customer segments and market size.",
    )
    analyst = AssistantAgent(
        "PricingAnalyst",
        description="Analyzes competitor pricing and proposes price tiers.",
        model_client=model_client,
        system_message="You analyze competitor pricing and propose price tiers with reasoning.",
    )

    termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(25)
    team = SelectorGroupChat(
        [planner, researcher, analyst],
        model_client=model_client,  # picks the next speaker after every message
        termination_condition=termination,
    )

    result = await team.run(
        task=(
            "Recommend launch pricing for an AI code review tool for teams of "
            "20 to 100 developers. Cover competitors, segments and three price tiers."
        )
    )
    print(result.messages[-1].content)

    await model_client.close()


asyncio.run(main())

import asyncio
import os

import aiohttp
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["SWARMS_API_KEY"]
BASE_URL = "https://api.swarms.world"
HEADERS = {"x-api-key": API_KEY, "Content-Type": "application/json"}


async def main() -> None:
    payload = {
        "name": "Pricing study",
        "description": "A director plans the work and delegates to two specialists",
        "swarm_type": "HierarchicalSwarm",
        "task": (
            "Recommend launch pricing for an AI code review tool for teams of "
            "20 to 100 developers. Cover competitors, segments and three price tiers."
        ),
        "director_model_name": "gpt-5.4",
        "director_settings": {"temperature": 0.2},
        "max_loops": 1,
        "agents": [
            {
                "agent_name": "Market Researcher",
                "description": "Researches competitors, segments and market size",
                "system_prompt": "You research competitors, customer segments and market size.",
                "model_name": "claude-sonnet-5",
                "max_loops": 1,
            },
            {
                "agent_name": "Pricing Analyst",
                "description": "Analyzes competitor pricing and proposes price tiers",
                "system_prompt": "You analyze competitor pricing and propose price tiers with reasoning.",
                "model_name": "gpt-5.4",
                "max_loops": 1,
            },
        ],
    }

    # Multi-agent runs can take minutes, so give the whole request room.
    timeout = aiohttp.ClientTimeout(total=600)
    async with aiohttp.ClientSession(headers=HEADERS, timeout=timeout) as session:
        async with session.post(f"{BASE_URL}/v1/swarm/completions", json=payload) as resp:
            resp.raise_for_status()
            result = await resp.json()

    for turn in result["output"]:
        print(f"{turn['role']}:\n{turn['content']}\n")


asyncio.run(main())