Agente & Entorno De Ejecución De Agentes

En esta y la siguiente sección, nos enfocamos en los conceptos centrales de Saptiva-Agents: agentes, entorno de ejecución de agentes, mensajes y comunicación, que son los bloques fundamentales para construir aplicaciones multi-agente.

Nota

La API principal está diseñada para ser flexible y sin opiniones impuestas. Por lo tanto, a veces puede parecer compleja. Continúa si estás construyendo un sistema multi-agente interactivo, escalable y distribuido, y deseas tener control total sobre todos los flujos de trabajo.

Un agente en Saptiva-Agents es una entidad definida por la interfaz base Agent. Tiene un identificador único del tipo AgentId, y un diccionario de metadatos del tipo AgentMetadata.

En la mayoría de los casos, puedes subclasificar tus agentes desde la clase de nivel superior RoutedAgent, que te permite enrutar mensajes al manejador de mensajes correspondiente, especificado con el decorador @message_handler() y una pista de tipo apropiada para la variable del mensaje. Un entorno de ejecución de agente es el entorno donde los agentes se ejecutan en Saptiva-Agents.

Similar al entorno de ejecución de un lenguaje de programación, el entorno de ejecución de agentes proporciona la infraestructura necesaria para facilitar la comunicación entre agentes, gestionar sus ciclos de vida, imponer límites de seguridad y soportar monitoreo y depuración.

Para desarrollo local, los desarrolladores pueden usar SingleThreadedAgentRuntime, que puede ser embebido en una aplicación Python.

Nota

Los agentes no son instanciados ni gestionados directamente por el código de la aplicación. En su lugar, son creados y gestionados por el entorno de ejecución cuando se necesitan.

Si ya estás familiarizado con Saptiva-Agents, es importante notar que los agentes de Saptiva-Agents, como AssistantAgent, son creados por la aplicación y por lo tanto no están gestionados directamente por el entorno de ejecución. Para usar un agente de Saptiva-Agents en el Core, necesitas crear un agente envoltorio (wrapper) que delegue mensajes al agente de Saptiva-Agents y permita al runtime gestionar dicho agente.

---

Implementación de un Agente

Para implementar un agente, el desarrollador debe subclasificar RoutedAgent y definir un método manejador para cada tipo de mensaje que el agente espera manejar, usando el decorador @message_handler. Por ejemplo, el siguiente agente maneja un tipo de mensaje simple MyMessageType y lo imprime:

# Importar definiciones necesarias
from dataclasses import dataclass
from saptiva_agents.core import AgentId, MessageContext, RoutedAgent, message_handler

# Definir el tipo de mensaje
@dataclass
class MyMessageType:
    content: str

# Crear una clase de agente personalizado
class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyAgent")

    @message_handler  # Decorador para manejar mensajes de tipo MyMessageType
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} recibió mensaje: {message.content}")

Este agente solo maneja MyMessageType y los mensajes se entregan al método handle_my_message_type.

---

Uso de un Agente de Saptiva-Agents

Si tienes un agente de Saptiva-Agents y deseas usarlo con la API Core, puedes crear un agente envoltorio (wrapper) que delegue los mensajes al agente de Saptiva-Agents. Ejemplo:

# Importar desde Saptiva-Agents
from saptiva_agents.agents import AssistantAgent
from saptiva_agents.messages import TextMessage
from saptiva_agents.base import SaptivaAIChatCompletionClient

# Crear un agente envoltorio para AssistantAgent
class MyAssistant(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name)
        model_client = SaptivaAIChatCompletionClient(model="llama3.3:70b")
        self._delegate = AssistantAgent(name, model_client=model_client, system_message="Responde en español."))

    @message_handler  # Decorador para manejar mensajes personalizados
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} recibió mensaje: {message.content}")
        response = await self._delegate.on_messages(
            [TextMessage(content=message.content, source="user")], ctx.cancellation_token
        )
        print(f"{self.id.type} respondió: {response.chat_message.content}")

---

Registro del Tipo de Agente

Para que el entorno de ejecución pueda usar un agente, el desarrollador debe registrar el tipo de agente usando el método register() de la clase BaseAgent. Esto vincula un nombre único de tipo con una función fábrica que crea una instancia del agente.

El tipo de agente (AgentType) no es lo mismo que la clase del agente. En este ejemplo, el tipo de agente es AgentType("my_agent") o AgentType("my_assistant"), mientras que la clase del agente es la clase de Python MyAgent o MyAssistantAgent. Se espera que la función fábrica devuelva una instancia de la clase del agente sobre la cual se invoca el método de clase register().

Nota

Se pueden registrar diferentes tipos de agentes con funciones fábrica que devuelven la misma clase de agente. Por ejemplo, en las funciones fábrica, se pueden usar variaciones en los parámetros del constructor para crear distintas instancias de la misma clase de agente.

Para registrar nuestros tipos de agentes con SingleThreadedAgentRuntime, se puede usar el siguiente código:

# Registro de agentes con el runtime
from saptiva_agents.core import SingleThreadedAgentRuntime

runtime = SingleThreadedAgentRuntime()
await MyAgent.register(runtime, "my_agent", lambda: MyAgent())
await MyAssistant.register(runtime, "my_assistant", lambda: MyAssistant("my_assistant"))

AgentType(type='my_assistant')

Una vez registrado, puedes enviar mensajes directamente:

# Iniciar el entorno de ejecución en segundo plano
runtime.start()

# Enviar mensajes directos a los agentes
await runtime.send_message(MyMessageType("Hola, mundo!"), AgentId("my_agent", "default"))
await runtime.send_message(MyMessageType("Hola, mundo!"), AgentId("my_assistant", "default"))

# Detener el entorno
await runtime.stop()

Salida esperada:

my_agent recibió mensaje: Hola, mundo!
my_assistant recibió mensaje: Hola, mundo!
my_assistant respondió: ¡Hola! ¿Cómo puedo ayudarte hoy?

---

Ejecutar el Entorno de Ejecución de Agentes en Hilo Único

start() inicia una tarea en segundo plano para entregar mensajes:

runtime.start()
# ... enviar mensajes, publicar mensajes, etc.
await runtime.stop()  # Esto detiene la tarea, pero no cancela mensajes en proceso

Puedes usar stop_when_idle() si deseas esperar a que todos los mensajes se procesen:

runtime.start()
# ... enviar mensajes, etc.
await runtime.stop_when_idle()  # Espera hasta que el entorno esté inactivo

Para liberar recursos completamente:

await runtime.close()

Otros entornos de ejecución pueden tener diferentes métodos para operar.