Cree un chatbot perimetral acelerado por NPU para PC con tecnología Snapdragon

Este tutorial te proporciona todo lo necesario para crear tu propia app de chat en una PC con Snapdragon y AnythingLLM. Describimos el proceso de configuración y pruebas, además de algunas recomendaciones para ampliar las capacidades básicas de la app.

Incluso si la IA no es su área de enfoque habitual, vale la pena aprender a trabajar y desarrollar con modelos de IA. Los modelos básicos de IA siguen creciendo en potencia y capacidad analítica, y el mercado de la IA se está expandiendo con nuevos modelos específicos para cada tarea.

Conocer los fundamentos de cómo configurar, probar y modificar una aplicación impulsada por IA es una forma de asegurar el futuro de tus habilidades. Este ejercicio es perfecto para quienes se inician en el desarrollo con IA o buscan un proyecto pequeño y divertido que les permita experimentar y experimentar con una configuración sencilla de IA de borde.

Lo que necesitarás

• Hardware:
  
  Esta demostración se creó con el siguiente hardware. La aplicación está diseñada para ser independiente del hardware, pero podrías notar diferencias de rendimiento según el hardware que elijas. Asegúrate de tener suficiente RAM para la inferencia local. AnythingLLM es muy ligero y permite usar funciones básicas y almacenar chats con tan solo 2 GB de RAM . 
  
  
  • Máquina: Dell Latitude 7455
  • Chip: Snapdragon X Elite
  • Sistema operativo: Windows 11
  • Memoria: 32 GB
     
• Software:
  
• Versión de Python: 3.12.6
  
• Proveedor LLM: AnythingLLM NPU (para modelos más antiguos, puede aparecer como Qualcomm QNN)
  
  Modelo de chat de AnythingLLM: Llama 3.1 8B Chat 8K
  
• Otros recursos:
  
  consulte este repositorio de GitHub  para obtener recursos y códigos adicionales.

Configuración

• Instale y configure AnythingLLM. Asegúrese de seleccionar AnythingLLM NPU cuando se le solicite que elija un proveedor LLM para la NPU. Elija un modelo; usamos Llama 3.1 8B Chat con contexto 8K, pero podría obtener un mejor rendimiento con otros modelos según las limitaciones de su hardware.
  
• Crea un espacio de trabajo haciendo clic en «+ Nuevo espacio de trabajo»
  
• Generar una clave API

1. Haga clic en el botón de configuración en la parte inferior del panel izquierdo.
2. Abra el menú desplegable «Herramientas»
3. Haga clic en «API para desarrolladores»
4. Haga clic en «Generar nueva clave API».

Abra una instancia de PowerShell y clone el repositorio

git clone https://github.com/thatrandomfrenchdude/simple-npu-chatbot.git

Crea y activa tu entorno virtual con requisitos

# 1. navigate to the cloned directory

cd simple-npu-chatbot



# 2. create the python virtual environment

python -m venv llm-venv



# 3. activate the virtual environment

./llm-venv/Scripts/Activate.ps1     # windows

source \llm-venv\bin\activate       # mac/linux



# 4. install the requirements

pip install -r requirements.txt

Crea tu archivo config.yaml con las siguientes variables

api_key: "your-key-here"

model_server_base_url: "http://localhost:3001/api/v1"

workspace_slug: "your-slug-here"

stream: true

stream_timeout: 60

Pruebe la autenticación del servidor modelo para verificar la clave API

python src/auth.py

Obtenga el slug de su espacio de trabajo usando la herramienta de espacios de trabajo

1. Ejecute Python src/workspaces.pyen su consola de línea de comandos
2. Encuentra tu espacio de trabajo y su slug a partir de la salida
3. Agregue el slug a la variable workspace_slug en config.yaml

Construyendo la aplicación

Una vez configurada la aplicación, conviene revisar el código a fondo para estar bien preparado para extenderlo a tu propia aplicación. Como siempre ocurre con el código, hay muchas maneras de hacer lo mismo, así que no lo consideres la única forma de crear una aplicación de chatbot.

Además de las auth.pyutilidades workspaces.pymencionadas, este código permite usar la terminal o una interfaz de Gradio para comunicarse con el Chatbot. Una terminal es más rápida de configurar y muy ligera, especialmente útil si se experimenta con limitaciones de dispositivo, pero la interfaz de usuario es limitada. Si no has usado la terminal antes, podría resultarte un poco contradictoria.

Si eliges Gradio, necesitarás acceso a un navegador web; esto significa que usarás algunos más recursos del sistema que con la Terminal, pero disfrutarás de una interfaz de usuario más intuitiva.

Además, ambas interfaces incluyen una versión de bloqueo y transmisión definida por el streamingbooleano en config.yaml. El bloqueo espera hasta que la respuesta se procese completamente para devolverla, y la transmisión devuelve fragmentos a medida que están disponibles.

Esta revisión de código solo cubrirá la interfaz de terminal, ya que su funcionalidad es similar. Tenga en cuenta que el código se ha simplificado en comparación con el repositorio de GitHub para mayor brevedad; se han eliminado los tipos y los comentarios.

Chatbot de terminal

Para comenzar con la versión de terminal, necesitarás las siguientes bibliotecas instaladas e importadas:

import asyncio

import httpx

import json

import requests

import sys

import threading

import time

import yaml

Asyncio, httpx, y requestsse utilizan para manejar las solicitudes de transmisión asincrónica al servidor de modelos, jsony yamlse utilizan para interactuar con las respuestas de la solicitud y el archivo de configuración respectivamente, y sys, threading, y timese utilizan para la barra de progreso mientras se procesa la respuesta de bloqueo.

La loading_indicatorfunción es sencilla: cada medio segundo imprime un punto en la línea de comandos, hasta 10, antes de borrar la línea y comenzar de nuevo. Esto se ejecuta en un hilo mientras se procesa la respuesta del modelo, y el código se ve así:

def loading_indicator():

    while not stop_loading:

        for _ in range(10):

            sys.stdout.write('.')

            sys.stdout.flush()

            time.sleep(0.5)

        sys.stdout.write('\r' + ' ' * 10 + '\r')

        sys.stdout.flush()

    print('')

El resto del código, con excepción de la invocación que se describe al final, existe en la Chatbotclase. La inicialización es sencilla: se lee el archivo de configuración y se asignan variables de clase para las referencias en las demás funciones:

class Chatbot:

    def __init__(self):

        with open("config.yaml", "r") as file:

            config = yaml.safe_load(file)



        self.api_key = config["api_key"]

        self.base_url = config["model_server_base_url"]

        self.stream = config["stream"]

        self.stream_timeout = config["stream_timeout"]

        self.workspace_slug = config["workspace_slug"]



        if self.stream:

            self.chat_url = f"{self.base_url}/workspace/{self.workspace_slug}/stream-chat"

        else:

            self.chat_url = f"{self.base_url}/workspace/{self.workspace_slug}/chat"



        self.headers = {

            "accept": "application/json",

            "Content-Type": "application/json",

            "Authorization": "Bearer " + self.api_key

        }

Preste especial atención a las últimas líneas de código. Se verifica si el usuario desea transmitir, y los encabezados de solicitud se definen una sola vez para facilitar su reutilización.

Después de la inicialización viene la función de ejecución, que puede considerarse la función principal de todos los chatbots:

def run(self):

        while True:

            user_message = input("You: ")

            if user_message.lower() in [

                "exit",

                "quit",

                "bye",

            ]:

                break

            print("")

            try:

                self.streaming_chat(user_message) if self.stream \

                    else self.blocking_chat(user_message)

            except Exception as e:

                print("Error! Check the model is correctly loaded. More details in README troubleshooting section.")

                sys.exit(f"Error details: {e}")

En cada bucle, la función solicita primero la entrada del usuario. Se verifica la entrada para determinar si el usuario desea salir del chat mediante palabras clave especiales asignadas para este fin. Finalmente, se proporciona una respuesta. Repetir hasta el infinito.

Se puede incluir funcionalidad personalizada y codificada mediante la adición de palabras clave adicionales o la inserción de código adicional entre la entrada del usuario y la respuesta del chatbot. Los agentes, una implementación especializada de chatbots capaces de realizar acciones independientes, pueden incluir un procesamiento exhaustivo antes de responder al usuario. Para una implementación sencilla de un agente, consulta este repositorio de GitHub .

Ahora, analicemos la funcionalidad de bloqueo del chat.

Las primeras cuatro líneas de código gestionan el hilo de la barra de carga. stop_loadingSe hace referencia a la variable global mencionada anteriormente, establecida en Falso, ya que queremos que comience la carga. A continuación, se lanza un hilo con la función. Se ve así:

        global stop_loading

        stop_loading = False

        loading_thread = threading.Thread(target=loading_indicator)

        loading_thread.start()

A continuación, se definen datay chat_response. Los datos contienen un diccionario JSON que se pasa a la API cuando se realiza la solicitud. sessionIdAnythingLLM no utiliza la clave del diccionario; su función es registrar las sesiones si el desarrollador desea hacerlo. La clave de archivos adjuntos se incluye porque AnythingLLM podría aceptar y analizar archivos adjuntos con una solicitud de finalización de chat. Chat_responseContiene los resultados de la solicitud realizada mediante la biblioteca estándar requestsde Python.

        data = {

            "message": message,

            "mode": "chat",

            "sessionId": "example-session-id",

            "attachments": []

        }



        chat_response = requests.post(

            self.chat_url,

            headers=self.headers,

            json=data

        )

La última sección de la función gestiona la respuesta del agente. Primero, stop_loadingse activa la variable y el hilo de la barra de carga se reincorpora al hilo principal. A continuación, la función intenta imprimir el mensaje mientras busca errores. Al igual que con la función de ejecución, esta función se puede ampliar con funciones personalizadas. Por ejemplo, aunque AnythingLLM ya registra el historial de chat para obtener contexto, es posible que desees registrarlo por separado dentro de tu aplicación para obtener contexto adicional con cada solicitud.

        stop_loading = True

        loading_thread.join()



        try:

            print("Agent: ", end="")

            print(chat_response.json()['textResponse'])

            print("")

        except ValueError:

            return "Response is not valid JSON"

        except Exception as e:

            return f"Chat request failed. Error: {e}"

En conjunto, la blocking_chatfunción se ve así y toma el mensaje que se enviará como argumento:

def blocking_chat(self, message):

        global stop_loading

        stop_loading = False

        loading_thread = threading.Thread(target=loading_indicator)

        loading_thread.start()



        data = {

            "message": message,

            "mode": "chat",

            "sessionId": "example-session-id",

            "attachments": []

        }



        chat_response = requests.post(

            self.chat_url,

            headers=self.headers,

            json=data

        )



        stop_loading = True

        loading_thread.join()



        try:

            print("Agent: ", end="")

            print(chat_response.json()['textResponse'])

            print("")

        except ValueError:

            return "Response is not valid JSON"

        except Exception as e:

            return f"Chat request failed. Error: {e}"

Ahora veamos cómo se compara el streaming. Consta de dos funciones: una función contenedora asíncrona streaming_chaty otra llamada streaming_chat_asyncpara gestionar los fragmentos conforme llegan. La función contenedora es simple y usa asyncio para procesar el mensaje. Consta de una sola línea, además de la línea de definición, y su aspecto es el siguiente:

def streaming_chat(self, message):

        asyncio.run(self.streaming_chat_async(message))

La función principal, llamada por el contenedor, comienza con un par de definiciones de variables. La variable de datos permanece igual. El búfer inferior se utiliza para recopilar los fragmentos conforme llegan.

data = {

            "message": message,

            "mode": "chat",

            "sessionId": "example-session-id",

            "attachments": []

        }



        buffer = ""

La variable de datos permanece igual. El búfer inferior se utiliza para recopilar los fragmentos conforme llegan. La siguiente sección es extensa, por lo que se divide en fragmentos más pequeños, comenzando con algunos contextos y bucles anidados:

        try:

            async with httpx.AsyncClient(timeout=self.stream_timeout) as client:

                async with client.stream("POST", self.chat_url, headers=self.headers, json=data) as response:

                    print("Agent: ", end="")

                    async for chunk in response.aiter_text():

                        if chunk:

                            buffer += chunk

El primer contexto configura un cliente HTTP asíncrono para gestionar la transmisión, seguido de un segundo contexto que utiliza dicho cliente para realizar la POSTsolicitud con el mensaje en el endpoint del modelo. Una comprobación de errores, no incluida aquí, detecta cualquier error con el cliente HTTPX. La declaración «prints» posterior a estos contextos imprime la etiqueta inicial a la que se añadirá la respuesta del chatbot.

A continuación, se muestran un par de líneas que gestionan fragmentos de la respuesta. Al igual que la llamada a la solicitud, estos fragmentos se iteran asincrónicamente para liberar la ejecución a otros subprocesos cuando un fragmento no está listo para ser procesado. Una vez recibidos, los fragmentos se almacenan en el búfer.

Las siguientes líneas se utilizan para comenzar a procesar los datos en el búfer:

                            while "\n" in buffer:

                                line, buffer = buffer.split("\n", 1)

                                if line.startswith("data: "):

                                    line = line[len("data: "):]

Mientras haya líneas completas en el búfer, se inicia el bucle de procesamiento del búfer para imprimir el fragmento en pantalla. Si no se detecta una línea completa (en este caso, utilizando el carácter de nueva línea), el bucle finaliza y espera más fragmentos.

Para procesar una línea, se extrae del búfer y este se actualiza para eliminar la línea que se está procesando. A partir de ahí, se limpia la línea. Esto es necesario porque AnythingLLM devuelve un diccionario JSON, por lo que se elimina la clave.

A continuación se procesa la línea extraída:

                                try:

                                    parsed_chunk = json.loads(line.strip())

                                    print(parsed_chunk.get("textResponse", ""), end="", flush=True)



                                    if parsed_chunk.get("close", False):

                                        print("")

                                except json.JSONDecodeError:

                                    # The line is not a complete JSON; wait for more data.

                                    continue

                                except Exception as e:

                                    # generic error handling, quit for debug

                                    print(f"Error processing chunk: {e}")

                                    sys.exit()

Dentro del bloque try, el fragmento analizado se carga como json y se imprime en la consola. La última sección comprueba si este es el mensaje final; de ser así, se imprime una nueva línea para separar la entrada del usuario de la salida del chatbot. Un procesamiento de errores ligero (uno para la decodificación de json y otro genérico) gestiona cualquier problema inesperado.

En conjunto, las streaming_chatfunciones se ven así y toman el mensaje que se enviará como argumento:

def streaming_chat(self, message):

        asyncio.run(self.streaming_chat_async(message))



async def streaming_chat_async(self, message):

        data = {

            "message": message,

            "mode": "chat",

            "sessionId": "example-session-id",

            "attachments": []

        }



        buffer = ""

        try:

            async with httpx.AsyncClient(timeout=self.stream_timeout) as client:

                async with client.stream("POST", self.chat_url, headers=self.headers, json=data) as response:

                    print("Agent: ", end="")

                    async for chunk in response.aiter_text():

                        if chunk:

                            buffer += chunk

                            while "\n" in buffer:

                                line, buffer = buffer.split("\n", 1)

                                if line.startswith("data: "):

                                    line = line[len("data: "):]

                                try:

                                    parsed_chunk = json.loads(line.strip())

                                    print(parsed_chunk.get("textResponse", ""), end="", flush=True)



                                    if parsed_chunk.get("close", False):

                                        print("")

                                except json.JSONDecodeError:

                                    # The line is not a complete JSON; wait for more data.

                                    continue

                                except Exception as e:

                                    # generic error handling, quit for debug

                                    print(f"Error processing chunk: {e}")

                                    sys.exit()

        except httpx.RequestError as e:

            print(f"Streaming chat request failed. Error: {e}")

Finalmente, se instancia el chatbot y se llama con una función principal:

if __name__ == '__main__':

    stop_loading = False

    chatbot = Chatbot()

    chatbot.run()

La primera línea de este bloque indica al intérprete de Python que ejecute este código si el archivo se llama como archivo principal. La segunda línea establece una variable global que indica al hilo indicador de carga si debe mostrar la barra de carga. Las dos últimas inicializan la Chatbotclase y la ejecutan.

Siga  el video De cero a chatbot: desafío de creación de 30 minutos. Siga el video aquí  para obtener instrucciones más completas, paso a paso.

Prueba tu aplicación de chat

Como se describió en la sección anterior, puede usar una terminal o la interfaz de chat de Gradio para comunicarse con el bot. Tras completar la configuración, ejecute la aplicación que elija desde la línea de comandos:

# terminal

python src/terminal_chatbot.py



# gradio

python src/gradio_chatbot.py

Solución de problemas comunes

Falta el tiempo de ejecución de AnythingLLM NPU

En un equipo con Snapdragon X Elite, AnythingLLM NPU debería ser el proveedor LLM predeterminado. Si no lo ve en el menú desplegable, significa que descargó la versión x64 de AnythingLLM. Desinstale la aplicación e instale la versión ARM64.

Modelo no descargado

A veces, el modelo seleccionado no se descarga, lo que provoca un error en la generación. Para solucionarlo, revise el modelo en Configuración -> Proveedores de IA -> LLM en AnythingLLM. Debería ver «desinstalar» en la tarjeta del modelo si está instalado correctamente. Si ve «el modelo requiere descarga», elija otro modelo, haga clic en guardar, vuelva a cambiar y luego guarde. Debería ver la descarga del modelo en la esquina superior derecha de la ventana de AnythingLLM.

Complemento opcional: Crear una barra de carga

Para implementar una barra de carga multiproceso para tareas de inferencia en AnythingLLM, necesitará utilizar los módulos de subprocesos y cola de Python.

1. Componentes principales

• Hilo principal: maneja las actualizaciones de la interfaz de usuario y la representación de la barra de progreso
• Hilo de trabajo: ejecuta la tarea de inferencia
• Cola de progreso: canal de comunicación seguro entre subprocesos

import threading

import queue

import time

from alive_progress import alive_bar  # Optional for advanced animations

2. Seguimiento del progreso seguro para subprocesos

def inference_task(progress_queue, query):

    # Simulate inference processing

    steps = ["Tokenizing", "Processing", "Generating"]

    for i, step in enumerate(steps, 1):

        time.sleep(1)  # Replace with actual inference work

        progress_queue.put((i/len(steps)*100, step))

    progress_queue.put((100, "Complete"))

3. Rosca de la barra de carga

def loading_bar(progress_queue):

    with alive_bar(100, title='Processing Query') as bar:

        while True:

            try:

                progress, status = progress_queue.get(timeout=0.1)

                bar.title(f'[{status}]')

                bar(progress - bar.current)  # Increment by delta

                if progress >= 100:

                    break

            except queue.Empty:

                continue

4. Integración de AnythingLLM

def run_inference_with_progress(query):

    progress_queue = queue.Queue()

    

    # Start inference thread

    inference_thread = threading.Thread(

        target=inference_task,

        args=(progress_queue, query)

    )

    inference_thread.start()

    

    # Start progress bar in main thread

    loading_bar(progress_queue)

    

    inference_thread.join()

    print("\nInference complete")

Notas de implementación

Sincronización de subprocesos

• Utilice queue.Queue para una comunicación segura entre subprocesos
• Las encuestas del hilo principal se ponen en cola cada 100 ms para actualizaciones
• El trabajador envía mensajes de porcentaje completado y estado
  
   

Integración de UI

# For web UI integration (Flask example)

@app.route('/chat', methods=['POST'])

def chat():

    query = request.json['query']

    thread = threading.Thread(target=run_inference_with_progress, args=(query,))

    thread.start()

    return jsonify({"status": "Processing started"})

Funciones avanzadas

• Úselo alive-progresspara hilanderos animados y estadísticas de rendimiento.
• Añadir cálculos de ETA: 

bar.title(f'ETA: {bar.eta}s | {status}')

Diagrama de arquitectura

[User Input]

    │

    ▼

[Main Thread] ─── starts ───▶ [Worker Thread]

    │  ▲                       │

    │  └── progress updates ───┘

    ▼

[Loading Bar Render]

    │

    ▼

[LLM Response]

Este patrón mantiene la capacidad de respuesta de la interfaz de usuario mientras muestra el progreso de la inferencia en tiempo real.

¿Buscas otra extensión de este proyecto básico? ¡Prueba a usar el flujo de texto para la comunicación asincrónica!

Recursos adicionales

¿Eres nuevo en la programación? ¡Bienvenido! Aprender a programar te ayudará a adquirir habilidades útiles que se pueden aplicar a diversos ámbitos.

Programar te enseña a descomponer problemas en tareas más sencillas y a identificar elementos recurrentes en dichas tareas; a menudo, se trata de tareas que se pueden automatizar o, como mínimo, puedes reutilizar el código que escribes para ejecutarlas en lugar de empezar desde cero cada vez.

Hay muchos cursos de programación excelentes para principiantes que podrían resultarte útiles:

• LearnPython.org tiene tutoriales gratuitos, incluidas opciones para principiantes y avanzados.
• Si aprendes mejor cuando tienes un problema específico que resolver, podrías disfrutar de las ofertas de Udemy basadas en proyectos. No son gratuitas, pero Udemy tiene ofertas frecuentes y a menudo puedes encontrar cupones.
• Si eres un principiante absoluto, el tutorial gratuito de Python de Free Code Camp te permitirá familiarizarte con los conceptos básicos en aproximadamente 4,5 horas.

¡Nos encantaría ver lo que creaste! ¡Únete a nuestro Discord de desarrolladores de Qualcomm para mostrar tu proyecto! O bien, visita Qualcomm AI Hub para obtener más información sobre el uso de modelos y herramientas de IA en dispositivos con tecnología Qualcomm.

Preguntas frecuentes

P: ¿Qué es Edge AI y por qué usar Snapdragon NPU?

R: Edge AI se refiere a la ejecución de modelos de aprendizaje automático directamente en dispositivos locales (como hardware con tecnología Snapdragon), lo que reduce la latencia, minimiza la transferencia de datos y mejora la privacidad.

La NPU de Snapdragon acelera la inferencia en el dispositivo, lo que permite que las aplicaciones de chat e IA funcionen más rápido y con mayor capacidad de respuesta sin depender de la nube.

P: ¿Qué es AnythingLLM y cómo se integra con Snapdragon?

R: AnythingLLM es un framework ligero e independiente del proveedor para ejecutar modelos de lenguaje extensos. En esta guía, elegimos «Qualcomm QNN» como referencia para la aceleración de NPU. Esto permite la inferencia local con modelos como Llama 3.1 8B utilizando el hardware acelerador de Snapdragon.

P: ¿Cómo puedo mejorar la respuesta de la interfaz de usuario durante la inferencia?

R: Implemente la inferencia asíncrona con un hilo de trabajo y comunique el progreso mediante un comando queue.Queueal hilo principal. Con herramientas como alive-progress[nombre del hilo], puede renderizar una barra de carga (o la interfaz de usuario de Gradio) mientras la NPU procesa los datos en segundo plano, manteniendo la interfaz fluida.

Qualcomm Blog. N. D. Traducido al español