Streaming LLama2 con FastAPI

En un post anterior vimos que es LLama2 y cómo usarlo con Python, si no lo viste aquí esta el link. Esta vez vamos a ver un ejemplo muy sencillo de cómo podemos usar Llama2 haciedno streaming via FastAPI.

Ahora si vamos a iniciar creando el API. Arrancamos importando las librerias que necesitamos y avanzaremos en el codigo por bloques.

Procedemos a crear el modelo base para nuestra pregunta, este constará de dos partes. La primera parte es un campo para indicar el autor de la pregunta y la segunda es un campo con la pregunta como tál, ambos del tipo string.

Ahora vamos a cargar el modelo, para más detalles y un tutorial breve de cómo cargarlo da click aquí.

Continuamos creando el API.

Ya tenemos el modelo cargado y el API creada, vamos a crear una funcion para que el modelo pueda hacer streaming de la respuesta. Para esto utilizamos la misma función del post pasado, la diferencia es el que vamos a quitar el parametro “echo” y en su lugar usamos el “stream”.

Esta función se encarga de generar respuestas a preguntas o indicaciones utilizando el modelo de IA Llama y transmitir esas respuestas de vuelta al cliente. Aquí tienes un análisis detallado de la función:

• Firma de la Función:
  • def answer_question(question: Question) -> str:: Esto define una función llamada answer_question que toma un solo argumento, question, que se espera que sea una instancia del modelo Pydantic Question. Se espera que la función devuelva una cadena de texto.
• Inferencia del Modelo Llama:
  • stream = llm(...): Esta línea invoca al modelo de IA Llama (llm) para generar respuestas a la pregunta proporcionada. Se pasan varios parámetros a la función de inferencia del modelo:
    • f"Question: {question.question} Answer:": Este es el texto de entrada que incluye la pregunta del usuario. Está formateado como “Question: [pregunta del usuario] Answer:”.
    • max_tokens=250: Esto establece un límite máximo en la cantidad de tokens en la respuesta generada, lo que ayuda a controlar la longitud de la respuesta.
    • stop=["/n", "Question:", "Q:"]: Estas son las condiciones de detención para la generación de respuestas. Si alguna de estas cadenas se encuentra en el texto generado, la generación de respuestas se detiene.
    • stream=True: Esta bandera indica que el modelo debe generar respuestas de manera continua, permitiendo el procesamiento en tiempo real.
• Inicio de la Transmisión de Respuestas:
  • print('Answer starts...'): Esta línea imprime un mensaje que indica que ha comenzado el proceso de generación de respuestas. Este mensaje ayuda a monitorear la ejecución del código y sirve como marcador para cuando comienza el proceso de generación.
• Transmisión de Respuestas:
  • for out in stream:: Este bucle itera sobre las respuestas generadas por el modelo Llama, que se proporcionan de manera continua.
  • completionFragment = copy.deepcopy(out): Se crea una copia profunda de cada fragmento de respuesta para asegurarse de que las modificaciones en el fragmento no afecten a los datos originales.
  • yield completionFragment['choices'][0]['text']: Esta línea emite el texto de la respuesta generada como una cadena de texto. Se extrae el campo choices del fragmento de respuesta y se emite el texto de la primera opción. Este texto representa la respuesta del modelo de IA a la pregunta del usuario.

En resumen, la función answer_question utiliza el modelo de IA Llama para generar respuestas a las preguntas de los usuarios. Inicia un proceso de transmisión, procesa las respuestas a medida que están disponibles y emite el texto de cada respuesta. Esta función es adecuada para aplicaciones en tiempo real donde los usuarios pueden recibir respuestas continuas del modelo de IA a medida que hacen preguntas.

Para terminar la parte del servidor vamos a crear el endpoint de FastAPI que maneja las solicitudes HTTP POST en la ruta /streaming/. A continuación, se explica lo que hace esta función:

• Decorador @app.post('/streaming/'):

  • El decorador @app.post se utiliza para definir un punto final para solicitudes HTTP POST.
  • /streaming/ especifica la ruta en la que este punto final será accesible. En este caso, es accesible en http://tu-dirección-de-servidor/streaming/.

• Parámetros de la Función:

  • async def main(question: Question): Esta función se llama main y toma un parámetro, question, que se espera que sea una instancia del modelo Pydantic Question. La palabra clave async indica que esta función puede ser asíncrona y usar await para operaciones asíncronas.

• Generación de la Respuesta:

  • return StreamingResponse(answer_question(question), media_type='text/event-stream'): Esta línea devuelve un objeto StreamingResponse.
  • answer_question(question) llama a la función answer_question, pasándole el parámetro question. Esta función genera respuestas a la pregunta del usuario y las entrega a medida que estén disponibles.
  • media_type='text/event-stream' especifica el tipo de medios de la respuesta, indicando que la respuesta estará en el formato de Eventos Enviados por el Servidor (SSE, por sus siglas en inglés). SSE es una tecnología que permite al servidor enviar actualizaciones al cliente a través de una única conexión HTTP, lo que lo hace adecuado para aplicaciones en tiempo real, como la transmisión de respuestas.

En resumen, este endpoint de FastAPI (main) escucha las solicitudes POST en la ruta /streaming/. Cuando se recibe una solicitud POST, llama a la función answer_question para generar respuestas a la pregunta del usuario de manera continua y las devuelve al cliente como Eventos Enviados por el Servidor (SSE), lo que permite una comunicación en tiempo real entre el cliente y el modelo de IA.

Ahora que tenemos nuestra API creada, vamos a crear el cliente que la consume para realizar las preguntas.

Aquí tienes una explicación de lo que hace este código del cliente:

• Importar la Biblioteca requests:

  • El código comienza importando la biblioteca requests, que se utiliza para realizar solicitudes HTTP al punto final proporcionado.

• Definir la Variable llama2_url:

  • llama2_url almacena la URL del servicio web con el que deseas interactuar. En este caso, se establece en “http://localhost:8000/streaming”. Esta URL apunta al punto final donde se espera que el servidor (presumiblemente en la máquina local) proporcione respuestas en tiempo real.

• Bucle Infinito para la Interacción del Usuario:

  • El código entra en un bucle infinito utilizando while True, lo que permite que el usuario interactúe con el cliente de manera continua.

• Obtener la Entrada del Usuario para la Pregunta:

  • La función input solicita al usuario que escriba una pregunta y almacena la entrada en la variable question.

• Verificar Comando de Salida:

  • El código verifica si el usuario desea salir del bucle. Si el usuario ingresa “salir” (en minúsculas), el bucle se termina con break, y el programa sale de manera ordenada.

• Preparar el Cuerpo de la Solicitud:

  • El código establece la variable author en la cadena “Squaid”.
  • Crea un diccionario llamado body con dos pares clave-valor: 'author' y 'question'. El autor se establece en “Squaid” y la pregunta se establece en la entrada del usuario.

• Realizar la Solicitud POST:

  • El código realiza una solicitud HTTP POST a llama2_url utilizando el método requests.post.
  • El parámetro json se establece en el diccionario body, lo que significa que el cliente envía un payload JSON con el autor y la pregunta al servidor.
  • stream=True indica que la respuesta del servidor se transmitirá y procesará de manera incremental.

• Procesar e Imprimir la Respuesta en Tiempo Real:

  • Si la solicitud es exitosa (código de estado 200), el código entra en un bloque with para manejar la respuesta (r) del servidor.
  • Itera sobre el contenido de la respuesta en fragmentos de 1024 bytes utilizando r.iter_content(1024).
  • Para cada fragmento, decodifica los bytes como UTF-8 y muestra el resultado. Esto permite que el cliente reciba y muestre las respuestas en tiempo real del servidor a medida que llegan.

• Manejar Errores de la Solicitud:

  • Si ocurre una excepción del tipo requests.exceptions.RequestException durante la solicitud (por ejemplo, si el servidor no está disponible), el código captura la excepción e imprime un mensaje de error.

• Mensaje de Salida:

  • Finalmente, el código imprime un mensaje que indica que el programa está saliendo.

Este código del cliente interactúa de manera efectiva con el servidor en la URL especificada, permitiendo que el usuario envíe preguntas y reciba respuestas en tiempo real desde el servidor hasta que decida salir. Es adecuado para probar e interactuar con el modelo de IA que proporciona respuestas en tiempo real.

Ya tenemos todo listo, asi que procedemos a ejecutar tanto el servidor con el API como el cliente. Para esto vamos usar dos terminales diferentes y ejecutaremos los siguientes comandos uno en cada terminal. Es importante esperar que el modelo cargue antes de realizar preguntas desde el cliente.

Una vez tu modelo cargue te aparecerá un mensaje similar a esto:

Y si tu streaming funciona bien vas a ver algo así:

Hemos aprendido como crear fácil y rápido tu propio ChatGPT con el ultimo modelo de Llama 2, ahora haciendo streaming via FastAPI

El código completo lo encuentra aquí. Suscríbete al canal de whatsapp en nuestro home o sigueme directamente en instagram @angeloftechml para estar al tanto de nuevos tutoriales o recursos.