RAG sin OpenAI: BentoML, OctoAI y Milvus
Este artículo se publicó originalmente en The New Stack y se vuelve a publicar aquí con permiso.
Las opciones ampliadas de generación aumentada por recuperación pueden eliminar la dependencia de los desarrolladores respecto de OpenAI
ChatGPT llevó la IA al primer plano del conocimiento público en 2023. Sin embargo, ahora hay muchas más opciones, por lo que ya no estamos atados a OpenAI. Esta es la tercera entrada de una serie de blogs sobre cómo puedes crear aplicaciones de generación aumentada por recuperación (RAG) utilizando LLM que no son GPT de OpenAI. Aquí puedes encontrar la parte 1 y la parte 2. El repositorio de GitHub para este proyecto se puede encontrar aquí.
En este tutorial usaremos BentoML para servir embeddings, OctoAI para obtener el LLM y Milvus como nuestra base de datos vectorial. Cubrimos:
Servir embeddings con BentoML
Insertar tus datos en una base de datos vectorial para RAG
Crear tu colección de Milvus
Analizar e incrustar tus datos para la inserción
Configurar tu LLM para RAG
Dar instrucciones al LLM
Un ejemplo de RAG
Resumen de BOM dot COM: BentoML, OctoAI y Milvus para RAG
Servir Embeddings con BentoML
Podemos usar embeddings de oraciones servidos a través de BentoML utilizando su repositorio Sentence Transformers Embeddings. Cubramos brevemente qué está ocurriendo con este repositorio. Lo principal que hay que mirar es el archivo service.py. Básicamente, pone en marcha un servidor y le añade un endpoint de API. Dentro del endpoint de API, está cargando all-MiniLM-L6-v2 desde Hugging Face y usándolo para crear embeddings.
Ese repositorio pone en marcha un servidor y nos da un endpoint al que acceder en <http://localhost:3000>. Para usar este endpoint, importamos bentoml y ponemos en marcha un cliente HTTP usando el tipo de objeto nativo SyncHTTPClient.
import bentoml
bento_client = bentoml.SyncHTTPClient("http://localhost:3000")
Una vez que nos conectamos al cliente, creamos una función que obtiene una lista de embeddings a partir de una lista de cadenas. Algo a tener en cuenta es que divido las listas en 25 cadenas a la vez. Esto se debe principalmente a que estamos usando un endpoint síncrono. Dividir la lista de cadenas hace que las llamadas sean menos costosas computacionalmente y evita tiempos de espera agotados.
Después de dividir la lista en secciones de 25, llamamos al bento_client que creamos arriba para codificar estas oraciones. El cliente de BentoML devuelve una lista de vectores, efectivamente una lista de listas. Tomamos cada uno de estos vectores y los añadimos a nuestra lista vacía de embeddings. Al final de este bucle devolvemos la lista final de embeddings.
Si no hay más de 25 cadenas en la lista de textos, simplemente llamamos al método encode del cliente sobre la lista de cadenas pasada.
def get_embeddings(texts: list) -> list:
if len(texts) > 25:
splits = [texts[x:x+25] for x in range(0, len(texts), 25)]
embeddings = []
for split in splits:
embedding_split = bento_client.encode(
sentences = split
)
for embedding in embedding_split:
embeddings.append(embedding)
return embeddings
return bento_client.encode(
sentences=texts,
)
Insertar Tus Datos en una Base de Datos Vectorial para RAG
Con nuestra función de embedding preparada, podemos preparar nuestros datos para insertarlos en Milvus para nuestra aplicación RAG. El primer paso en esta sección es iniciar y conectarse a Milvus. Hay un archivo docker-compose.yml ubicado en el repositorio enlazado arriba. También puedes encontrar el Milvus Docker Compose en esta página de documentación.
Si tienes Docker instalado y descargaste ese repo, deberías poder ejecutar docker compose up -d para iniciar Milvus. Una vez que tu servidor Milvus esté activo, es hora de conectarse a él. Para esta parte, simplemente importamos el módulo connections y llamamos a connect con el host (localhost o 127.0.0.1) y el puerto (19530). El bloque de código de abajo también define dos constantes: un nombre de colección y la dimensión. Puedes inventar el nombre de colección que quieras. El tamaño de la dimensión proviene del tamaño del modelo de embedding, all-MiniLM-L6-v2.
from pymilvus import connections
COLLECTION_NAME = "bmo_test"
DIMENSION = 384
connections.connect(host="localhost", port=19530)
Crear tu colección de Milvus
Crear una colección en Milvus implica dos pasos: primero, definir el esquema y, segundo, definir el índice. Para esta sección, necesitamos cuatro módulos: FieldSchema define un campo, CollectionSchema define una colección, DataType nos dice qué tipo de datos habrá en un campo y Collection es el objeto que Milvus usa para crear colecciones.
Podemos definir aquí todo el esquema de la colección. O bien, podemos simplemente definir las dos piezas necesarias: id y embedding. Luego, cuando llega el momento de definir el esquema, pasamos un parámetro, enabled_dynamic_field, que nos permite insertar los campos que queramos siempre que también tengamos los campos id y embedding. Esto nos permite tratar la inserción de datos en Milvus de la misma manera en que trataríamos una base de datos NoSQL como MongoDB. A continuación, simplemente creamos la colección con el nombre y el esquema proporcionados anteriormente.
from pymilvus import FieldSchema, CollectionSchema, DataType, Collection
# id and embedding are required to define
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=DIMENSION)
]
# "enable_dynamic_field" lets us insert data with any metadata fields
schema = CollectionSchema(fields=fields, enable_dynamic_field=True)
# define the collection name and pass the schema
collection = Collection(name=COLLECTION_NAME, schema=schema)
Ahora que hemos creado nuestra colección, necesitamos definir el índice. En términos de búsqueda, un “índice” define cómo vamos a mapear nuestros datos para la recuperación. Usamos HSNW (hierarchical navigable small worlds) para indexar nuestros datos para este proyecto. También necesitamos definir cómo vamos a medir la distancia vectorial. En este ejemplo, usamos el producto interno o IP.
Cada uno de los 11 tipos de índice ofrecidos en Milvus tiene un conjunto diferente de parámetros. Para HNSW, tenemos dos parámetros para ajustar: “M” y “efConstruction”. “M” es el límite superior para el grado de un nodo en cada grafo, y “efConstruction” es el factor exploratorio utilizado durante la construcción del índice.
Desde un punto de vista práctico, valores más altos de “M” y “efConstruction” conducen a una mejor búsqueda. Un valor más alto de “M” significa que el índice ocupará más memoria. Un valor más alto de “efConstruction” significa que llevará más tiempo construir el índice. Tendrás que experimentar con estos valores para encontrar los mejores.
Una vez definido el índice, creamos el índice en un campo elegido, en este caso, embedding. Luego llamamos a load para cargar la colección en memoria.
index_params = {
"index_type": "HNSW", # one of 11 Milvus indexes
"metric_type": "IP", # L2, Cosine, or IP
"params": {
"M": 8, # higher M = consumes more memory but better search quality
"efConstruction": 64 # higher efConstruction = slower build, better search
},
}
# pass the field to index on and the parameters to index with
collection.create_index(field_name="embedding", index_params=index_params)
# load the collection into memory
collection.load()
Análisis e incrustación de tus datos para la inserción
Con Milvus listo y la conexión realizada, podemos insertar datos en nuestra base de datos vectorial. Pero primero tenemos que preparar los datos para insertarlos. Para este ejemplo, tenemos un montón de archivos txt, disponibles en la carpeta de datos del repositorio. Dividimos estos datos en fragmentos, los incrustamos y los almacenamos en Milvus.
Empecemos creando una función que divida este texto en fragmentos. Hay muchas formas de hacer chunking, pero lo haremos de forma ingenua para este ejemplo. La función de abajo toma un archivo, lo lee como una cadena y luego lo divide en cada nueva línea. Devuelve la lista de cadenas recién creada.
# naively chunk on newlines
def chunk_text(filename: str) -> list:
with open(filename, "r") as f:
text = f.read()
sentences = text.split("n")
return sentences
A continuación, procesamos cada uno de los archivos que tenemos. Obtenemos una lista de todos los nombres de archivo y creamos una lista vacía para contener la información fragmentada. Luego, recorremos todos los archivos y ejecutamos la función anterior en cada uno para obtener una fragmentación ingenua de cada archivo. Antes de almacenar los fragmentos, debemos limpiarlos.
Si observas cómo se fragmenta un archivo individual, verás muchas líneas vacías, y no queremos líneas vacías. Algunas líneas son solo tabulaciones u otros caracteres especiales. Para evitarlas, creamos una lista vacía y almacenamos solo los fragmentos que superan cierta longitud. Para simplificar, podemos usar siete caracteres.
Una vez que tenemos una lista limpia de fragmentos de cada documento, podemos almacenar nuestros datos. Creamos un diccionario que mapea cada lista de fragmentos al nombre del documento, en este caso, el nombre de la ciudad. Luego, agregamos todos estos a la lista vacía que creamos antes.
import os
cities = os.listdir("data")
# store chunked text for each of the cities in a list of dicts
city_chunks = []
for city in cities:
chunked = chunk_text(f"data/{city}")
cleaned = []
for chunk in chunked:
if len(chunk) > 7:
cleaned.append(chunk)
mapped = {
"city_name": city.split(".")[0],
"chunks": cleaned
}
city_chunks.append(mapped)
Con un conjunto de textos fragmentados para cada ciudad listo, es hora de obtener algunas incrustaciones. Milvus puede tomar una lista de diccionarios para insertarla en una colección, así que podemos empezar con otra lista vacía. Para cada uno de los diccionarios que creamos arriba, necesitamos obtener una lista de incrustaciones que coincida con la lista de oraciones.
Hacemos esto llamando directamente a la función get_embeddings que creamos en la sección usando BentoML en cada uno de la lista de fragmentos. Ahora, necesitamos emparejarlos. Dado que los embeddings de la lista y la lista de oraciones deberían coincidir por índice, podemos usar enumerate en cualquiera de las dos listas para emparejarlos.
Los emparejamos creando un diccionario que representa una sola entrada en Milvus. Cada entrada incluye el embedding, la oración relacionada y la ciudad. Es opcional incluir la ciudad, pero incluyámosla solo para que podamos usarla. Observa que no hay necesidad de incluir un id en esta entrada. Eso se debe a que elegimos autoincrementar el id cuando creamos el esquema anterior.
Añadimos cada una de estas entradas a la lista mientras las recorremos en el bucle. Al final, tenemos una lista de diccionarios en la que cada diccionario representa una entrada de una sola fila para Milvus. Luego podemos simplemente insertar estas entradas en nuestra colección de Milvus. El último paso aquí es hacer flush de las entradas para que podamos comenzar a indexarlas.
entries = []
for city_dict in city_chunks:
embedding_list = get_embeddings(city_dict["chunks"]) # returns a list of lists
# now match texts with embeddings and city name
for i, embedding in enumerate(embedding_list):
entry = {"embedding": embedding,
"sentence": city_dict["chunks"][i], # poorly named cuz it's really a bunch of sentences, but meh
"city": city_dict["city_name"]}
entries.append(entry)
collection.insert(entries)
collection.flush()
Configura tu LLM para RAG
Ahora, obtengamos nuestro LLM y preparémonos para la acción. Y con acción, quiero decir hacer algo de RAG. Para hacer esta sección exactamente tal como está, necesitas una cuenta de OctoAI. También puedes optar por usar cualquier LLM de tu elección como reemplazo.
En este primer bloque de código, simplemente cargamos nuestras variables de entorno, extraemos nuestro token de API de OctoAI y ponemos en marcha su cliente.
from dotenv import load_dotenv
load_dotenv()
os.environ["OCTOAI_TOKEN"] = os.getenv("OCTOAI_API_TOKEN")
from octoai.client import Client
octo_client = Client()
Dar instrucciones al LLM
Hay dos cosas que el LLM necesita saber para hacer RAG: la pregunta y el contexto. Podemos pasar ambas a la vez creando una función que toma dos cadenas: la pregunta y el contexto. Usando esta función, utilizamos la finalización de chat del cliente de OctoAI para llamar a un LLM. Para este ejemplo, usamos el modelo Mixtral ajustado finamente de Nous Research.
Le damos a este modelo dos “mensajes” que indican cómo debe comportarse. Primero, le damos un mensaje al LLM para decirle que está respondiendo una pregunta del usuario basándose únicamente en el contexto dado. Luego, le decimos que habrá un usuario, y simplemente pasamos la pregunta.
Los otros parámetros son para ajustar el comportamiento del modelo. Podemos controlar el número máximo de tokens y qué tan “creativo” actúa el modelo.
Luego, la función devuelve la salida del cliente en formato JSON.
def dorag(question: str, context: str):
completion = octo_client.chat.completions.create(
messages=[
{
"role": "system",
"content": f"You are a helpful assistant. The user has a question. Answer the user question based only on the context: {context}"
},
{
"role": "user",
"content": f"{question}"
}
],
model="nous-hermes-2-mixtral-8x7b-dpo",
max_tokens=512,
presence_penalty=0,
temperature=0.1,
top_p=0.9,
)
return completion.model_dump()
Un ejemplo de RAG
Ahora estamos listos. Es hora de hacer una pregunta. Probablemente podríamos hacer esto sin crear una función, pero crear una función lo hace agradable y repetible. Esta función simplemente recibe una pregunta y luego hace RAG para responderla.
Comenzamos incrustando la pregunta usando el mismo modelo de embeddings que usamos para incrustar los documentos. A continuación, ejecutamos una búsqueda en Milvus. Observa que pasamos la pregunta a la función get_embeddings en formato de lista y luego pasamos la lista resultante directamente a la sección data de nuestra búsqueda de Milvus. Esto se debe simplemente a la forma en que están configuradas las firmas de las funciones; es más fácil reutilizarlas que reescribir varias funciones.
Dentro de nuestra llamada de búsqueda, también necesitamos proporcionar algunos parámetros más. anns_field le indica a Milvus en qué campo realizar una búsqueda aproximada de vecinos más cercanos (ANNS). También necesitamos pasar algunos parámetros para el índice. Asegúrate de que el tipo de métrica coincida con el que usamos para crear el índice, en este caso IP. También tenemos que usar un parámetro de índice coincidente, en este caso ef, o factor exploratorio.
Un ef más alto significa un mayor tiempo de búsqueda, pero un mayor recall. Puedes experimentar con esto, ef puede llegar hasta 2048; estamos usando 16 por velocidad y simplicidad. Solo hay miles de entradas en este conjunto de datos. A continuación, también pasamos un parámetro limit que nos indica cuántos resultados obtener de Milvus; para este ejemplo, podemos quedarnos simplemente con cinco.
El último parámetro de búsqueda define qué campos queremos que nos devuelva nuestra búsqueda. Para este ejemplo, podemos obtener simplemente sentence, que es el campo que usamos para almacenar nuestro fragmento de texto. Una vez que recibimos los resultados de búsqueda, necesitamos procesarlos. Milvus devuelve una entidad con hits dentro, así que simplemente tomamos la “sentence” de los cinco hits y las unimos con un punto para que formen un párrafo de lista.
Luego, pasamos la pregunta que hizo el usuario junto con ese párrafo a la función dorag que creamos arriba y simplemente devolvemos la respuesta.
def ask_a_question(question):
embeddings = get_embeddings([question])
res = collection.search(
data=embeddings, # search for the one (1) embedding returned as a list of lists
anns_field="embedding", # Search across embeddings
param={"metric_type": "IP",
"params": {"ef": 16}},
limit = 5, # get me the top 5 results
output_fields=["sentence"] # get the sentence/chunk and city
)
sentences = []
for hits in res:
for hit in hits:
sentences.append(hit.entity.get("sentence"))
context = ". ".join(sentences)
return dorag(question, context)
print(ask_a_question("What state is Cambridge in?")["choices"][0]["message"]["content"])
Para la pregunta de ejemplo que pregunta en qué estado está Cambridge, podemos simplemente imprimir la respuesta completa de OctoAI. Sin embargo, si nos tomamos el tiempo de analizarla, se ve mejor y debería indicarnos que Cambridge se encuentra en Massachusetts.
Resumen de BOM dot COM: BentoML, OctoAI y Milvus para RAG
Este ejemplo cubrió cómo puedes hacer RAG sin OpenAI ni un framework. Observa que, a diferencia de algunos de nuestros ejemplos anteriores, tampoco usamos LangChain ni LlamaIndex. Esta vez nuestro stack fue BOM.COM — BentoML, OctoAI y Milvus. Usamos las capacidades de serving de BentoML para servir un endpoint de modelo de embeddings, los endpoints LLM de OctoAI para acceder a un modelo de código abierto y Milvus como nuestra base de datos vectorial.
Hay muchas maneras de estructurar el orden en que usamos estas diferentes piezas del rompecabezas. Para este ejemplo, comenzamos poniendo en marcha un servidor local con BentoML para alojar un modelo de embeddings de Hugging Face. A continuación, pusimos en marcha una instancia local de Milvus usando Docker Compose.
Usamos un método sencillo para dividir nuestros datos en fragmentos, que se extrajeron de Wikipedia. Luego, tomamos esos fragmentos y los pasamos a nuestro modelo de embeddings, alojado en BentoML, para obtener los embeddings vectoriales que se insertarían en Milvus. Con todos los embeddings vectoriales en Milvus, estábamos totalmente listos para hacer RAG.
El LLM que elegimos esta vez fue el modelo Mixtral ajustado Nous Hermes, uno de los muchos modelos de código abierto disponibles en OctoAI. Creamos dos funciones para habilitar RAG. Una función que pasaba la pregunta y el contexto al LLM, dorag, y otra función que incrustaba la pregunta del usuario, buscaba en Milvus y luego pasaba los resultados de búsqueda junto con la pregunta a la función RAG original. Al final, probamos nuestro RAG con una pregunta sencilla como comprobación de coherencia.
Sigue leyendo
Why We Built Vector Lakebase: Rethinking Unstructured Data Architecture for AI
Vector Lakebase: a unified, lake-native data foundation for AI workloads — and an answer to what happens after vector databases succeed.
Balancing Precision and Performance: How Zilliz Cloud's New Parameters Help You Optimize Vector Search
Optimize vector search with Zilliz Cloud’s level and recall features to tune accuracy, balance performance, and power AI applications.
OpenAI o1: What Developers Need to Know
In this article, we will talk about the o1 series from a developer's perspective, exploring how these models can be implemented for sophisticated use cases.