¿Cómo OpenAI Swarm mejora la colaboración entre múltiples agentes?

El marco Swarm de OpenAI está diseñado para crear un entorno flexible y fácil de usar para coordinar múltiples agentes. Si bien está destinado principalmente para uso educativo y experimental, OpenAI desaconseja el uso de Swarm en entornos de producción, pero es un marco que vale la pena explorar. Su objetivo principal es demostrar los conceptos de “traspasos” y “rutinas”, patrones que ayudan a los agentes a colaborar de manera eficiente. Swarm no es una biblioteca independiente sino una herramienta para explorar estos patrones. Profundicemos en qué son las rutinas y los traspasos y cómo desempeñan un papel en la orquestación del comportamiento de los agentes.

Descripción general

1. OpenAI Swarm es un marco diseñado para coordinar múltiples agentes a través de rutinas y traspasos.
2. Ofrece un entorno fácil de usar ideal para fines educativos y experimentales.
3. Swarm no está diseñado para uso en producción, pero sirve como herramienta de aprendizaje para la orquestación de múltiples agentes.
4. El marco ayuda a los desarrolladores a comprender los patrones de colaboración de los agentes, mejorando la flexibilidad y la ejecución de tareas.
5. Swarm enfatiza la interacción fluida de los agentes, haciendo que los sistemas complejos sean manejables sin una curva de aprendizaje pronunciada.
6. A través de ejemplos prácticos, Swarm ilustra cómo las rutinas y los traspasos pueden optimizar el comportamiento y la coordinación de los agentes.

¿Qué es OpenAI Swarm?

OpenAI ha agrupado estas ideas en una biblioteca de muestra llamada Swarm, diseñada como prueba de concepto. Si bien Swarm no está diseñado para uso en producción, sirve como un excelente punto de partida para la experimentación, ya que ofrece ideas y códigos que puede aprovechar para crear sus propios sistemas.

Swarm se centra en hacer que la coordinación de agentes y la ejecución de tareas sean ligeras, fáciles de controlar y sencillas de probar. Para ello, se basa en dos conceptos centrales: agentes y traspasos. Un Agente representa un conjunto de instrucciones y herramientas y, en cualquier momento, puede traspasar una conversación a otro Agente.

Estas abstracciones centrales son lo suficientemente poderosas como para modelar interacciones complejas entre herramientas y redes de agentes. Esto hace posible la construcción de sistemas escalables del mundo real sin enfrentar una curva de aprendizaje pronunciada.

¿Por qué utilizar OpenAI Swarm?

AbiertoAI Swarm explora patrones livianos, escalables e inherentemente personalizables. Es ideal para escenarios que involucran muchas tareas e instrucciones independientes, que son difíciles de capturar en un solo mensaje.

La API de Asistentes podría ser más adecuada para los desarrolladores que buscan soluciones totalmente alojadas con administración de memoria integrada. Sin embargo, Swarm es un recurso educativo fantástico para aquellos que quieran sumergirse en la mecánica de la orquestación multiagente. Swarm, que se ejecuta principalmente en el cliente, es similar a la API Chat Completions y no almacena el estado entre llamadas, lo que la convierte en una herramienta eficaz para aprender y experimentar.

Ejemplo de uso del marco OpenAI Swarm

Este código demuestra cómo El marco Swarm de OpenAI puede hacer que la colaboración de los agentes sea divertida, flexible y dinámica. ¡Vamos a sumergirnos en lo que está pasando aquí!

Preparando el escenario

Primero, importamos lo esencial:

from swarm import Swarm, Agent
client = Swarm()

Esto crea el cliente Swarm, que organiza las interacciones entre nuestros agentes. Piense en él como el cerebro detrás de escena, asegurándose de que los agentes hagan lo suyo.

Los agentes suben al escenario

A continuación, definimos una función simple pero crucial:

def transfer_to_agent_b():
   return agent_b

Esta función es la mecánica de transferencia. Permite al Agente A pasar cortésmente la conversación al Agente B cuando sea el momento adecuado.

Ahora, conozcamos a los agentes:

agent_a = Agent(
   name="Agent A",
   instructions="You are a helpful agent.",
   functions=(transfer_to_agent_b),
)

agent_b = Agent(
   name="Agent B",
   instructions="Only speak in Haikus.",

)

El Agente A es su amigable ayudante: siempre está listo para ayudar, pero también es lo suficientemente inteligente como para saber cuándo es el momento de llamar a un colega. El Agente B es un poco más poético y misterioso, y solo se comunica en la elegante forma de haikus.

Ahora lo reunimos todo:

response = client.run(
   agent=agent_a,
   messages=({"role": "user", "content": "I want to talk to agent B."}),
)


print(response.messages(-1)("content"))

Esto inicia una conversación con el Agente A, pero el usuario solicita un chat con el Agente B. Gracias a la función transfer_to_agent_b, el Agente A reconoce que es momento de hacerse a un lado y deja que el Agente B tome el relevo. El Agente B, fiel a su forma, responderá en haikus, ¡agregando un toque creativo a la interacción!

Producción

Creación de un complejo sistema multiagente de servicio al cliente

Abordaremos esto entendiendo cómo funcionan la rutina y los traspasos en el enjambre.

Importando nuestras dependencias

from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json

client = OpenAI()

Rutinas

Una “rutina” no está definida rígidamente sino que captura la idea de una secuencia de acciones. Piense en ello como un conjunto de instrucciones en lenguaje natural (proporcionadas a través de un mensaje del sistema) y las herramientas necesarias para llevarlas a cabo.

Analicémoslo con un ejemplo:

Imagine crear un agente de servicio al cliente que ayude a los usuarios a resolver sus problemas. El agente sigue estos pasos:

1. Deducir la información: Primero, el agente pregunta al usuario sobre el problema al que se enfrenta.
2. Solicite más detalles (si es necesario): Si el agente necesita más información para comprender el problema, hace preguntas de seguimiento.
3. Proporcionar una solución: Según la información, el agente sugiere una solución para solucionar el problema.
4. Ofrezca un reembolso (si es necesario): Si el usuario aún no está satisfecho, el agente ofrece un reembolso.
5. Procesar el reembolso: Si se acepta el reembolso, el agente encontrará la identificación correspondiente y completará el proceso de reembolso.

Este proceso paso a paso ayuda al agente a resolver de manera eficiente los problemas del usuario y, al mismo tiempo, garantiza que el usuario esté satisfecho.

El verdadero poder de las rutinas reside en su simplicidad y adaptabilidad. Observe cómo las tareas son condicionales, muy parecidas a las ramas de una máquina de estados. Pero las rutinas van un paso más allá. Con una adherencia “suave”, el modelo de lenguaje grande (LLM) no se queda estancado en un bucle; guía hábilmente la conversación, haciendo que estas rutinas sean altamente efectivas para tareas pequeñas y medianas.

Aquí está el enlace de GitHub a Enjambre.

Ejecutar rutinas

Comience con un bucle básico: recopile la entrada del usuario, agregue el mensaje al historial de conversaciones, llame al modelo y luego agregue la respuesta del modelo al historial.

def run_full_turn(system_message, messages):
   response = client.chat.completions.create(
       model="gpt-4o-mini",
       messages=({"role": "system", "content": system_message}) + messages,
   )
   message = response.choices(0).message
   messages.append(message)


   if message.content: print("Assistant:", message.content)


   return message

messages = ()
while True:
   user = input("User: ")
   messages.append({"role": "user", "content": user})


   run_full_turn(system_message, messages)

Como aún no hemos integrado las llamadas a funciones, debemos agregarlas a continuación. Las funciones deben formatearse como esquemas de funciones de acuerdo con las especificaciones del modelo. Para facilitar esto, podemos crear una función auxiliar que convierta funciones de Python al formato de esquema correcto.

import inspect

def function_to_schema(func) -> dict:
   type_map = {
       str: "string",
       int: "integer",
       float: "number",
       bool: "boolean",
       list: "array",
       dict: "object",
       type(None): "null",
   }


   try:
       signature = inspect.signature(func)
   except ValueError as e:
       raise ValueError(
           f"Failed to get signature for function {func.__name__}: {str(e)}"
       )

   parameters = {}
   for param in signature.parameters.values():
       try:
           param_type = type_map.get(param.annotation, "string")
       except KeyError as e:
           raise KeyError(
               f"Unknown type annotation {param.annotation} for parameter {param.name}: {str(e)}"
           )
       parameters(param.name) = {"type": param_type}

   required = (
       param.name
       for param in signature.parameters.values()
       if param.default == inspect._empty
   )

   return {
       "type": "function",
       "function": {
           "name": func.__name__,
           "description": (func.__doc__ or "").strip(),
           "parameters": {
               "type": "object",
               "properties": parameters,
               "required": required,
           },
       },
   }

1. Ahora, cuando el modelo activa una herramienta, debemos ejecutar la función adecuada y devolver el resultado. Esto se puede hacer asignando nombres de herramientas a funciones de Python en un
2. En la práctica, deberíamos permitir que el modelo reaccione de manera diferente según el resultado de la llamada a la herramienta. Este proceso puede repetirse mientras haya más llamadas a herramientas, ya que la respuesta del modelo puede provocar otra. Así es como se ve el bucle cuando juntamos todo:

# Customer Service Routine

system_message = (
   "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 satesfied, offer a refund.\n"
   "4. If accepted, search for the ID and then execute refund."
   ""
)

def look_up_item(search_query):
   """Use to find item ID.
   Search query can be a description or keywords."""

   # return hard-coded item ID - in reality would be a lookup
   return "item_132612938"

def execute_refund(item_id, reason="not provided"):

   print("Summary:", item_id, reason) # lazy summary
   return "success"

tools = (execute_refund, look_up_item)

def run_full_turn(system_message, tools, messages):

   num_init_messages = len(messages)
   messages = messages.copy()

   while True:

       # turn python functions into tools and save a reverse map
       tool_schemas = (function_to_schema(tool) for tool in tools)
       tools_map = {tool.__name__: tool for tool in tools}

       # === 1. get openai completion ===
       response = client.chat.completions.create(
           model="gpt-4o-mini",
           messages=({"role": "system", "content": system_message}) + messages,
           tools=tool_schemas or None,
       )
       message = response.choices(0).message
       messages.append(message)

       if message.content:  # print assistant response
           print("Assistant:", message.content)

       if not message.tool_calls:  # if finished handling tool calls, break
           break

       # === 2. handle tool calls ===

       for tool_call in message.tool_calls:
           result = execute_tool_call(tool_call, tools_map)

           result_message = {
               "role": "tool",
               "tool_call_id": tool_call.id,
               "content": result,
           }
           messages.append(result_message)

   # ==== 3. return new messages =====
   return messages(num_init_messages:)

def execute_tool_call(tool_call, tools_map):
   name = tool_call.function.name
   args = json.loads(tool_call.function.arguments)

   print(f"Assistant: {name}({args})")

   # call corresponding function with provided arguments
   return tools_map(name)(**args)

messages = ()
while True:
   user = input("User: ")
   messages.append({"role": "user", "content": user})


   new_messages = run_full_turn(system_message, tools, messages)
   messages.extend(new_messages)

Una vez que la rutina básica esté en funcionamiento, podemos considerar agregar más pasos y herramientas. Al cargar las herramientas y procesos necesarios, podemos ampliar las rutinas para manejar diferentes tipos de solicitudes de los usuarios. Sin embargo, a medida que intentamos ampliar las rutinas a través de demasiadas tareas, es posible que comiencen a fallar.

Ahí es donde resulta útil el concepto de rutinas múltiples. Podemos cambiar a la rutina adecuada con las herramientas adecuadas para manejar las diferentes solicitudes de los usuarios. Al principio, cambiar dinámicamente herramientas e instrucciones puede parecer complejo. Pero si pensamos en las rutinas como “agentes” individuales, el concepto de transferencias lo hace más fácil: un agente puede simplemente pasar la conversación a otro, manteniendo el flujo de trabajo fluido.

Lea también: Los 4 principales patrones de diseño de IA agente para diseñar sistemas de IA

Transferencias en el marco OpenAI Swarm

De manera similar a ser transferido a otro representante durante una llamada telefónica, un “traspaso” en el marco de Swarm ocurre cuando un agente (o rutina) pasa una conversación en curso a otro. Pero a diferencia de las transferencias en la vida real, estos agentes son plenamente conscientes de sus interacciones anteriores, lo que garantiza una transición sin problemas.

Para implementar transferencias en el código, primero debemos definir una clase para un Agente. Esto permitirá a los agentes gestionar conversaciones y transferirlas cuando sea necesario.

class Agent(BaseModel):
   name: str = "Agent"
   model: str = "gpt-4o-mini"
   instructions: str = "You are a helpful Agent"
   tools: list = ()

A continuación, modificaremos el código de rutina existente para admitir agentes. En lugar de pasar un mensaje_sistema y herramientas directamente a la función run_full_turn, haremos que acepte un objeto Agente.

def run_full_turn(agent, messages):


   num_init_messages = len(messages)
   messages = messages.copy()


   while True:


       # turn python functions into tools and save a reverse map
       tool_schemas = (function_to_schema(tool) for tool in agent.tools)
       tools_map = {tool.__name__: tool for tool in agent.tools}


       # === 1. get openai completion ===
       response = client.chat.completions.create(
           model=agent.model,
           messages=({"role": "system", "content": agent.instructions}) + messages,
           tools=tool_schemas or None,
       )
       message = response.choices(0).message
       messages.append(message)


       if message.content:  # print assistant response
           print("Assistant:", message.content)


       if not message.tool_calls:  # if finished handling tool calls, break
           break


       # === 2. handle tool calls ===


       for tool_call in message.tool_calls:
           result = execute_tool_call(tool_call, tools_map)


           result_message = {
               "role": "tool",
               "tool_call_id": tool_call.id,
               "content": result,
           }
           messages.append(result_message)


   # ==== 3. return new messages =====
   return messages(num_init_messages:)

def execute_tool_call(tool_call, tools_map):
   name = tool_call.function.name
   args = json.loads(tool_call.function.arguments)


   print(f"Assistant: {name}({args})")


   # call corresponding function with provided arguments
   return tools_map(name)(**args)

Con esta configuración, ejecutar varios agentes resulta sencillo:

def execute_refund(item_name):
   return "success"


refund_agent = Agent(
   name="Refund Agent",
   instructions="You are a refund agent. Help the user with refunds.",
   tools=(execute_refund),
)


def place_order(item_name):
   return "success"


sales_assistant = Agent(
   name="Sales Assistant",
   instructions="You are a sales assistant. Sell the user a product.",
   tools=(place_order),
)

messages = ()
user_query = "Place an order for a black boot."
print("User:", user_query)
messages.append({"role": "user", "content": user_query})


response = run_full_turn(sales_assistant, messages) # sales assistant
messages.extend(response)

user_query = "Actually, I want a refund." # implitly refers to the last item
print("User:", user_query)
messages.append({"role": "user", "content": user_query})
response = run_full_turn(refund_agent, messages) # refund agent

En este ejemplo, las transferencias se realizan manualmente, pero lo ideal es que los agentes se pasen tareas entre sí automáticamente. Una forma sencilla de lograr esto es mediante llamadas a funciones. Cada agente puede invocar una función de transferencia específica, como transfer_to_xxx, para transferir sin problemas la conversación al siguiente agente de la fila.

¡Este método permite a los agentes manejar conversaciones sin problemas, sin intervención manual!

Funciones de transferencia

Ahora que nuestro agente puede comunicar su intención de transferir una tarea, debemos implementar la transferencia real. Si bien hay varias formas de hacer esto, hay disponible un enfoque particularmente elegante.

Hasta ahora, hemos estado devolviendo cadenas de las funciones de nuestro agente, como ejecutar_reembolso o colocar_orden. Pero, ¿qué pasa si devolvemos un objeto Agente cuando llega el momento de realizar la transferencia en lugar de simplemente devolver una cadena? Por ejemplo:

refund_agent = Agent(
   name="Refund Agent",
   instructions="You are a refund agent. Help the user with refunds.",
   tools=(execute_refund),
)

def transfer_to_refunds():
   return refund_agent

sales_assistant = Agent(
   name="Sales Assistant",
   instructions="You are a sales assistant. Sell the user a product.",
   tools=(place_order),
)

Ahora, actualicemos la función run_full_turn para acomodar este tipo de transferencia:

def run_full_turn(agent, messages):

   current_agent = agent
   num_init_messages = len(messages)
   messages = messages.copy()

   while True:

       # turn python functions into tools and save a reverse map
       tool_schemas = (function_to_schema(tool) for tool in current_agent.tools)
       tools = {tool.__name__: tool for tool in current_agent.tools}

       # === 1. get openai completion ===
       response = client.chat.completions.create(
           model=agent.model,
           messages=({"role": "system", "content": current_agent.instructions})
           + messages,
           tools=tool_schemas or None,
       )
       message = response.choices(0).message
       messages.append(message)


       if message.content:  # print agent response
           print(f"{current_agent.name}:", message.content)


       if not message.tool_calls:  # if finished handling tool calls, break
           break


       # === 2. handle tool calls ===


       for tool_call in message.tool_calls:
           result = execute_tool_call(tool_call, tools, current_agent.name)


           if type(result) is Agent:  # if agent transfer, update current agent
               current_agent = result
               result = (
                   f"Transfered to {current_agent.name}. Adopt persona immediately."
               )


           result_message = {
               "role": "tool",
               "tool_call_id": tool_call.id,
               "content": result,
           }
           messages.append(result_message)


   # ==== 3. return last agent used and new messages =====
   return Response(agent=current_agent, messages=messages(num_init_messages:))

def execute_tool_call(tool_call, tools, agent_name):
   name = tool_call.function.name
   args = json.loads(tool_call.function.arguments)

   print(f"{agent_name}:", f"{name}({args})")

   return tools(name)(**args)  # call corresponding function with provided arguments

Echemos un vistazo a un ejemplo en el que participan varios agentes, lo que les permite transferir tareas entre sí:

def escalate_to_human(summary):
   """Only call this if explicitly asked to."""
   print("Escalating to human agent...")
   print("\n=== Escalation Report ===")
   print(f"Summary: {summary}")
   print("=========================\n")
   exit()

def transfer_to_sales_agent():
   """User for anything sales or buying related."""
   return sales_agent

def transfer_to_issues_and_repairs():
   """User for issues, repairs, or refunds."""
   return issues_and_repairs_agent

def transfer_back_to_triage():
   """Call this if the user brings up a topic outside of your purview,
   including escalating to human."""
   return triage_agent

triage_agent = Agent(
   name="Triage Agent",
   instructions=(
       "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."
   ),
   tools=(transfer_to_sales_agent, transfer_to_issues_and_repairs, escalate_to_human),
)

def execute_order(product, price: int):
   """Price should be in USD."""
   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."

sales_agent = Agent(
   name="Sales Agent",
   instructions=(
       "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"
       ""
   ),
   tools=(execute_order, transfer_back_to_triage),
)

def look_up_item(search_query):
   """Use to find item ID.
   Search query can be a description or keywords."""
   item_id = "item_132612938"
   print("Found item:", item_id)
   return item_id

def execute_refund(item_id, reason="not provided"):
   print("\n\n=== Refund Summary ===")
   print(f"Item ID: {item_id}")
   print(f"Reason: {reason}")
   print("=================\n")
   print("Refund execution successful!")
   return "success"

issues_and_repairs_agent = Agent(
   name="Issues and Repairs Agent",
   instructions=(
       "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."
       ""
   ),
   tools=(execute_refund, look_up_item, transfer_back_to_triage),
)

Finalmente, podemos ejecutar esto en bucle para ver todo en acción. Dado que esto no funcionará directamente en un cuaderno de Python, pruébelo en un archivo de Python separado:

agent = triage_agent
messages = ()

while True:
   user = input("User: ")
   messages.append({"role": "user", "content": user})


   response = run_full_turn(agent, messages)
   agent = response.agent
   messages.extend(response.messages)

Con este método, los agentes pueden transferirse tareas entre sí sin problemas, lo que permite transiciones fluidas sin complejidad adicional.

Además, para comprender mejor la IA del agente, explore: El programa pionero de la IA del agente

Conclusión

El marco OpenAI Swarm proporciona un enfoque innovador para coordinar múltiples agentes de una manera dinámica y fácil de usar. Al centrarse en los principios de rutinas y traspasos, Swarm facilita interacciones fluidas entre agentes, permitiéndoles trabajar de forma colaborativa y adaptable para satisfacer las solicitudes de los usuarios.

Este marco simplifica la gestión de los comportamientos de los agentes y mejora la experiencia general del usuario al garantizar transiciones fluidas y continuidad en las conversaciones. Con su arquitectura liviana y personalizable, Swarm sirve como un excelente punto de partida para los desarrolladores que buscan explorar la orquestación multiagente en sus aplicaciones.

Si bien puede que no sea adecuado para uso en producción, Swarm se destaca como un valioso recurso educativo que inspira a los desarrolladores a construir sus propios sistemas y comprender las complejidades de la coordinación de agentes. A medida que experimente con Swarm, descubrirá nuevas posibilidades para crear interacciones atractivas y receptivas en sus proyectos. Ya sea para aprender o experimentar, Swarm es un ejemplo de cómo aprovechar el poder de los agentes impulsados ​​por IA para abordar tareas complejas de manera efectiva.

Preguntas frecuentes