Swarm

Swarm (핸드오프 방식 팀)

Swarm은 에이전트가 자기 능력에 따라 다른 에이전트에게 작업을 넘겨주는(hand off) 팀이에요. OpenAI가 Swarm에서 처음 소개한 멀티 에이전트 설계 패턴이죠. 핵심 아이디어는 특수한 도구 호출로 에이전트가 다른 에이전트에게 작업을 위임하게 하면서, 모든 에이전트가 같은 메시지 컨텍스트를 공유한다는 거예요. 이렇게 하면 중앙 오케스트레이터(예: SelectorGroupChat)에 의존하는 대신, 에이전트가 작업 계획에 대한 지역적 결정을 내릴 수 있어요.

Swarm은 고수준 API예요. 이 API가 지원하지 않는 더 세밀한 제어·커스터마이징이 필요하면 Core API 문서의 Handoff 패턴을 보고 직접 Swarm 패턴을 구현할 수 있어요.

어떻게 동작할까요?

Swarm 팀은 근본적으로 참가자들이 돌아가며 응답을 생성하는 그룹 채팅이에요. SelectorGroupChat·RoundRobinGroupChat처럼 참가 에이전트가 응답을 브로드캐스트해서 모두 같은 메시지 컨텍스트를 공유하죠. 다른 두 그룹 채팅 팀과 달리, 각 차례의 발언자 에이전트는 컨텍스트에서 가장 최근 HandoffMessage 메시지를 기준으로 선택돼요. 그래서 팀의 각 에이전트가 어떤 에이전트로 넘길지 알려주는 HandoffMessage를 생성할 수 있어야 해요.

AssistantAgent에서는 handoffs 인자로 넘길 수 있는 에이전트를 지정해요. Handoff를 쓰면 메시지 내용과 핸드오프 동작을 커스터마이징할 수 있어요.

전체 과정은 이렇게 정리할 수 있어요.

1. 각 에이전트는 어느 에이전트로 넘길 수 있는지 알려주는 HandoffMessage를 생성할 수 있어요. AssistantAgent에서는 handoffs 인자를 설정하면 돼요.
2. 팀이 작업을 시작하면 첫 발언자 에이전트가 작업을 처리하고, 넘길지·누구에게 넘길지 지역적 결정을 내려요.
3. 에이전트가 HandoffMessage를 생성하면, 받는 에이전트가 같은 메시지 컨텍스트로 작업을 이어받아요.
4. 종료 조건이 충족될 때까지 이 과정을 반복해요.

AssistantAgent는 모델의 도구 호출 능력으로 핸드오프를 생성해요. 그래서 모델이 도구 호출을 지원해야 해요. 모델이 병렬 도구 호출을 하면 여러 핸드오프가 동시에 생성될 수 있어 예기치 못한 동작이 날 수 있어요. 피하려면 모델 클라이언트 설정에서 병렬 도구 호출을 끄면 돼요. OpenAIChatCompletionClient·AzureOpenAIChatCompletionClient에서는 설정에 parallel_tool_calls=False를 주면 돼요.

이번 장에서는 Swarm 팀을 쓰는 두 가지 예시를 보여줄게요.

1. 휴먼-인-더-루프 핸드오프가 있는 고객 지원 팀
2. 콘텐츠 생성을 위한 자율 팀

고객 지원 예시 (Customer Support Example)

이 시스템은 두 에이전트로 항공편 환불 시나리오를 구현해요.

• Travel Agent: 일반 여행·환불 조정 담당.
• Flights Refunder: refund_flight 도구로 항공편 환불 처리 전문.

에이전트가 "user"에게 핸드오프할 때는 사용자가 에이전트와 상호작용하도록 해요.

워크플로

1. Travel Agent가 대화를 시작하고 사용자 요청을 평가해요.
2. 요청에 따라:
   • 환불 관련 작업이면 Travel Agent가 Flights Refunder에게 핸드오프.
   • 고객에게 정보가 필요하면 어느 에이전트든 "user"에게 핸드오프.
3. Flights Refunder가 적절할 때 refund_flight 도구로 환불을 처리해요.
4. 에이전트가 "user"에게 핸드오프하면 팀 실행이 멈추고 사용자 응답을 기다려요.
5. 사용자가 입력하면 그 내용이 HandoffMessage로 팀에 다시 보내져요. 이 메시지는 원래 사용자 입력을 요청한 에이전트를 향해요.
6. Travel Agent가 작업이 완료됐다고 판단하고 워크플로를 종료할 때까지 이 과정이 이어져요.

출처: 공식 문서 - Swarm

from typing import Any, Dict, List

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.messages import HandoffMessage
from autogen_agentchat.teams import Swarm
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

도구 (Tools)

def refund_flight(flight_id: str) -> str:
    """Refund a flight"""
    return f"Flight {flight_id} refunded"

에이전트 (Agents)

model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    # api_key="YOUR_API_KEY",
)

travel_agent = AssistantAgent(
    "travel_agent",
    model_client=model_client,
    handoffs=["flights_refunder", "user"],
    system_message="""You are a travel agent.
    The flights_refunder is in charge of refunding flights.
    If you need information from the user, you must first send your message, then you can handoff to the user.
    Use TERMINATE when the travel planning is complete.""",
)

flights_refunder = AssistantAgent(
    "flights_refunder",
    model_client=model_client,
    handoffs=["travel_agent", "user"],
    tools=[refund_flight],
    system_message="""You are an agent specialized in refunding flights.
    You only need flight reference numbers to refund a flight.
    You have the ability to refund a flight using the refund_flight tool.
    If you need information from the user, you must first send your message, then you can handoff to the user.
    When the transaction is complete, handoff to the travel agent to finalize.""",
)

termination = HandoffTermination(target="user") | TextMentionTermination("TERMINATE")
team = Swarm([travel_agent, flights_refunder], termination_condition=termination)

task = "I need to refund my flight."


async def run_team_stream() -> None:
    task_result = await Console(team.run_stream(task=task))
    last_message = task_result.messages[-1]

    while isinstance(last_message, HandoffMessage) and last_message.target == "user":
        user_message = input("User: ")

        task_result = await Console(
            team.run_stream(task=HandoffMessage(source="user", target=last_message.source, content=user_message))
        )
        last_message = task_result.messages[-1]


# Use asyncio.run(...) if you are running this in a script.
await run_team_stream()
await model_client.close()

주식 리서치 예시 (Stock Research Example)

이 시스템은 네 에이전트로 주식 리서치 작업을 수행해요.

• Planner: 전문 에이전트에게 전문성에 따라 작업을 위임하는 중앙 조정자. 각 에이전트를 효율적으로 쓰도록 하고 전체 워크플로를 감독해요.
• Financial Analyst: get_stock_data 같은 도구로 재무 지표와 주가 데이터를 분석하는 전문 에이전트.
• News Analyst: get_news 같은 도구로 주식 관련 최신 뉴스를 수집·요약하는 에이전트.
• Writer: 주식·뉴스 분석 결과를 하나의 최종 리포트로 엮는 에이전트.

워크플로

1. Planner가 적절한 에이전트에게 단계적으로 작업을 위임하며 리서치를 시작해요.
2. 각 에이전트는 독립적으로 작업을 수행하고 그 작업을 공유 메시지 스레드/기록에 추가해요. 플래너에게 결과를 직접 돌려주는 대신, 모든 에이전트가 이 공유 기록에 기여하고 읽어요. LLM으로 작업을 만들 때 이 공유 기록을 컨텍스트로 삼아 전체 진행 상황을 추적할 수 있죠.
3. 에이전트가 작업을 끝내면 제어권을 플래너에게 핸드오프해요.
4. 플래너가 필요한 작업이 모두 완료됐다고 판단해 워크플로를 종료할 때까지 이 과정이 이어져요.

도구 (Tools)

async def get_stock_data(symbol: str) -> Dict[str, Any]:
    """Get stock market data for a given symbol"""
    return {"price": 180.25, "volume": 1000000, "pe_ratio": 65.4, "market_cap": "700B"}


async def get_news(query: str) -> List[Dict[str, str]]:
    """Get recent news articles about a company"""
    return [
        {
            "title": "Tesla Expands Cybertruck Production",
            "date": "2024-03-20",
            "summary": "Tesla ramps up Cybertruck manufacturing capacity at Gigafactory Texas, aiming to meet strong demand.",
        },
        {
            "title": "Tesla FSD Beta Shows Promise",
            "date": "2024-03-19",
            "summary": "Latest Full Self-Driving beta demonstrates significant improvements in urban navigation and safety features.",
        },
        {
            "title": "Model Y Dominates Global EV Sales",
            "date": "2024-03-18",
            "summary": "Tesla's Model Y becomes best-selling electric vehicle worldwide, capturing significant market share.",
        },
    ]

model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    # api_key="YOUR_API_KEY",
)

planner = AssistantAgent(
    "planner",
    model_client=model_client,
    handoffs=["financial_analyst", "news_analyst", "writer"],
    system_message="""You are a research planning coordinator.
    Coordinate market research by delegating to specialized agents:
    - Financial Analyst: For stock data analysis
    - News Analyst: For news gathering and analysis
    - Writer: For compiling final report
    Always send your plan first, then handoff to appropriate agent.
    Always handoff to a single agent at a time.
    Use TERMINATE when research is complete.""",
)

financial_analyst = AssistantAgent(
    "financial_analyst",
    model_client=model_client,
    handoffs=["planner"],
    tools=[get_stock_data],
    system_message="""You are a financial analyst.
    Analyze stock market data using the get_stock_data tool.
    Provide insights on financial metrics.
    Always handoff back to planner when analysis is complete.""",
)

news_analyst = AssistantAgent(
    "news_analyst",
    model_client=model_client,
    handoffs=["planner"],
    tools=[get_news],
    system_message="""You are a news analyst.
    Gather and analyze relevant news using the get_news tool.
    Summarize key market insights from news.
    Always handoff back to planner when analysis is complete.""",
)

writer = AssistantAgent(
    "writer",
    model_client=model_client,
    handoffs=["planner"],
    system_message="""You are a financial report writer.
    Compile research findings into clear, concise reports.
    Always handoff back to planner when writing is complete.""",
)

# Define termination condition
text_termination = TextMentionTermination("TERMINATE")
termination = text_termination

research_team = Swarm(
    participants=[planner, financial_analyst, news_analyst, writer], termination_condition=termination
)

task = "Conduct market research for TSLA stock"
await Console(research_team.run_stream(task=task))
await model_client.close()

더 알아보기 (Learn more)

• Teams - 멀티 에이전트 팀
• Hand-off 워크플로 (Core API)
• Selector Group Chat - 모델 기반 발언자 선택
• Termination - 종료 조건