RAG senza OpenAI: BentoML, OctoAI e Milvus
Questo articolo è stato originariamente pubblicato su The New Stack e viene ripubblicato qui con autorizzazione.
Le opzioni ampliate di retrieval augmented generation possono eliminare la dipendenza degli sviluppatori da OpenAI
ChatGPT ha portato l'AI al centro dell'attenzione pubblica nel 2023. Tuttavia, ora ci sono molte più opzioni, quindi non siamo più vincolati a OpenAI. Questa è la terza voce di una serie di blog su come puoi creare app di retrieval augmented generation (RAG) utilizzando LLM che non sono il GPT di OpenAI. Qui puoi trovare la parte 1 e la parte 2. Il repo GitHub per questo progetto può essere trovato qui.
In questo tutorial useremo BentoML per servire gli embedding, OctoAI per ottenere l'LLM e Milvus come nostro database vettoriale. Tratteremo:
Servire gli embedding con BentoML
Inserire i tuoi dati in un database vettoriale per RAG
Creare la tua collection Milvus
Analizzare e incorporare i tuoi dati per l'inserimento
Configurare il tuo LLM per RAG
Fornire istruzioni all'LLM
Un esempio di RAG
Riepilogo del BOM dot COM: BentoML, OctoAI e Milvus per RAG
Servire gli embedding con BentoML
Possiamo utilizzare sentence embedding serviti tramite BentoML usando il suo repository Sentence Transformers Embeddings. Vediamo brevemente cosa succede in questo repo. La cosa principale da esaminare è il file service.py. In sostanza, avvia un server e vi inserisce un endpoint API. All'interno dell'endpoint API, carica all-MiniLM-L6-v2 da Hugging Face e lo utilizza per creare embedding.
Quel repository avvia un server e ci fornisce un endpoint da raggiungere su <http://localhost:3000>. Per utilizzare questo endpoint, importiamo bentoml e avviamo un client HTTP usando il tipo di oggetto nativo SyncHTTPClient.
import bentoml
bento_client = bentoml.SyncHTTPClient("http://localhost:3000")
Una volta connessi al client, creiamo una funzione che ottiene un elenco di embedding da un elenco di stringhe. Una cosa da notare è che divido gli elenchi in 25 stringhe alla volta. Questo principalmente perché stiamo usando un endpoint sincrono. Suddividere l'elenco di stringhe rende le chiamate meno costose dal punto di vista computazionale ed evita i timeout.
Dopo aver suddiviso l'elenco in sezioni da 25, chiamiamo il bento_client che abbiamo creato sopra per codificare queste frasi. Il client BentoML restituisce un elenco di vettori, di fatto un elenco di elenchi. Prendiamo ciascuno di questi vettori e lo aggiungiamo al nostro elenco vuoto di embedding. Alla fine di questo ciclo restituiamo l'elenco finale di embedding.
Se nell'elenco dei testi non ci sono più di 25 stringhe, chiamiamo semplicemente il metodo encode del client sull'elenco di stringhe passato.
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,
)
Inserire i tuoi dati in un database vettoriale per RAG
Con la nostra funzione di embedding pronta, possiamo preparare i nostri dati da inserire in Milvus per la nostra applicazione RAG. Il primo passo in questa sezione è avviare e connettersi a Milvus. C’è un file docker-compose.yml situato nel repository collegato sopra. Puoi trovare anche il Milvus Docker Compose in questa pagina della documentazione.
Se hai Docker installato e hai scaricato quel repository, dovresti poter eseguire docker compose up -d per avviare Milvus. Una volta che il tuo server Milvus è attivo, è il momento di connettersi. Per questa parte, importiamo semplicemente il modulo connections e chiamiamo connect con l’host (localhost o 127.0.0.1) e la porta (19530). Il blocco di codice qui sotto definisce anche due costanti: un nome di collection e la dimensione. Puoi inventare qualsiasi nome di collection tu voglia. La dimensione deriva dalla dimensione del modello di embedding, all-MiniLM-L6-v2.
from pymilvus import connections
COLLECTION_NAME = "bmo_test"
DIMENSION = 384
connections.connect(host="localhost", port=19530)
Creazione della tua collection Milvus
La creazione di una collection in Milvus comporta due passaggi: innanzitutto, la definizione dello schema e, in secondo luogo, la definizione dell’indice. Per questa sezione, abbiamo bisogno di quattro moduli: FieldSchema definisce un campo, CollectionSchema definisce una collection, DataType ci dice quale tipo di dati sarà contenuto in un campo e Collection è l’oggetto che Milvus usa per creare collection.
Possiamo definire qui l’intero schema per la collection. Oppure, possiamo semplicemente definire i due elementi necessari: id ed embedding. Poi, quando arriva il momento di definire lo schema, passiamo un parametro, enabled_dynamic_field, che ci consente di inserire qualsiasi campo desideriamo purché siano presenti anche i campi id ed embedding. Questo ci permette di trattare l’inserimento dei dati in Milvus nello stesso modo in cui tratteremmo un database NoSQL come MongoDB. Successivamente, creiamo semplicemente la collection con il nome e lo schema forniti in precedenza.
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)
Ora che abbiamo creato la nostra collection, dobbiamo definire l’indice. In termini di ricerca, un “indice” definisce come mapperemo i nostri dati per il recupero. Usiamo HSNW (hierarchical navigable small worlds) per indicizzare i nostri dati per questo progetto. Dobbiamo anche definire come andremo a misurare la distanza vettoriale. In questo esempio, usiamo il prodotto interno o IP.
Ciascuno degli 11 tipi di indice offerti in Milvus ha un insieme diverso di parametri. Per HNSW, abbiamo due parametri da regolare: “M” ed “efConstruction”. “M” è il limite superiore per il grado di un nodo in ciascun grafo, mentre “efConstruction” è il fattore esplorativo utilizzato durante la costruzione dell’indice.
Da un punto di vista pratico, valori più alti di “M” ed “efConstruction” portano entrambi a una ricerca migliore. Un valore “M” più alto significa che l’indice occuperà più memoria. Un valore “efConstruction” più alto significa che ci vorrà più tempo per costruire l’indice. Dovrai fare qualche prova con questi parametri per trovare i valori migliori.
Una volta definito l’indice, lo creiamo su un campo scelto, in questo caso embedding. Poi chiamiamo load per caricare la collection in 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()
Analisi e embedding dei tuoi dati per l’inserimento
Con Milvus pronto e la connessione stabilita, possiamo inserire i dati nel nostro database vettoriale. Ma prima dobbiamo preparare i dati da inserire. Per questo esempio, abbiamo diversi file txt, disponibili nella cartella data del repo. Dividiamo questi dati in blocchi, ne creiamo gli embedding e li archiviamo in Milvus.
Iniziamo creando una funzione che suddivide questo testo in blocchi. Esistono molti modi per fare il chunking, ma per questo esempio lo faremo in modo semplice. La funzione qui sotto prende un file, lo legge come stringa e poi lo divide a ogni nuova riga. Restituisce la lista di stringhe appena creata.
# 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
Successivamente, elaboriamo ciascuno dei file che abbiamo. Otteniamo una lista di tutti i nomi dei file e creiamo una lista vuota per contenere le informazioni suddivise in blocchi. Poi, iteriamo su tutti i file ed eseguiamo la funzione sopra su ciascuno per ottenere un chunking semplice di ogni file. Prima di archiviare i blocchi, dobbiamo pulirli.
Se osservi come viene suddiviso un singolo file, vedrai molte righe vuote, e non vogliamo righe vuote. Alcune righe sono solo tabulazioni o altri caratteri speciali. Per evitarle, creiamo una lista vuota e archiviamo solo i blocchi al di sopra di una certa lunghezza. Per semplicità, possiamo usare sette caratteri.
Una volta ottenuta una lista pulita di blocchi da ciascun documento, possiamo archiviare i nostri dati. Creiamo un dizionario che mappa ogni lista di blocchi al nome del documento, in questo caso il nome della città. Poi aggiungiamo tutti questi elementi alla lista vuota che abbiamo creato sopra.
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 insieme di testi suddivisi in blocchi per ogni città pronto, è il momento di ottenere alcuni embedding. Milvus può accettare una lista di dizionari da inserire in una collection, quindi possiamo iniziare con un’altra lista vuota. Per ciascuno dei dizionari creati sopra, dobbiamo ottenere una lista di embedding che corrisponda alla lista di frasi.
Lo facciamo chiamando direttamente la funzione get_embeddings che abbiamo creato nella sezione usando BentoML su ciascuno dell’elenco di chunk. Ora dobbiamo abbinarli. Poiché gli embedding dell’elenco e l’elenco delle frasi dovrebbero corrispondere per indice, possiamo usare enumerate su uno dei due elenchi per abbinarli.
Li abbiniamo creando un dizionario che rappresenta una singola entry in Milvus. Ogni entry include l’embedding, la frase correlata e la città. Includere la città è facoltativo, ma includiamola così possiamo usarla. Nota che non è necessario includere un id in questa entry. Questo perché abbiamo scelto di auto-incrementare l’id quando abbiamo creato lo schema sopra.
Aggiungiamo ciascuna di queste entry all’elenco mentre le attraversiamo nel loop. Alla fine, abbiamo un elenco di dizionari in cui ogni dizionario rappresenta una entry a riga singola per Milvus. Possiamo quindi semplicemente inserire queste entry nella nostra collection Milvus. L’ultimo passaggio qui è fare il flush delle entry così possiamo iniziare a indicizzarle.
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 il tuo LLM per RAG
Ora prendiamo il nostro LLM e prepariamoci a entrare in azione. E per azione, intendo fare un po’ di RAG. Per seguire questa sezione esattamente così com’è, hai bisogno di un account su OctoAI. Puoi anche scegliere di inserire qualsiasi LLM di tua scelta come sostituto.
In questo primo blocco di codice, carichiamo semplicemente le nostre variabili d’ambiente, estraiamo il nostro token API OctoAI e avviamo il loro client.
from dotenv import load_dotenv
load_dotenv()
os.environ["OCTOAI_TOKEN"] = os.getenv("OCTOAI_API_TOKEN")
from octoai.client import Client
octo_client = Client()
Dare istruzioni all’LLM
Ci sono due cose che l’LLM deve sapere per fare RAG: la domanda e il contesto. Possiamo passarle entrambe insieme creando una funzione che accetta due stringhe: la domanda e il contesto. Usando questa funzione, utilizziamo il chat completion client di OctoAI per chiamare un LLM. Per questo esempio, usiamo il modello Mixtral fine-tuned di Nous Research.
Diamo a questo modello due “messaggi” che indicano come dovrebbe comportarsi. Per prima cosa, diamo un messaggio all’LLM per dirgli che sta rispondendo a una domanda dell’utente basandosi esclusivamente sul contesto fornito. Poi gli diciamo che ci sarà un utente, e passiamo semplicemente la domanda.
Gli altri parametri servono per regolare il comportamento del modello. Possiamo controllare il numero massimo di token e quanto “creativo” si comporta il modello.
La funzione restituisce quindi l’output del client in 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 esempio di RAG
Ora siamo pronti. È il momento di fare una domanda. Probabilmente potremmo farlo senza creare una funzione, ma creare una funzione lo rende comodo e ripetibile. Questa funzione accetta semplicemente una domanda e poi fa RAG per risponderle.
Iniziamo incorporando la domanda usando lo stesso modello di embedding che abbiamo usato per incorporare i documenti. Successivamente, eseguiamo una ricerca su Milvus. Nota che passiamo la domanda alla funzione get_embeddings in formato lista, e poi passiamo la lista restituita direttamente nella sezione data della nostra ricerca Milvus. Questo dipende semplicemente da come sono impostate le firme delle funzioni: è più facile riutilizzarle che riscrivere più funzioni.
All’interno della nostra chiamata di ricerca, dobbiamo anche fornire alcuni altri parametri. anns_field indica a Milvus su quale campo eseguire una ricerca approssimata del vicino più prossimo (ANNS). Dobbiamo anche passare alcuni parametri per l’indice. Assicurati che il tipo di metrica corrisponda a quello che abbiamo usato per creare l’indice, in questo caso IP. Dobbiamo anche usare un parametro di indice corrispondente, in questo caso ef, o fattore esplorativo.
Un ef più alto significa un tempo di ricerca più lungo, ma un richiamo più alto. Puoi sperimentare con questo: ef può arrivare fino a 2048; noi usiamo 16 per velocità e semplicità. In questo dataset ci sono solo migliaia di voci. Successivamente, passiamo anche un parametro limit che ci dice quanti risultati ottenere da Milvus; per questo esempio, possiamo semplicemente scegliere cinque.
L’ultimo parametro di ricerca definisce quali campi vogliamo ottenere dalla nostra ricerca. Per questo esempio, possiamo semplicemente ottenere sentence, che è il campo che abbiamo usato per archiviare il nostro blocco di testo. Una volta ottenuti i risultati della ricerca, dobbiamo elaborarli. Milvus restituisce un’entità con hits al suo interno, quindi prendiamo semplicemente la “sentence” da tutti e cinque gli hit e le uniamo con un punto, in modo che formi un paragrafo elenco.
Poi, passiamo la domanda posta dall’utente insieme a quel paragrafo nella funzione dorag che abbiamo creato sopra e restituiamo semplicemente la risposta.
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"])
Per la domanda di esempio che chiede in quale stato si trovi Cambridge, possiamo semplicemente stampare l’intera risposta da OctoAI. Tuttavia, se ci prendiamo il tempo di analizzarla, ha semplicemente un aspetto migliore e dovrebbe dirci che Cambridge si trova in Massachusetts.
Riepilogo del BOM dot COM: BentoML, OctoAI e Milvus per RAG
Questo esempio ha mostrato come puoi fare RAG senza OpenAI o un framework. Nota che, a differenza di alcuni dei nostri esempi precedenti, non abbiamo usato nemmeno LangChain o LlamaIndex. Questa volta il nostro stack era il BOM.COM — BentoML, OctoAI e Milvus. Abbiamo usato le capacità di serving di BentoML per servire un endpoint del modello di embedding, gli endpoint LLM di OctoAI per accedere a un modello open source e Milvus come nostro database vettoriale.
Esistono molti modi per strutturare l’ordine in cui utilizziamo questi diversi tasselli del puzzle. Per questo esempio, abbiamo iniziato avviando un server locale con BentoML per ospitare un modello di embedding da Hugging Face. Poi, abbiamo avviato un’istanza locale di Milvus usando Docker Compose.
Abbiamo usato un metodo semplice per suddividere in blocchi i nostri dati, che erano stati estratti da Wikipedia. Poi, abbiamo preso quei blocchi e li abbiamo passati al nostro modello di embedding, ospitato su BentoML, per ottenere gli embedding vettoriali da inserire in Milvus. Con tutti gli embedding vettoriali in Milvus, eravamo completamente pronti per fare RAG.
L’LLM che abbiamo scelto questa volta è stato il modello Mixtral fine-tuned Nous Hermes, uno dei tanti modelli open source disponibili su OctoAI. Abbiamo creato due funzioni per abilitare RAG. Una funzione che passava la domanda e il contesto all’LLM, dorag, e un’altra funzione che eseguiva l’embedding della domanda dell’utente, cercava in Milvus e poi passava i risultati della ricerca insieme alla domanda alla funzione RAG originale. Alla fine, abbiamo testato il nostro RAG con una domanda semplice come controllo di correttezza.
Continua a leggere
Zilliz Cloud Now Available in AWS Asia Pacific (Seoul)
Zilliz Cloud is now available in AWS Seoul — low-latency vector search, in-country data residency, and one-step migration for Korean AI teams. 31 regions across 5 clouds.
Introducing Functions and Model Inference on Zilliz Cloud: Automatic Embedding and Reranking with Hosted Models
Zilliz Cloud Functions auto-generate embeddings via OpenAI, Voyage AI, Cohere, or Zilliz Hosted Models. Built-in reranking — just insert text and search.
Build for the Boom: Why AI Agent Startups Should Build Scalable Infrastructure Early
Explore strategies for developing AI agents that can handle rapid growth. Don't let inadequate systems undermine your success during critical breakthrough moments.