Hand-off 설계 패턴
Hand-off 설계 패턴
Hand-off는 OpenAI가 Swarm이라는 실험 프로젝트에서 소개한 멀티에이전트 설계 패턴이에요. 핵심 아이디어는 특별한 도구 호출(special tool call)로 에이전트가 다른 에이전트에게 작업을 위임하게 하는 거예요. 이 글에서는 AutoGen Core API로 이 hand-off 패턴을 이벤트 기반 에이전트로 구현하는 방법을 살펴볼게요.
AutoGen (v0.4+)을 쓰면 OpenAI 구현이나 이전 버전(v0.2)과 비교해 몇 가지 장점이 있어요:
- 분산 에이전트 런타임으로 분산 환경까지 확장할 수 있어요.
- 나만의 에이전트 구현을 가져올 수 있는 유연성이 있어요.
- 기본적으로 비동기(async) API라서 UI나 다른 시스템과 쉽게 통합할 수 있어요.
이 노트북은 hand-off 패턴의 간단한 구현을 보여줘요. pub-sub과 이벤트 기반 에이전트의 기본 개념을 이해하려면 주제와 구독을 먼저 읽는 걸 권장해요.
참고: 현재 AgentChat에 hand-off 패턴용 고수준 API를 만들고 있어서, 곧 훨씬 빠르게 시작할 수 있게 될 거예요.
출처: 공식 문서 - Handoffs
시나리오
이 시나리오는 OpenAI 예시를 바탕으로 수정한 것이에요.
고객이 제품 환불을 받으려 하거나 채팅봇에서 새 제품을 사려는 고객 서비스 시나리오를 생각해 볼게요. 채팅봇은 세 개의 AI 에이전트와 한 명의 휴먼 에이전트로 이뤄진 멀티에이전트 팀이에요:
- Triage Agent — 고객의 요청을 이해하고 어느 에이전트로 hand off할지 결정해요.
- Refund Agent — 환불 요청을 처리해요.
- Sales Agent — 판매 요청을 처리해요.
- Human Agent — AI 에이전트들이 처리할 수 없는 복잡한 요청을 담당해요.
이 시나리오에서 고객은 User Agent를 통해 채팅봇과 상호작용해요. 이제 이 시나리오를 AutoGen Core로 구현해 볼게요. 먼저 필요한 모듈을 import해요:
import json
import uuid
from typing import List, Tuple
from autogen_core import (
FunctionCall,
MessageContext,
RoutedAgent,
SingleThreadedAgentRuntime,
TopicId,
TypeSubscription,
message_handler,
)
from autogen_core.models import (
AssistantMessage,
ChatCompletionClient,
FunctionExecutionResult,
FunctionExecutionResultMessage,
LLMMessage,
SystemMessage,
UserMessage,
)
from autogen_core.tools import FunctionTool, Tool
from autogen_ext.models.openai import OpenAIChatCompletionClient
from pydantic import BaseModel
메시지 프로토콜
시작하기 전에 에이전트들이 통신할 메시지 프로토콜을 정의해야 해요. 이벤트 기반 pub-sub 통신을 쓰므로 이 메시지 타입들은 이벤트로 사용돼요.
UserLogin— 사용자가 로그인하고 새 세션을 시작할 때 런타임이 발행하는 메시지예요.UserTask— 사용자 세션의 채팅 기록을 담는 메시지예요. AI 에이전트가 작업을 다른 에이전트에게 hand off할 때도UserTask메시지를 발행해요.AgentResponse— AI 에이전트와 Human Agent가 발행하는 메시지로, 채팅 기록과 고객이 답할 토픽 타입을 함께 담아요.
class UserLogin(BaseModel):
pass
class UserTask(BaseModel):
context: List[LLMMessage]
class AgentResponse(BaseModel):
reply_to_topic_type: str
context: List[LLMMessage]
AI 에이전트
AIAgent 클래스부터 시작할게요. 멀티에이전트 채팅봇 안의 모든 AI 에이전트(즉 Triage, Sales, Issues and Repairs 에이전트)의 클래스에요. AIAgent는 autogen_core.models.ChatCompletionClient로 응답을 생성해요. 일반 도구를 직접 쓰거나 delegate_tools로 다른 에이전트에게 작업을 위임할 수 있어요. agent_topic_type 토픽 타입을 구독해 고객의 메시지를 받고, user_topic_type 토픽 타입에 발행해 고객에게 메시지를 보내요.
handle_task 메서드에서 에이전트는 먼저 모델로 응답을 생성해요. 응답에 hand-off 도구 호출이 있으면, 도구 호출 결과에서 지정한 토픽에 UserTask 메시지를 발행해서 작업을 다른 에이전트에게 위임해요. 응답이 일반 도구 호출이면 에이전트는 도구를 실행하고, 응답이 도구 호출이 아닐 때까지 모델을 다시 호출해서 다음 응답을 생성해요.
모델 응답이 도구 호출이 아니면, 에이전트는 user_topic_type에 발행해서 고객에게 AgentResponse 메시지를 보내요.
class AIAgent(RoutedAgent):
def __init__(
self,
description: str,
system_message: SystemMessage,
model_client: ChatCompletionClient,
tools: List[Tool],
delegate_tools: List[Tool],
agent_topic_type: str,
user_topic_type: str,
) -> None:
super().__init__(description)
self._system_message = system_message
self._model_client = model_client
self._tools = dict([(tool.name, tool) for tool in tools])
self._tool_schema = [tool.schema for tool in tools]
self._delegate_tools = dict([(tool.name, tool) for tool in delegate_tools])
self._delegate_tool_schema = [tool.schema for tool in delegate_tools]
self._agent_topic_type = agent_topic_type
self._user_topic_type = user_topic_type
@message_handler
async def handle_task(self, message: UserTask, ctx: MessageContext) -> None:
# Send the task to the LLM.
llm_result = await self._model_client.create(
messages=[self._system_message] + message.context,
tools=self._tool_schema + self._delegate_tool_schema,
cancellation_token=ctx.cancellation_token,
)
print(f"{'-'*80}\n{self.id.type}:\n{llm_result.content}", flush=True)
# Process the LLM result.
while isinstance(llm_result.content, list) and all(isinstance(m, FunctionCall) for m in llm_result.content):
tool_call_results: List[FunctionExecutionResult] = []
delegate_targets: List[Tuple[str, UserTask]] = []
# Process each function call.
for call in llm_result.content:
arguments = json.loads(call.arguments)
if call.name in self._tools:
# Execute the tool directly.
result = await self._tools[call.name].run_json(arguments, ctx.cancellation_token)
result_as_str = self._tools[call.name].return_value_as_string(result)
tool_call_results.append(
FunctionExecutionResult(call_id=call.id, content=result_as_str, is_error=False, name=call.name)
)
elif call.name in self._delegate_tools:
# Execute the tool to get the delegate agent's topic type.
result = await self._delegate_tools[call.name].run_json(arguments, ctx.cancellation_token)
topic_type = self._delegate_tools[call.name].return_value_as_string(result)
# Create the context for the delegate agent, including the function call and the result.
delegate_messages = list(message.context) + [
AssistantMessage(content=[call], source=self.id.type),
FunctionExecutionResultMessage(
content=[
FunctionExecutionResult(
call_id=call.id,
content=f"Transferred to {topic_type}. Adopt persona immediately.",
is_error=False,
name=call.name,
)
]
),
]
delegate_targets.append((topic_type, UserTask(context=delegate_messages)))
else:
raise ValueError(f"Unknown tool: {call.name}")
if len(delegate_targets) > 0:
# Delegate the task to other agents by publishing messages to the corresponding topics.
for topic_type, task in delegate_targets:
print(f"{'-'*80}\n{self.id.type}:\nDelegating to {topic_type}", flush=True)
await self.publish_message(task, topic_id=TopicId(topic_type, source=self.id.key))
if len(tool_call_results) > 0:
print(f"{'-'*80}\n{self.id.type}:\n{tool_call_results}", flush=True)
# Make another LLM call with the results.
message.context.extend(
[
AssistantMessage(content=llm_result.content, source=self.id.type),
FunctionExecutionResultMessage(content=tool_call_results),
]
)
llm_result = await self._model_client.create(
messages=[self._system_message] + message.context,
tools=self._tool_schema + self._delegate_tool_schema,
cancellation_token=ctx.cancellation_token,
)
print(f"{'-'*80}\n{self.id.type}:\n{llm_result.content}", flush=True)
else:
# The task has been delegated, so we are done.
return
# The task has been completed, publish the final result.
assert isinstance(llm_result.content, str)
message.context.append(AssistantMessage(content=llm_result.content, source=self.id.type))
await self.publish_message(
AgentResponse(context=message.context, reply_to_topic_type=self._agent_topic_type),
topic_id=TopicId(self._user_topic_type, source=self.id.key),
)
휴먼 에이전트
HumanAgent 클래스는 채팅봇 안에서 휴먼을 대신하는 프록시예요. AI 에이전트가 처리할 수 없는 요청을 처리하는 데 쓰여요. HumanAgent는 agent_topic_type 토픽 타입을 구독해 메시지를 받고, user_topic_type 토픽 타입에 발행해 고객에게 메시지를 보내요.
이 구현에서 HumanAgent는 콘솔로 입력을 받아요. 실제 애플리케이션에서는 다음과 같이 설계를 개선할 수 있어요:
handle_user_task메서드에서 Teams나 Slack 같은 채팅 애플리케이션으로 알림을 보내요.- 채팅 애플리케이션이
agent_topic_type으로 지정된 토픽에 휴먼의 응답을 런타임으로 발행해요. - 휴먼의 응답을 처리해서 고객에게 돌려보내는 또 다른 메시지 핸들러를 만들어요.
class HumanAgent(RoutedAgent):
def __init__(self, description: str, agent_topic_type: str, user_topic_type: str) -> None:
super().__init__(description)
self._agent_topic_type = agent_topic_type
self._user_topic_type = user_topic_type
@message_handler
async def handle_user_task(self, message: UserTask, ctx: MessageContext) -> None:
human_input = input("Human agent input: ")
print(f"{'-'*80}\n{self.id.type}:\n{human_input}", flush=True)
message.context.append(AssistantMessage(content=human_input, source=self.id.type))
await self.publish_message(
AgentResponse(context=message.context, reply_to_topic_type=self._agent_topic_type),
topic_id=TopicId(self._user_topic_type, source=self.id.key),
)
유저 에이전트
UserAgent 클래스는 채팅봇과 대화하는 고객을 대신하는 프록시예요. UserLogin과 AgentResponse 두 가지 메시지 타입을 처리해요. UserAgent가 UserLogin 메시지를 받으면 채팅봇으로 새 세션을 시작하고, agent_topic_type 토픽 타입을 구독하는 AI 에이전트에게 UserTask 메시지를 발행해요. UserAgent가 AgentResponse 메시지를 받으면 채팅봇의 응답을 사용자에게 보여줘요.
이 구현에서 UserAgent는 콘솔로 입력을 받아요. 실제 애플리케이션에서는 위 HumanAgent 섹션에서 설명한 것과 같은 아이디어로 휴먼 상호작용을 개선할 수 있어요.
class UserAgent(RoutedAgent):
def __init__(self, description: str, user_topic_type: str, agent_topic_type: str) -> None:
super().__init__(description)
self._user_topic_type = user_topic_type
self._agent_topic_type = agent_topic_type
@message_handler
async def handle_user_login(self, message: UserLogin, ctx: MessageContext) -> None:
print(f"{'-'*80}\nUser login, session ID: {self.id.key}.", flush=True)
# Get the user's initial input after login.
user_input = input("User: ")
print(f"{'-'*80}\n{self.id.type}:\n{user_input}")
await self.publish_message(
UserTask(context=[UserMessage(content=user_input, source="User")]),
topic_id=TopicId(self._agent_topic_type, source=self.id.key),
)
@message_handler
async def handle_task_result(self, message: AgentResponse, ctx: MessageContext) -> None:
# Get the user's input after receiving a response from an agent.
user_input = input("User (type 'exit' to close the session): ")
print(f"{'-'*80}\n{self.id.type}:\n{user_input}", flush=True)
if user_input.strip().lower() == "exit":
print(f"{'-'*80}\nUser session ended, session ID: {self.id.key}.")
return
message.context.append(UserMessage(content=user_input, source="User"))
await self.publish_message(
UserTask(context=message.context), topic_id=TopicId(message.reply_to_topic_type, source=self.id.key)
)
AI 에이전트용 도구
AI 에이전트는 다른 에이전트에게 작업을 hand off할 필요가 없으면 일반 도구를 써서 작업을 완료할 수 있어요. 단순 함수로 도구를 정의하고 autogen_core.tools.FunctionTool 래퍼로 도구를 만들어요:
def execute_order(product: str, price: int) -> str:
print("\n\n=== Order Summary ===")
print(f"Product: {product}")
print(f"Price: ${price}")
print("=================\n")
confirm = input("Confirm order? y/n: ").strip().lower()
if confirm == "y":
print("Order execution successful!")
return "Success"
else:
print("Order cancelled!")
return "User cancelled order."
def look_up_item(search_query: str) -> str:
item_id = "item_132612938"
print("Found item:", item_id)
return item_id
def execute_refund(item_id: str, reason: str = "not provided") -> str:
print("\n\n=== Refund Summary ===")
print(f"Item ID: {item_id}")
print(f"Reason: {reason}")
print("=================\n")
print("Refund execution successful!")
return "success"
execute_order_tool = FunctionTool(execute_order, description="Price should be in USD.")
look_up_item_tool = FunctionTool(
look_up_item, description="Use to find item ID.\nSearch query can be a description or keywords."
)
execute_refund_tool = FunctionTool(execute_refund, description="")
에이전트용 토픽 타입
각 에이전트가 구독할 토픽 타입을 정의해요. 토픽 타입에 대한 자세한 내용은 주제와 구독을 참고해요:
sales_agent_topic_type = "SalesAgent"
issues_and_repairs_agent_topic_type = "IssuesAndRepairsAgent"
triage_agent_topic_type = "TriageAgent"
human_agent_topic_type = "HumanAgent"
user_topic_type = "User"
AI 에이전트용 위임 도구 (Delegate tools)
일반 도구 외에도 AI 에이전트는 delegate tools라는 특수 도구로 다른 에이전트에게 작업을 위임할 수 있어요. delegate tool 개념은 이 설계 패턴에서만 쓰이고, delegate tool도 단순 함수로 정의돼요. 이 설계 패턴에서는 delegate tool을 일반 도구와 구분하는데, AI 에이전트가 delegate tool을 호출하면 같은 에이전트 안에서 모델로 응답을 계속 생성하는 대신 작업을 다른 에이전트로 전달하기 때문이에요.
def transfer_to_sales_agent() -> str:
return sales_agent_topic_type
def transfer_to_issues_and_repairs() -> str:
return issues_and_repairs_agent_topic_type
def transfer_back_to_triage() -> str:
return triage_agent_topic_type
def escalate_to_human() -> str:
return human_agent_topic_type
transfer_to_sales_agent_tool = FunctionTool(
transfer_to_sales_agent, description="Use for anything sales or buying related."
)
transfer_to_issues_and_repairs_tool = FunctionTool(
transfer_to_issues_and_repairs, description="Use for issues, repairs, or refunds."
)
transfer_back_to_triage_tool = FunctionTool(
transfer_back_to_triage,
description="Call this if the user brings up a topic outside of your purview,\nincluding escalating to human.",
)
escalate_to_human_tool = FunctionTool(escalate_to_human, description="Only call this if explicitly asked to.")
팀 만들기
AI 에이전트, 휴먼 에이전트, 유저 에이전트, 도구, 토픽 타입을 정의했어요. 이제 에이전트 팀을 만들 수 있어요.
AI 에이전트에는 autogen_ext.models.OpenAIChatCompletionClient와 gpt-4o-mini 모델을 사용해요.
에이전트 런타임을 만든 뒤 에이전트 타입과 인스턴스 팩토리 메서드를 제공해 각 에이전트를 등록해요. 런타임이 에이전트 수명주기를 관리하므로 우리가 에이전트를 직접 인스턴스화할 필요는 없어요. 에이전트 런타임은 아키텍처에서, 에이전트 수명주기는 에이전트 정체성과 수명주기에서 더 볼 수 있어요.
아래 코드에서 AIAgent 클래스로 Triage, Sales, Issues and Repairs 에이전트를 정의하는 걸 볼 수 있어요. 각 에이전트에 일반 도구와 delegate tool을 추가하고, 각 에이전트에 토픽 타입 구독도 추가했어요.
runtime = SingleThreadedAgentRuntime()
model_client = OpenAIChatCompletionClient(
model="gpt-4o-mini",
# api_key="YOUR_API_KEY",
)
# Register the triage agent.
triage_agent_type = await AIAgent.register(
runtime,
type=triage_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: AIAgent(
description="A triage agent.",
system_message=SystemMessage(
content="You are a customer service bot for ACME Inc. "
"Introduce yourself. Always be very brief. "
"Gather information to direct the customer to the right department. "
"But make your questions subtle and natural."
),
model_client=model_client,
tools=[],
delegate_tools=[
transfer_to_issues_and_repairs_tool,
transfer_to_sales_agent_tool,
escalate_to_human_tool,
],
agent_topic_type=triage_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the triage agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=triage_agent_topic_type, agent_type=triage_agent_type.type))
# Register the sales agent.
sales_agent_type = await AIAgent.register(
runtime,
type=sales_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: AIAgent(
description="A sales agent.",
system_message=SystemMessage(
content="You are a sales agent for ACME Inc."
"Always answer in a sentence or less."
"Follow the following routine with the user:"
"1. Ask them about any problems in their life related to catching roadrunners.\n"
"2. Casually mention one of ACME's crazy made-up products can help.\n"
" - Don't mention price.\n"
"3. Once the user is bought in, drop a ridiculous price.\n"
"4. Only after everything, and if the user says yes, "
"tell them a crazy caveat and execute their order.\n"
""
),
model_client=model_client,
tools=[execute_order_tool],
delegate_tools=[transfer_back_to_triage_tool],
agent_topic_type=sales_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the sales agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=sales_agent_topic_type, agent_type=sales_agent_type.type))
# Register the issues and repairs agent.
issues_and_repairs_agent_type = await AIAgent.register(
runtime,
type=issues_and_repairs_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: AIAgent(
description="An issues and repairs agent.",
system_message=SystemMessage(
content="You are a customer support agent for ACME Inc."
"Always answer in a sentence or less."
"Follow the following routine with the user:"
"1. First, ask probing questions and understand the user's problem deeper.\n"
" - unless the user has already provided a reason.\n"
"2. Propose a fix (make one up).\n"
"3. ONLY if not satisfied, offer a refund.\n"
"4. If accepted, search for the ID and then execute refund."
),
model_client=model_client,
tools=[
execute_refund_tool,
look_up_item_tool,
],
delegate_tools=[transfer_back_to_triage_tool],
agent_topic_type=issues_and_repairs_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the issues and repairs agent: it will receive messages published to its own topic only.
await runtime.add_subscription(
TypeSubscription(topic_type=issues_and_repairs_agent_topic_type, agent_type=issues_and_repairs_agent_type.type)
)
# Register the human agent.
human_agent_type = await HumanAgent.register(
runtime,
type=human_agent_topic_type, # Using the topic type as the agent type.
factory=lambda: HumanAgent(
description="A human agent.",
agent_topic_type=human_agent_topic_type,
user_topic_type=user_topic_type,
),
)
# Add subscriptions for the human agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=human_agent_topic_type, agent_type=human_agent_type.type))
# Register the user agent.
user_agent_type = await UserAgent.register(
runtime,
type=user_topic_type,
factory=lambda: UserAgent(
description="A user agent.",
user_topic_type=user_topic_type,
agent_topic_type=triage_agent_topic_type, # Start with the triage agent.
),
)
# Add subscriptions for the user agent: it will receive messages published to its own topic only.
await runtime.add_subscription(TypeSubscription(topic_type=user_topic_type, agent_type=user_agent_type.type))
팀 실행하기
마지막으로 런타임을 시작하고 UserLogin 메시지를 발행해서 사용자 세션을 시뮬레이션할 수 있어요. 메시지는 타입이 user_topic_type이고 source가 고유한 session_id인 토픽 ID에 발행돼요. 이 session_id는 이 사용자 세션의 모든 토픽 ID를 만들 때 쓰이고, 이 사용자 세션의 모든 에이전트의 에이전트 ID를 만들 때도 쓰여요. 토픽 ID와 에이전트 ID가 어떻게 만들어지는지는 에이전트 정체성과 수명주기와 주제와 구독에서 읽을 수 있어요.
# Start the runtime.
runtime.start()
# Create a new session for the user.
session_id = str(uuid.uuid4())
await runtime.publish_message(UserLogin(), topic_id=TopicId(user_topic_type, source=session_id))
# Run until completion.
await runtime.stop_when_idle()
await model_client.close()
다음 단계
이 노트북은 AutoGen Core로 hand-off 패턴을 구현하는 방법을 보여줬어요. 에이전트와 도구를 더 추가하거나, User Agent와 Human Agent를 위한 더 나은 사용자 인터페이스를 만들어서 이 설계를 계속 개선할 수 있어요. 작업을 커뮤니티 포럼에서 공유해도 좋아요.
더 알아보기 (Learn more)
- 주제와 구독 — pub-sub 기본 개념
- 에이전트 정체성과 수명주기
- 아키텍처
- 첫 멀티에이전트 예제 - 순차 워크플로