Construire RAG avec Milvus et Fireworks AI

Fireworks AI est une plateforme d'inférence générative d'IA qui offre une vitesse et une préparation à la production de pointe pour l'exécution et la personnalisation des modèles. Fireworks AI fournit une variété de services d'IA générative, y compris des modèles sans serveur, des déploiements à la demande et des capacités de réglage fin. Il offre un environnement complet pour le déploiement de divers modèles d'IA, y compris des modèles de langage de grande taille (LLM) et des modèles d'intégration. Fireworks AI regroupe de nombreux modèles, ce qui permet aux utilisateurs d'accéder facilement à ces ressources et de les utiliser sans avoir à mettre en place une infrastructure étendue.

Dans ce tutoriel, nous allons vous montrer comment construire un pipeline RAG (Retrieval-Augmented Generation) avec Milvus et Fireworks AI.

Préparation

Dépendances et environnement

$ pip install --upgrade pymilvus openai requests tqdm

Fireworks AI utilise l'API de type OpenAI. Vous pouvez vous connecter à son site officiel et préparer la clé api FIREWORKS_API_KEY comme variable d'environnement.

import os

os.environ["FIREWORKS_API_KEY"] = "***********"

Préparer les données

Nous utilisons les pages FAQ de la documentation Milvus 2.4.x comme connaissance privée dans notre RAG, qui est une bonne source de données pour un pipeline RAG simple.

Téléchargez le fichier zip et extrayez les documents dans le dossier milvus_docs.

$ wget https://github.com/milvus-io/milvus-docs/releases/download/v2.4.6-preview/milvus_docs_2.4.x_en.zip
$ unzip -q milvus_docs_2.4.x_en.zip -d milvus_docs

Nous chargeons tous les fichiers markdown à partir du dossier milvus_docs/en/faq. Pour chaque document, nous utilisons simplement "# " pour séparer le contenu du fichier, ce qui permet de séparer grossièrement le contenu de chaque partie principale du fichier markdown.

from glob import glob

text_lines = []

for file_path in glob("milvus_docs/en/faq/*.md", recursive=True):
    with open(file_path, "r") as file:
        file_text = file.read()

    text_lines += file_text.split("# ")

Préparer le LLM et le modèle d'intégration

Nous initialisons un client pour préparer le LLM et le modèle d'intégration. Fireworks AI active l'API de type OpenAI et vous pouvez utiliser la même API avec des ajustements mineurs pour appeler le modèle d'intégration et le LLM.

from openai import OpenAI

fireworks_client = OpenAI(
    api_key=os.environ["FIREWORKS_API_KEY"],
    base_url="https://api.fireworks.ai/inference/v1",
)

Définissez une fonction pour générer des enchâssements de texte à l'aide du client. Nous utilisons le modèle nomic-ai/nomic-embed-text-v1.5 comme exemple.

def emb_text(text):
    return (
        fireworks_client.embeddings.create(
            input=text, model="nomic-ai/nomic-embed-text-v1.5"
        )
        .data[0]
        .embedding
    )

Générez un embedding de test et imprimez sa dimension et ses premiers éléments.

test_embedding = emb_text("This is a test")
embedding_dim = len(test_embedding)
print(embedding_dim)
print(test_embedding[:10])

768
[0.04815673828125, 0.0261993408203125, -0.1749267578125, -0.03131103515625, 0.068115234375, -0.00621795654296875, 0.03955078125, -0.0210723876953125, 0.039703369140625, -0.0286102294921875]

Chargement des données dans Milvus

Créer la collection

from pymilvus import MilvusClient

milvus_client = MilvusClient(uri="./milvus_demo.db")

collection_name = "my_rag_collection"

Vérifier si la collection existe déjà et la supprimer si c'est le cas.

if milvus_client.has_collection(collection_name):
    milvus_client.drop_collection(collection_name)

Créer une nouvelle collection avec les paramètres spécifiés.

Si nous ne spécifions aucune information de champ, Milvus créera automatiquement un champ id par défaut pour la clé primaire et un champ vector pour stocker les données vectorielles. Un champ JSON réservé est utilisé pour stocker les champs non définis par le schéma et leurs valeurs.

milvus_client.create_collection(
    collection_name=collection_name,
    dimension=embedding_dim,
    metric_type="IP",  # Inner product distance
    consistency_level="Strong",  # Strong consistency level
)

Insérer des données

Parcourez les lignes de texte, créez des enchâssements, puis insérez les données dans Milvus.

Voici un nouveau champ text, qui est un champ non défini dans le schéma de la collection. Il sera automatiquement ajouté au champ dynamique JSON réservé, qui peut être traité comme un champ normal à un niveau élevé.

from tqdm import tqdm

data = []

for i, line in enumerate(tqdm(text_lines, desc="Creating embeddings")):
    data.append({"id": i, "vector": emb_text(line), "text": line})

milvus_client.insert(collection_name=collection_name, data=data)

Creating embeddings: 100%|██████████| 72/72 [00:28<00:00,  2.51it/s]





{'insert_count': 72, 'ids': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71], 'cost': 0}

Construire un RAG

Récupérer des données pour une requête

Spécifions une question fréquente sur Milvus.

question = "How is data stored in milvus?"

Cherchons la question dans la collection et récupérons les 3 meilleures réponses sémantiques.

search_res = milvus_client.search(
    collection_name=collection_name,
    data=[
        emb_text(question)
    ],  # Use the `emb_text` function to convert the question to an embedding vector
    limit=3,  # Return top 3 results
    search_params={"metric_type": "IP", "params": {}},  # Inner product distance
    output_fields=["text"],  # Return the text field
)

Jetons un coup d'œil aux résultats de la recherche de la question.

import json

retrieved_lines_with_distances = [
    (res["entity"]["text"], res["distance"]) for res in search_res[0]
]
print(json.dumps(retrieved_lines_with_distances, indent=4))

[
    [
        " Where does Milvus store data?\n\nMilvus deals with two types of data, inserted data and metadata. \n\nInserted data, including vector data, scalar data, and collection-specific schema, are stored in persistent storage as incremental log. Milvus supports multiple object storage backends, including [MinIO](https://min.io/), [AWS S3](https://aws.amazon.com/s3/?nc1=h_ls), [Google Cloud Storage](https://cloud.google.com/storage?hl=en#object-storage-for-companies-of-all-sizes) (GCS), [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs), [Alibaba Cloud OSS](https://www.alibabacloud.com/product/object-storage-service), and [Tencent Cloud Object Storage](https://www.tencentcloud.com/products/cos) (COS).\n\nMetadata are generated within Milvus. Each Milvus module has its own metadata that are stored in etcd.\n\n###",
        0.8334928750991821
    ],
    [
        "How does Milvus flush data?\n\nMilvus returns success when inserted data are loaded to the message queue. However, the data are not yet flushed to the disk. Then Milvus' data node writes the data in the message queue to persistent storage as incremental logs. If `flush()` is called, the data node is forced to write all data in the message queue to persistent storage immediately.\n\n###",
        0.746377170085907
    ],
    [
        "What is the maximum dataset size Milvus can handle?\n\n  \nTheoretically, the maximum dataset size Milvus can handle is determined by the hardware it is run on, specifically system memory and storage:\n\n- Milvus loads all specified collections and partitions into memory before running queries. Therefore, memory size determines the maximum amount of data Milvus can query.\n- When new entities and and collection-related schema (currently only MinIO is supported for data persistence) are added to Milvus, system storage determines the maximum allowable size of inserted data.\n\n###",
        0.7328270673751831
    ]
]

Utiliser LLM pour obtenir une réponse RAG

Convertir les documents récupérés dans un format de chaîne.

context = "\n".join(
    [line_with_distance[0] for line_with_distance in retrieved_lines_with_distances]
)

Définir les messages-guides du système et de l'utilisateur pour le modèle de langue. Cette invite est assemblée avec les documents récupérés de Milvus.

SYSTEM_PROMPT = """
Human: You are an AI assistant. You are able to find answers to the questions from the contextual passage snippets provided.
"""
USER_PROMPT = f"""
Use the following pieces of information enclosed in <context> tags to provide an answer to the question enclosed in <question> tags.
<context>
{context}
</context>
<question>
{question}
</question>
"""

Utiliser le modèle llama-v3p1-405b-instruct fourni par Fireworks pour générer une réponse basée sur les invites.

response = fireworks_client.chat.completions.create(
    model="accounts/fireworks/models/llama-v3p1-405b-instruct",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": USER_PROMPT},
    ],
)
print(response.choices[0].message.content)

According to the provided context, Milvus stores data in two ways:

1. Inserted data, including vector data, scalar data, and collection-specific schema, are stored in persistent storage as incremental logs. This can be done using multiple object storage backends such as MinIO, AWS S3, Google Cloud Storage, Azure Blob Storage, Alibaba Cloud OSS, and Tencent Cloud Object Storage.
2. Metadata, which are generated within Milvus, are stored in etcd, with each Milvus module having its own metadata.

Additionally, when data is inserted, it is first loaded into a message queue, and then written to persistent storage as incremental logs by the data node. The `flush()` function can be used to force the data node to write all data in the message queue to persistent storage immediately.

C'est super ! Nous avons réussi à construire un pipeline RAG avec Milvus et Fireworks AI.