> For the complete documentation index, see [llms.txt](https://saptiva.gitbook.io/saptiva-docs/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://saptiva.gitbook.io/saptiva-docs/saptiva-agents/patrones-de-diseno-multi-agente/transferencia-de-tareas-handoffs.md). # Transferencia De Tareas (Handoffs) 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 [**Tema & Suscripción**](/saptiva-docs/saptiva-agents/conceptos-del-nucleo/tema-and-suscripcion-topic-and-subscription.md) 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`. ```python 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 SAPTIVA_LEGACY 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. ```python 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í. ```python 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**. ```python 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), ) ``` *** ### Clase `UserAgent` 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. ```python 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`. ```python 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="") ``` *** ### Tipos de Tópico para los Agentes 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 [**Temas & Suscripciones**](/saptiva-docs/saptiva-agents/conceptos-del-nucleo/tema-and-suscripcion-topic-and-subscription.md). ```python 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" ``` *** ### Herramientas Delegadas para los Agentes de IA 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**. ```python 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 `Saptiva Legacy`. 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. ```python runtime = SingleThreadedAgentRuntime() model_client = SaptivaAIChatCompletionClient( model=SAPTIVA_LEGACY, 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. ```python # 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. ```