지난 글에서 AutoGen의 대화 기반 멀티 에이전트 협업을 살펴봤다. 이번 글에서는 OpenAI가 발표한 Swarm을 다룬다. Swarm은 LangChain, AutoGen, CrewAI에 비해 극도로 단순한 경량 프레임워크로, 핸드오프(Handoff)라는 단 하나의 핵심 메커니즘으로 멀티 에이전트를 구현한다.
Swarm이란
Swarm은 OpenAI가 교육·실험 목적으로 공개한 멀티 에이전트 프레임워크다. 핵심 아이디어는 두 가지다:
- Routines: 에이전트의 역할을
instructions(시스템 프롬프트)와functions(도구)로 정의 - Handoffs: 에이전트가 다른 에이전트를 반환해 실행권을 이전
의존성이 거의 없고(openai 패키지만), 코드베이스 자체가 수백 줄에 불과해 동작 원리를 완전히 이해하며 사용할 수 있다.
기본 구현
from swarm import Swarm, Agent
client = Swarm() # OpenAI 클라이언트 기반
# 에이전트 정의 (instructions + functions)
def transfer_to_sales():
"""구매 관련 문의를 Sales 에이전트로 이전합니다."""
return sales_agent # Agent 객체 반환 = 핸드오프
def transfer_to_support():
"""기술 지원 문의를 Support 에이전트로 이전합니다."""
return support_agent
def transfer_to_billing():
"""결제·환불 문의를 Billing 에이전트로 이전합니다."""
return billing_agent
def transfer_back_to_triage():
"""Triage 에이전트로 돌아갑니다."""
return triage_agent
# Triage: 의도 파악 + 라우팅
triage_agent = Agent(
name="Triage Agent",
instructions="""당신은 고객 지원 접수 담당자입니다.
사용자의 문의를 파악하고 적절한 팀으로 이전하세요:
- 구매·가격 문의 → transfer_to_sales()
- 기술적 문제 → transfer_to_support()
- 결제·환불 → transfer_to_billing()
절대 직접 답변하지 말고 항상 전문 팀으로 이전하세요.""",
functions=[transfer_to_sales, transfer_to_support, transfer_to_billing],
)
sales_agent = Agent(
name="Sales Agent",
instructions="""당신은 영업 전문가입니다.
제품 가격과 기능을 안내하고 구매를 도와줍니다.
결제 문제가 있으면 transfer_to_billing()을 호출하세요.""",
functions=[transfer_to_billing, transfer_back_to_triage],
)
support_agent = Agent(
name="Support Agent",
instructions="""당신은 기술 지원 엔지니어입니다.
버그, 오류, 설정 문제를 해결합니다.
해결이 어려우면 transfer_back_to_triage()를 호출하세요.""",
functions=[transfer_back_to_triage],
)
billing_agent = Agent(
name="Billing Agent",
instructions="환불, 결제 오류, 구독 변경을 처리합니다.",
functions=[transfer_back_to_triage],
)
# 실행 (단일 턴)
response = client.run(
agent=triage_agent,
messages=[{"role": "user", "content": "Claude Pro 구독 환불 신청하고 싶어요"}],
)
print(response.messages[-1]["content"])
# triage → billing_agent로 핸드오프 → 환불 안내context_variables: 에이전트 간 공유 상태
from swarm import Swarm, Agent
from swarm.types import Result
client = Swarm()
def set_user_tier(context_variables: dict, tier: str) -> Result:
"""사용자 등급을 업데이트합니다."""
context_variables["tier"] = tier
return Result(
value=f"등급이 {tier}로 업데이트되었습니다.",
context_variables=context_variables, # 업데이트된 context 반환
)
def get_personalized_greeting(context_variables: dict) -> str:
"""사용자 정보를 바탕으로 맞춤 인사를 반환합니다."""
name = context_variables.get("user_name", "고객님")
tier = context_variables.get("tier", "일반")
return f"안녕하세요, {name}({tier} 등급)! 무엇을 도와드릴까요?"
personalized_agent = Agent(
name="Personalized Agent",
instructions="""사용자의 context를 활용해 맞춤 서비스를 제공합니다.
항상 get_personalized_greeting으로 시작하세요.""",
functions=[get_personalized_greeting, set_user_tier],
)
# context_variables 초기값 설정
response = client.run(
agent=personalized_agent,
messages=[{"role": "user", "content": "안녕하세요"}],
context_variables={
"user_name": "김철수",
"user_id": "u12345",
"tier": "premium",
},
)
print(response.messages[-1]["content"])
print("업데이트된 context:", response.context_variables)멀티 턴 대화 루프
def run_conversation(starting_agent: Agent, context: dict = None):
"""인터랙티브 멀티 턴 대화 루프"""
client = Swarm()
messages = []
active_agent = starting_agent
context_variables = context or {}
print(f"[{active_agent.name}] 대화 시작. 'quit'으로 종료.\n")
while True:
user_input = input("사용자: ").strip()
if user_input.lower() == "quit":
break
messages.append({"role": "user", "content": user_input})
response = client.run(
agent=active_agent,
messages=messages,
context_variables=context_variables,
)
# 핸드오프 발생 시 active_agent 업데이트
if response.agent != active_agent:
print(f"\n[핸드오프: {active_agent.name} → {response.agent.name}]\n")
active_agent = response.agent
# context 업데이트
context_variables = response.context_variables
# 응답 출력
for msg in response.messages:
if msg["role"] == "assistant":
print(f"[{active_agent.name}]: {msg['content']}\n")
messages.append(msg)
# 실행
run_conversation(
starting_agent=triage_agent,
context={"user_name": "이영희", "tier": "standard"},
)도구 함수와 핸드오프 함수 조합
import json
from swarm import Agent, Swarm
# 실제 비즈니스 로직이 있는 도구
def check_order_status(order_id: str) -> str:
"""주문 상태를 확인합니다."""
# 실제로는 DB 조회
orders = {
"ORD-001": "배송중 (서울 → 부산)",
"ORD-002": "배송 완료",
"ORD-003": "처리 중",
}
return orders.get(order_id, f"주문 번호 {order_id}를 찾을 수 없습니다.")
def process_refund(order_id: str, reason: str) -> str:
"""환불을 처리합니다."""
return f"주문 {order_id}의 환불이 접수되었습니다. 사유: {reason}. 3-5 영업일 소요."
def escalate_to_human() -> str:
"""인간 상담원에게 연결합니다."""
return "상담원 연결 중... 예상 대기 시간: 5분"
# 역할이 명확한 전문 에이전트
order_agent = Agent(
name="Order Agent",
instructions="""주문 관련 문의를 처리합니다.
check_order_status로 상태를 확인하고,
복잡한 문제는 transfer_to_support()로 이전하세요.
매우 복잡하거나 불만이 심각하면 escalate_to_human()을 사용하세요.""",
functions=[check_order_status, escalate_to_human, transfer_to_support],
)
refund_agent = Agent(
name="Refund Agent",
instructions="""환불 및 결제 문제를 처리합니다.
process_refund로 환불 접수를 진행하세요.
주문 상태 확인이 필요하면 transfer_to_order()를 사용하세요.""",
functions=[process_refund, escalate_to_human],
)Swarm을 선택해야 할 때
# ✅ Swarm이 적합한 경우
# 1. 명확하게 분리된 역할의 에이전트가 3-7개 이내
# 2. 선형 또는 트리 구조의 핸드오프
# 3. 빠른 프로토타이핑, 개념 검증
# 4. OpenAI API 직접 사용 환경
# ❌ Swarm보다 다른 프레임워크가 나은 경우
# 1. 복잡한 순환 루프 → LangGraph
# 2. 역할 기반 팀 협업 + 메모리 → CrewAI
# 3. 코드 생성·실행 루프 → AutoGen
# 4. 대용량 문서 RAG → LlamaIndex
# 핵심 비교: 코드 복잡도
# Swarm: ~300줄 코어, 의존성 거의 없음
# AutoGen: ~10k줄, 풍부한 기능
# CrewAI: ~20k줄, 완전한 에코시스템
# LangGraph:~15k줄, 강력한 상태 관리정리
Swarm은 극도로 단순한 핸드오프 패턴으로 멀티 에이전트 시스템의 본질을 보여준다:
- 핸드오프: 함수가
Agent객체를 반환하면active_agent가 교체되는 단순한 메커니즘 - context_variables: 모든 에이전트가 공유하는 딕셔너리로 대화 상태 유지
- 단일 루프: 복잡한 오케스트레이터 없이 단순 루프로 에이전트 전환
- 경량성: OpenAI SDK만 의존, 코드가 작아 동작 원리를 완전히 이해 가능
고객 지원, 인테이크 라우팅, 단계별 워크플로우처럼 명확한 역할 분리와 선형 전환이 있는 시스템에 최적이다.
지난 글: AutoGen 완전 가이드: 대화 기반 멀티 에이전트 프레임워크
다음 글: 에이전트 메모리: 단기·장기·시맨틱 메모리 아키텍처
읽어주셔서 감사합니다. 😊