El patrón de diseño handoff es un enfoque multi-agente introducido por OpenAI en un proyecto experimental llamado Swarm. La idea clave es permitir que un agente delegue tareas a otros agentes utilizando una llamada especial a herramientas.
Esto ofrece varias ventajas frente a la implementación de OpenAI:
Se puede escalar a entornos distribuidos mediante distributed agent runtime.
Permite traer tu propia implementación de agentes con flexibilidad.
La API nativa basada en async facilita la integración con interfaces de usuario u otros sistemas.
Este ejemplo demuestra una implementación sencilla del patrón handoff. Te recomendamos leer primero sobre para entender cómo funciona el modelo pub-sub (publicador-suscriptor) y los agentes orientados a eventos.
Escenario
Imaginemos un escenario de atención al cliente en el que un cliente quiere solicitar un reembolso o comprar un nuevo producto a través de un chatbot. El chatbot está compuesto por un equipo multi-agente que incluye:
🧠 Triage Agent: entiende la solicitud del cliente y decide a quién delegar la tarea.
💸 Refund Agent: se encarga de solicitudes de reembolso.
🛍️ Sales Agent: maneja solicitudes de compra.
👨💼 Human Agent: se encarga de solicitudes complejas que los otros agentes no pueden resolver.
El cliente interactúa mediante un User Agent.
El siguiente diagrama muestra la topología de interacción entre los agentes.
Vamos a implementar este escenario paso a paso usando el core de Saptiva-Agents.
import json
import uuid
from typing import List, Tuple
from saptiva_agents.core import (
FunctionCall,
MessageContext,
RoutedAgent,
SingleThreadedAgentRuntime,
TopicId,
TypeSubscription,
message_handler,
)
from saptiva_agents.models import (
AssistantMessage,
ChatCompletionClient,
FunctionExecutionResult,
FunctionExecutionResultMessage,
LLMMessage,
SystemMessage,
UserMessage,
)
from saptiva_agents import LLAMA_MODEL
from saptiva_agents.tools import FunctionTool, Tool
from saptiva_agents.base import SaptivaAIChatCompletionClient
from pydantic import BaseModel
Protocolo de Mensajes
Primero definimos los tipos de mensajes que utilizarán los agentes para comunicarse.
UserLogin: publicado cuando un usuario inicia sesión y comienza una sesión.
UserTask: contiene el historial del chat de la sesión. También es publicado al hacer un handoff.
AgentResponse: publicado por los agentes IA o humanos, contiene historial y el topic para responder.
class UserLogin(BaseModel):
pass
class UserTask(BaseModel):
context: List[LLMMessage]
class AgentResponse(BaseModel):
reply_to_topic_type: str
context: List[LLMMessage]
Clase AIAgent
Clase base para los agentes IA (Triage, Sales, Refund). Utiliza un modelo LLM (via ChatCompletionClient) para generar respuestas y ejecutar herramientas.
Los agentes:
Generan una respuesta.
Si es una llamada a herramienta especial (handoff), publican un UserTask al topic correspondiente.
Si es una herramienta común, la ejecutan y siguen generando respuesta.
Si la respuesta es texto, la publican como AgentResponse para el cliente.
En bloques de código como este, los comentarios se dejan ya traducidos, por lo tanto no repito el código aquí.
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),
)
Clase HumanAgent
En el método handle_user_task, envía una notificación a través de una aplicación de chat como Microsoft Teams o Slack.
La aplicación de chat publica la respuesta del humano a través del runtime, al tópico especificado por agent_topic_type.
Crea otro manejador de mensajes para procesar la respuesta del humano y enviarla de vuelta al cliente.
En esta implementación, el HumanAgent simplemente utiliza la consola para recibir tu entrada.
En una aplicación del mundo real, puedes mejorar este diseño de la siguiente manera:
La clase HumanAgent es un proxy del humano en el chatbot. Se utiliza para manejar solicitudes que los agentes de IA no pueden resolver. El HumanAgent se suscribe al tipo de tópico agent_topic_type para recibir mensajes, y publica al tipo de tópico user_topic_type para enviar mensajes al cliente.
La clase UserAgent es un proxy para el cliente que interactúa con el chatbot. Maneja dos tipos de mensajes: UserLogin y AgentResponse.
Cuando UserAgent recibe un mensaje UserLogin, inicia una nueva sesión con el chatbot y publica un mensaje UserTask al agente de IA que está suscrito al tipo de tópico agent_topic_type.
Cuando UserAgent recibe un mensaje AgentResponse, muestra al usuario la respuesta del chatbot.
En esta implementación, UserAgent utiliza la consola para obtener tu entrada.
En una aplicación del mundo real, puedes mejorar la interacción humana usando la misma idea descrita en la sección de HumanAgent más arriba.
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)
)
Herramientas para los Agentes
Los agentes de IA pueden utilizar herramientas regulares para completar tareas si no necesitan delegar la tarea a otros agentes.
Definimos las herramientas utilizando funciones simples y creamos las herramientas usando el contenedor FunctionTool.
Además de las herramientas regulares, los agentes de IA pueden delegar tareas a otros agentes utilizando herramientas especiales llamadas herramientas delegadas (delegate tools).
El concepto de herramienta delegada se utiliza únicamente en este patrón de diseño, y dichas herramientas también se definen como funciones simples.
Diferenciamos las herramientas delegadas de las herramientas regulares en este patrón porque cuando un agente de IA llama a una herramienta delegada, transferimos la tarea a otro agente en lugar de continuar generando respuestas con el modelo dentro del mismo agente.
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.")
Creando el Equipo
Ya hemos definido los agentes de IA, el agente humano, el agente usuario, las herramientas y los tipos de tópicos. Ahora podemos crear el equipo de agentes.
Para los agentes de IA, usamos SaptivaAIChatCompletionClient junto con el modelo llama3.3:70b.
Después de crear el runtime del agente, registramos cada agente proporcionando un tipo de agente y un método de fábrica para crear la instancia del agente. El runtime es responsable de gestionar el ciclo de vida de los agentes, por lo tanto, no necesitamos instanciar los agentes manualmente.
Puedes leer más sobre el runtime de agentes en [Agent Runtime Environments] y sobre el ciclo de vida en [Agent Identity and Lifecycle].
En el código a continuación, puedes ver que estamos utilizando la clase AIAgent para definir los agentes de Triage, Ventas, y Problemas y Reparaciones.
Añadimos herramientas regulares y herramientas delegadas a cada uno de ellos.
También agregamos las suscripciones a los tipos de tópico correspondientes para cada agente.
runtime = SingleThreadedAgentRuntime()
model_client = SaptivaAIChatCompletionClient(
model=LLAMA_MODEL,
api_key="TU_SAPTIVA_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))
Ejecutando el Equipo
Finalmente, podemos iniciar el runtime y simular una sesión de usuario publicando un mensaje UserLogin al runtime.
Este mensaje se publica a un topic_id con el tipo definido como user_topic_type y con un source establecido como un session_id único.
Este session_id se utilizará para crear todos los topic IDs en esta sesión de usuario, y también se usará para crear el agent ID de todos los agentes involucrados en esta misma sesión.
# 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()
--------------------------------------------------------------------------------
User login, session ID: 7a568cf5-13e7-4e81-8616-8265a01b3f2b.
--------------------------------------------------------------------------------
User:
I want a refund
--------------------------------------------------------------------------------
TriageAgent:
I can help with that! Could I ask what item you're seeking a refund for?
--------------------------------------------------------------------------------
User:
A pair of shoes I bought
--------------------------------------------------------------------------------
TriageAgent:
[FunctionCall(id='call_qPx1DXDL2NLcHs8QNo47egsJ', arguments='{}', name='transfer_to_issues_and_repairs')]
--------------------------------------------------------------------------------
TriageAgent:
Delegating to IssuesAndRepairsAgent
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
I see you're looking for a refund on a pair of shoes. Can you tell me what the issue is with the shoes?
--------------------------------------------------------------------------------
User:
The shoes are too small
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
I recommend trying a size up as a fix; would that work for you?
--------------------------------------------------------------------------------
User:
no I want a refund
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionCall(id='call_Ytp8VUQRyKFNEU36mLE6Dkrp', arguments='{"search_query":"shoes"}', name='look_up_item')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionExecutionResult(content='item_132612938', call_id='call_Ytp8VUQRyKFNEU36mLE6Dkrp')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionCall(id='call_bPm6EKKBy5GJ65s9OKt9b1uE', arguments='{"item_id":"item_132612938","reason":"not provided"}', name='execute_refund')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionExecutionResult(content='success', call_id='call_bPm6EKKBy5GJ65s9OKt9b1uE')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
Your refund has been successfully processed! If you have any other questions, feel free to ask.
--------------------------------------------------------------------------------
User:
I want to talk to your manager
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
I can help with that, let me transfer you to a supervisor.
--------------------------------------------------------------------------------
User:
Okay
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
[FunctionCall(id='call_PpmLZvwNoiDPUH8Tva3eAwHX', arguments='{}', name='transfer_back_to_triage')]
--------------------------------------------------------------------------------
IssuesAndRepairsAgent:
Delegating to TriageAgent
--------------------------------------------------------------------------------
TriageAgent:
[FunctionCall(id='call_jSL6IBm5537Dr74UbJSxaj6I', arguments='{}', name='escalate_to_human')]
--------------------------------------------------------------------------------
TriageAgent:
Delegating to HumanAgent
--------------------------------------------------------------------------------
HumanAgent:
Hello this is manager
--------------------------------------------------------------------------------
User:
Hi! Thanks for your service. I give you 5 stars!
--------------------------------------------------------------------------------
HumanAgent:
Thanks.
--------------------------------------------------------------------------------
User:
exit
--------------------------------------------------------------------------------
User session ended, session ID: 7a568cf5-13e7-4e81-8616-8265a01b3f2b.
Definimos los tipos de tópico a los que cada uno de los agentes se suscribirá.
Lee más sobre los tipos de tópico en la sección .
