Esta es la primera parte de una serie de varias partes sobre la creación de agentes con la API Asistente de OpenAI utilizando el SDK de Python.

¿Qué son los agentes?

A mi modo de ver, un agente es en realidad solo una pieza de software que aprovecha un LLM (modelo de lenguaje grande) y trata de imitar el comportamiento humano. Eso significa que no sólo puede conversar y comprender el lenguaje, sino que también puede realizar acciones que tienen un impacto en el mundo real. Estas acciones suelen denominarse herramientas.

En esta publicación de blog, exploraremos cómo crear un agente utilizando la API Asistente de OpenAI utilizando su SDK de Python. La parte 1 será sólo el esqueleto del asistente. Es decir, sólo la parte conversacional.

Elegí crear una aplicación CLI a propósito para que fuera independiente del marco. A propósito llamaremos Agente a nuestra implementación y nos referiremos a la implementación del SDK de OpenAI como Asistente para distinguir fácilmente entre los dos.

Utilizo los términos herramientas y funciones indistintamente cuando se trata de funciones que el Agente puede invocar. La parte 2 cubrirá la llamada a funciones con más detalle.

Requisitos previos

Para seguir este tutorial, necesitará lo siguiente:

• Python3 instalado en su máquina
• Una clave API de OpenAI
• Conocimientos básicos de programación Python.

Conceptos del asistente OpenAI

Asistente : un asistente en la API de Asistentes es una entidad configurada para responder a los mensajes de los usuarios. Utiliza instrucciones, un modelo elegido y herramientas para interactuar con funciones y proporcionar respuestas.

Hilo : un hilo representa una conversación o diálogo en la API de Asistentes. Se crea para cada interacción del usuario y puede contener múltiples mensajes, sirviendo como contenedor para la conversación en curso.

Mensaje : un mensaje es una unidad de comunicación en un hilo. Contiene texto (y potencialmente archivos en el futuro) y se utiliza para transmitir consultas de usuarios o respuestas de asistente dentro de un hilo.

Ejecutar : una ejecución es una instancia en la que el Asistente procesa un subproceso. Implica leer el hilo, decidir si llamar a las herramientas y generar respuestas basadas en la interpretación del modelo de los mensajes del hilo.

Configurar el entorno de desarrollo

El primer paso es crear un entorno virtual usando venv y activarlo. Esto asegurará que nuestras dependencias estén aisladas de la instalación de Python del sistema:

 python3 -m venv venv source venv/bin/activate

Instalemos nuestra única dependencia: el paquete openai :

 pip install openai

Cree un archivo main.py Completemos con algo de lógica de tiempo de ejecución básica para nuestra aplicación CLI:

 while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the assistant...") break print(f"Assistant: You said {user_input}")

Pruébelo ejecutando python3 main.py :

 python3 main.py User: hi Assistant: You said hi

Como puede ver, la CLI acepta un mensaje de usuario como entrada y nuestro genial Asistente aún no tiene cerebro 🧠, por lo que simplemente repite el mensaje. No tan inteligente todavía.

El agente

Ahora comienza la diversión 😁 (o los dolores de cabeza 🤕). Proporcionaré todas las importaciones necesarias para la clase final ahora mismo, para que no se dedique a pensar de dónde vienen las cosas, ya que mantuve las importaciones fuera de las muestras de código por motivos de brevedad. Comencemos construyendo una clase Agent en un nuevo archivo agent.py :

 import time import openai from openai.types.beta.threads.run import Run class Agent: def __init__(self, name: str, personality: str): self.name = name self.personality = personality self.client = openai.OpenAI(api_key="sk-*****") self.assistant = self.client.beta.assistants.create( name=self.name, model="gpt-4-turbo-preview" )

En el constructor de clases, inicializamos el cliente OpenAI como una propiedad de clase pasando nuestra clave API OpenAI. A continuación, creamos una propiedad de clase assistant que se asigna a nuestro Asistente recién creado. Almacenamos name y personality como propiedades de clase para su uso posterior.

El argumento name que le pasamos al método de creación es solo para identificar el Asistente en el panel de OpenAI, y la IA no lo sabe en este momento. De hecho tienes que pasar el nombre a las instructions que veremos más adelante.

Ya podías configurar instructions al crear el Asistente, pero en realidad hará que tu Asistente sea menos flexible a los cambios dinámicos.

Puede actualizar un Asistente llamando client.beta.assistants.update , pero hay un lugar mejor para pasar valores dinámicos que veremos cuando lleguemos a Ejecuciones.

Tenga en cuenta que si pasa instructions aquí y luego nuevamente al crear una ejecución, las instructions del Asistente se sobrescribirán con las instructions de la ejecución. No se complementan entre sí, así que elige uno según tus necesidades: nivel de asistente para instrucciones estáticas o nivel de ejecución para instrucciones dinámicas.

Para el modelo, elegí el modelo gpt-4-turbo-preview para que podamos agregar llamadas a funciones en la parte 2 de esta serie. Podrías usar gpt-3.5-turbo si quieres ahorrar algunas fracciones de un centavo mientras te provocas una migraña de pura frustración en el futuro cuando implementemos herramientas.

GPT 3.5 es terrible llamando herramientas; Las horas que he perdido intentando afrontarlo me permiten decirlo. 😝 Lo dejaré así y hablaré más sobre esto más adelante.

Crear un hilo, agregar mensajes y recuperar el último mensaje

Después de crear un agente, necesitaremos iniciar un hilo de conversación.

 class Agent: # ... (rest of code) def create_thread(self): self.thread = self.client.beta.threads.create()

Y querremos una forma de agregar mensajes a ese hilo:

 class Agent: # ... (rest of code) def add_message(self, message): self.client.beta.threads.messages.create( thread_id=self.thread.id, role="user", content=message )

Tenga en cuenta que por el momento solo es posible agregar mensajes con el rol user . Creo que OpenAI planea cambiar esto en una versión futura, ya que es bastante limitante.

Ahora podemos recibir el último mensaje del hilo:

 class Agent: # ... (rest of code) def get_last_message(self): return self.client.beta.threads.messages.list( thread_id=self.thread.id ).data[0].content[0].text.value

A continuación, creamos un método run_agent de punto de entrada para probar lo que tenemos hasta ahora. Actualmente, el método run_agent solo devuelve el último mensaje del hilo. En realidad, no realiza una ejecución. Todavía es una tontería.

 class Agent: # ... (rest of code) def run_agent(self): message = self.get_last_message() return message

De vuelta en main.py , creamos el agente y nuestro primer hilo. Añadimos un mensaje al hilo. Luego, devuelva el mismo mensaje al usuario, pero esta vez, procedente de ese hilo en vivo.

 from agent import Agent agent = Agent(name="Bilbo Baggins", personality="You are the accomplished and renowned adventurer from The Hobbit. You act like you are a bit of a homebody, but you are always up for an adventure. You worry a bit too much about breakfast.") agent.create_thread() while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the agent...") break agent.add_message(user_input) answer = agent.run_agent() print(f"Assistant: {answer}")

Ejecutémoslo:

 python3 main.py User: hi Assistant: hi

Todavía no soy muy inteligente. Más cerca de un loro 🦜 que de un hobbit. En la siguiente sección comienza la verdadera diversión.

Crear y sondear una ejecución

Cuando crea una ejecución, debe recuperar periódicamente el objeto Run para verificar el estado de la ejecución. Esto se llama encuesta y apesta. Debe realizar una encuesta para determinar qué debe hacer su agente a continuación. OpenAI planea agregar soporte para transmisión para simplificar esto. Mientras tanto, le mostraré cómo configurar el sondeo en la siguiente sección.

Tenga en cuenta el _ en los siguientes nombres de métodos, que es el estándar en Python para indicar que el método está diseñado para uso interno y no se debe acceder a él directamente mediante código externo.

Primero, creemos un método auxiliar _create_run para crear un Run y actualicemos run_agent para llamar a este método:

 class Agent: # ... (rest of code) def get_breakfast_count_from_db(self): return 1 def _create_run(self): count = self.get_breakfast_count_from_db() return self.client.beta.threads.runs.create( thread_id=self.thread.id, assistant_id=self.assistant.id, instructions=f""" Your name is: {self.name} Your personality is: {self.personality} Metadata related to this conversation: {{ "breakfast_count": {count} }} """, ) def run_agent(self): run = self._create_run() # add this line message = self.get_last_message() return message

Observe cómo pasamos thread.id y assistant.id para crear una ejecución.

¿Recuerda que dije al principio que había un lugar mejor para pasar instrucciones y datos dinámicos? Ese sería el parámetro instructions al crear el Run. En nuestro caso, podríamos obtener el count del desayuno de una base de datos. Esto le permitirá pasar fácilmente diferentes datos dinámicos relevantes cada vez que desee generar una respuesta.

Ahora, su agente es consciente de que el mundo cambia a su alrededor y puede actuar en consecuencia. Me gusta tener un objeto JSON de metadatos en mis instrucciones que mantenga un contexto dinámico relevante. Esto me permite pasar datos siendo menos detallado y en un formato que el LLM entiende muy bien.

No ejecutes esto todavía; no funcionará porque no estamos esperando a que se complete la ejecución cuando recibimos el último mensaje, por lo que seguirá siendo el último mensaje del usuario.

Resolvamos esto construyendo nuestro mecanismo de encuesta. Primero, necesitaremos una forma de recuperar una ejecución repetida y fácilmente, así que agreguemos un método _retrieve_run :

 class Agent: # ... (rest of code) def _retrieve_run(self, run: Run): return self.client.beta.threads.runs.retrieve( run_id=run.id, thread_id=self.thread.id)

Observe cómo necesitamos pasar run.id y thread.id para encontrar una ejecución específica.

Agregue un método _poll_run a nuestra clase Agente:

 class Agent: # ... (rest of code) def _cancel_run(self, run: Run): self.client.beta.threads.runs.cancel( run_id=run.id, thread_id=self.thread.id) def _poll_run(self, run: Run): status = run.status start_time = time.time() while status != "completed": if status == 'failed': raise Exception(f"Run failed with error: {run.last_error}") if status == 'expired': raise Exception("Run expired.") time.sleep(1) run = self._retrieve_run(run) status = run.status elapsed_time = time.time() - start_time if elapsed_time > 120: # 2 minutes self._cancel_run(run) raise Exception("Run took longer than 2 minutes.")

🥵 Uf, eso es mucho... Analicémoslo.

_poll_run recibe un objeto Run como argumento y extrae el status actual de Run. Todos los estados disponibles se pueden encontrar en los documentos de OpenAI. Solo usaremos algunos que se adapten a nuestro propósito actual.

Ahora ejecutamos un bucle while para verificar el estado completo mientras manejamos algunos escenarios de error. La facturación real de la API del Asistente es un poco confusa, por lo que, para estar seguro, opté por cancelar mis ejecuciones después de 2 minutos.

Aunque hay un estado expired cuando OpenAI cancela ejecuciones después de 10 minutos. Si una carrera dura más de 2 minutos, probablemente tengas un problema de todos modos.

Como tampoco quiero sondear cada pocos milisegundos, limito mi solicitud sondeando solo cada 1 segundo hasta que llego a la marca de 2 minutos y cancelo mi ejecución. Puedes ajustar esto a lo que creas conveniente.

En cada iteración después del retraso, recuperamos el estado Ejecutar nuevamente.

Ahora, conectemos todo eso a nuestro método run_agent . Notará que primero creamos la ejecución con _create_run luego sondeamos con _poll_run hasta que obtenemos una respuesta o se genera un error y, finalmente, cuando finaliza el sondeo, recuperamos el último mensaje del hilo que ahora será del agente.

Luego devolvemos el mensaje a nuestro bucle de tiempo de ejecución, para que pueda ser enviado de vuelta al usuario.

 class Agent: # ... (rest of code) def run_agent(self): run = self._create_run() self._poll_run(run) # add this line message = self.get_last_message() return message

Voilà, ahora, cuando vuelvas a ejecutar tu agente, recibirás una respuesta de nuestro amigable Agente:

 python3 main.py User: hi Assistant: Hello there! What adventure can we embark on today? Or perhaps, before we set out, we should think about breakfast. Have you had yours yet? I've had mine, of course – can't start the day without a proper breakfast, you know. User: how many breakfasts have you had? Assistant: Ah, well, I've had just 1 breakfast today. But the day is still young, and there's always room for a second, isn't there? What about you? How can I assist you on this fine day?

En la parte 2, agregaremos la capacidad de que nuestro Agente llame a herramientas.

Puedes encontrar el código completo en mi GitHub .

Gracias por su lectura. Me alegra escuchar cualquier opinión y comentario en los comentarios. Sígueme en Linkedin para más contenido como este: https://www.linkedin.com/in/jean-marie-dalmasso-1b5473141/