Handoff Agent 오케스트레이션
Handoff 오케스트레이션
[!IMPORTANT] Agent Framework의 오케스트레이션 기능은 실험 단계(experimental)예요. 활발히 개발 중이며, preview/RC 단계로 넘어가기 전에 크게 바뀔 수 있어요.
핸드오프(handoff) 오케스트레이션 은 에이전트가 컨텍스트나 사용자 요청에 따라 서로에게 제어권을 넘겨주는 패턴이에요. 각 에이전트는 적절한 전문성을 가진 다른 에이전트에게 대화를 "인계"할 수 있고, 그래서 작업의 각 부분을 적절한 에이전트가 처리하게 됩니다. 고객 지원, 전문가 시스템, 또는 동적 위임이 필요한 모든 시나리오에서 특히 유용해요. 패턴을 언제 쓰고 피할지는 핸드오프 오케스트레이션 문서를 참고하세요.
흔한 사용 사례
고객 지원 에이전트가 일반 문의를 처리하다가, 문제 해결을 위해 기술 전문가 에이전트로, 필요하면 결제 담당 에이전트로 인계하는 식이에요.
배울 내용
- 에이전트와 그 핸드오프 관계를 정의하는 법
- 동적 에이전트 라우팅을 위한 핸드오프 오케스트레이션을 구성하는 법
- 대화 루프에 인간을 끌어들이는 법
특화 에이전트 정의하기
각 에이전트가 하나의 영역을 담당해요. 아래 예시는 트라이지(triage) 에이전트, 환불 에이전트, 주문 상태 에이전트, 주문 반품 에이전트로 구성돼요. 일부 에이전트는 특정 작업을 처리하기 위해 플러그인을 사용해요.
from semantic_kernel.functions import kernel_function
class OrderStatusPlugin:
@kernel_function
def check_order_status(self, order_id: str) -> str:
"""Check the status of an order."""
return f"Order {order_id} is shipped and will arrive in 2-3 days."
class OrderRefundPlugin:
@kernel_function
def process_refund(self, order_id: str, reason: str) -> str:
"""Process a refund for an order."""
print(f"Processing refund for order {order_id} due to: {reason}")
return f"Refund for order {order_id} has been processed successfully."
class OrderReturnPlugin:
@kernel_function
def process_return(self, order_id: str, reason: str) -> str:
"""Process a return for an order."""
print(f"Processing return for order {order_id} due to: {reason}")
return f"Return for order {order_id} has been processed successfully."
이 플러그인들을 사용하는 에이전트를 정의해요. ChatCompletionAgent 를 쓰되, 어떤 에이전트 타입이나 모델 서비스든 사용 가능해요.
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion
support_agent = ChatCompletionAgent(
name="TriageAgent",
description="A customer support agent that triages issues.",
instructions="Handle customer requests.",
service=OpenAIChatCompletion(),
)
refund_agent = ChatCompletionAgent(
name="RefundAgent",
description="A customer support agent that handles refunds.",
instructions="Handle refund requests.",
service=OpenAIChatCompletion(),
plugins=[OrderRefundPlugin()],
)
order_status_agent = ChatCompletionAgent(
name="OrderStatusAgent",
description="A customer support agent that checks order status.",
instructions="Handle order status requests.",
service=OpenAIChatCompletion(),
plugins=[OrderStatusPlugin()],
)
order_return_agent = ChatCompletionAgent(
name="OrderReturnAgent",
description="A customer support agent that handles order returns.",
instructions="Handle order return requests.",
service=OpenAIChatCompletion(),
plugins=[OrderReturnPlugin()],
)
핸드오프 관계 정의하기
OrchestrationHandoffs 로 어떤 에이전트가 어떤 에이전트에게, 어떤 상황에서 인계할 수 있는지 지정해요. 대상 에이전트에 전달되는 설명(description)이 라우팅 판단의 근거가 돼요.
from semantic_kernel.agents import OrchestrationHandoffs
handoffs = (
OrchestrationHandoffs()
.add_many( # 같은 소스 에이전트에 여러 핸드오프를 한 번에 추가
source_agent=support_agent.name,
target_agents={
refund_agent.name: "Transfer to this agent if the issue is refund related",
order_status_agent.name: "Transfer to this agent if the issue is order status related",
order_return_agent.name: "Transfer to this agent if the issue is order return related",
},
)
.add( # 단일 핸드오프 추가
source_agent=refund_agent.name,
target_agent=support_agent.name,
description="Transfer to this agent if the issue is not refund related",
)
.add(
source_agent=order_status_agent.name,
target_agent=support_agent.name,
description="Transfer to this agent if the issue is not order status related",
)
.add(
source_agent=order_return_agent.name,
target_agent=support_agent.name,
description="Transfer to this agent if the issue is not order return related",
)
)
에이전트 응답 관찰하기
대화가 진행되면서 각 에이전트의 메시지를 출력하는 콜백을 정의할 수 있어요.
from semantic_kernel.contents import ChatMessageContent
def agent_response_callback(message: ChatMessageContent) -> None:
print(f"{message.name}: {message.content}")
Human in the Loop
핸드오프 오케스트레이션의 핵심 기능은 인간이 대화에 참여할 수 있다는 점이에요. 에이전트가 사용자 입력이 필요할 때마다 호출되는 human_response_function 콜백을 제공해서 구현해요.
from semantic_kernel.contents import AuthorRole, ChatMessageContent
def human_response_function() -> ChatMessageContent:
user_input = input("User: ")
return ChatMessageContent(role=AuthorRole.USER, content=user_input)
핸드오프 오케스트레이션 구성
HandoffOrchestration 객체에 에이전트 목록, 핸드오프 관계, 콜백들을 넘겨요.
from semantic_kernel.agents import HandoffOrchestration
handoff_orchestration = HandoffOrchestration(
members=[
support_agent,
refund_agent,
order_status_agent,
order_return_agent,
],
handoffs=handoffs,
agent_response_callback=agent_response_callback,
human_response_function=human_response_function,
)
런타임 시작 ~ 호출 ~ 결과 수집
런타임을 시작하고, 초기 작업으로 오케스트레이션을 호출하면 에이전트들이 필요에 따라 대화를 라우팅하고, 필요할 때 인간을 끌어들여요. 끝나면 결과를 받아요.
from semantic_kernel.agents.runtime import InProcessRuntime
runtime = InProcessRuntime()
runtime.start()
orchestration_result = await handoff_orchestration.invoke(
task="A customer is on the line.",
runtime=runtime,
)
value = await orchestration_result.get()
print(value)
await runtime.stop_when_idle()
샘플 출력
트라이지 에이전트가 "주문 상태를 확인하고 싶어요"라는 사용자 요청을 받고 OrderStatusAgent 로 인계하고, 다시 반품 문의가 나오면 OrderReturnAgent 로 인계하는 흐름이 보여요. 인간이 주문 번호와 사유를 알려주면, 그 에이전트가 플러그인으로 반품을 처리해요.
TriageAgent: Hello! ... How can I assist you today?
User: I'd like to track the status of my order
OrderStatusAgent: ... Could you please provide me with your order ID?
User: My order ID is 123
OrderStatusAgent: Your order with ID 123 has been shipped and is expected to arrive in 2-3 days. ...
User: I want to return another order of mine
OrderReturnAgent: ... Please provide the order ID for the return and the reason ...
User: Order ID 321
OrderReturnAgent: Please provide the reason for returning the order with ID 321.
User: Broken item
OrderReturnAgent: The return for your order with ID 321 has been successfully processed ...
작업이 끝나면 "Handled order return for order ID 321 due to a broken item, and successfully processed the return." 같은 요약으로 마무리돼요.