Cognita : Création d'applications RAG modulaires et open source pour la production

Supposons qu'il existe un équipe A chargé de développer l'application RAG pour cas d'utilisation 1, alors il y a équipe B qui développe une application RAG pour cas d'utilisation 2, et puis il y a équipe C, qui ne fait que planifier le prochain cas d'utilisation de son application RAG. Avez-vous souhaité qu'il soit facile de créer des pipelines RAG pour plusieurs équipes ? Chaque équipe n'a pas besoin de partir de zéro, mais d'une manière modulaire permettant à chaque équipe d'utiliser les mêmes fonctionnalités de base et de développer efficacement ses propres applications en plus de celles-ci, sans aucune interférence ?

Ne t'inquiète pas ! C'est pourquoi nous avons créé Cognita. Bien que RAG soit indéniablement impressionnant, le processus de création d'une application fonctionnelle avec celui-ci peut être intimidant. Il y a beaucoup à comprendre en ce qui concerne les pratiques de mise en œuvre et de développement, allant de la sélection des modèles d'IA appropriés pour un cas d'utilisation spécifique à l'organisation efficace des données pour obtenir les informations souhaitées. Alors que des outils tels que Chaîne Lang et Indice de lama existent pour simplifier le processus de conception des prototypes, il n'existe pas encore de modèle RAG open source accessible et prêt à l'emploi qui intègre les meilleures pratiques et offre un support modulaire, permettant à quiconque de l'utiliser rapidement et facilement.

Avantages de Cognita :

1. Un référentiel central réutilisable d'analyseurs, de chargeurs, d'intégrateurs et de récupérateurs.
2. Possibilité pour les utilisateurs non techniques de jouer avec l'interface utilisateur - Téléchargez des documents et exécutez QnA à l'aide de modules conçus par l'équipe de développement.
3. Entièrement piloté par API, ce qui permet l'intégration avec d'autres systèmes.

Vue d'ensemble

En explorant le fonctionnement interne de Cognita, notre objectif était de trouver un équilibre entre personnalisation complète et adaptabilité tout en garantissant la facilité d'utilisation dès le départ. Compte tenu du rythme rapide des avancées en matière de RAG et d'IA, il était impératif pour nous de concevoir Cognita en tenant compte de l'évolutivité, afin de permettre une intégration fluide des nouvelles avancées et des divers cas d'utilisation. Cela nous a amenés à décomposer le processus RAG en étapes modulaires distinctes (comme le montre le schéma ci-dessus, qui sera discuté dans les sections suivantes), facilitant ainsi la maintenance du système, l'ajout de nouvelles fonctionnalités telles que l'interopérabilité avec d'autres bibliothèques d'IA et permettant aux utilisateurs d'adapter la plateforme à leurs besoins spécifiques. Notre objectif reste de fournir aux utilisateurs un outil robuste qui non seulement répond à leurs besoins actuels, mais qui évolue également en même temps que la technologie, y compris des changements architecturaux plus généraux tels que MCP contre RAG, garantissant une valeur à long terme.

Composantes

Cognita est conçu autour de sept modules différents, chacun étant personnalisable et contrôlable pour répondre à différents besoins :

1. Chargeurs de données
2. Analyseurs
3. Intégrateurs
4. Reclassement
5. DB vectorielle
6. Magasin de métadonnées
7. Contrôleurs de requêtes

Chargeurs de données

Ceux-ci chargent les données à partir de différentes sources telles que les répertoires locaux, les compartiments S3, les bases de données, Véritable fonderie artefacts, etc. Cognita prend actuellement en charge le chargement de données depuis le répertoire local, l'URL Web, le référentiel Github et les artefacts Truefoundry. D'autres chargeurs de données peuvent être facilement ajoutés sous backend/modules/chargeurs de données/ . Une fois qu'un chargeur de données est ajouté, vous devez l'enregistrer afin qu'il puisse être utilisé par l'application RAG sous backend/modules/dataloaders/__init__.py Pour enregistrer un chargeur de données, ajoutez les éléments suivants :

register_dataloader("MyDataLoader", MyDataLoaderClass)

Analyseurs

Au cours de cette étape, nous traitons différents types de données, tels que des fichiers texte classiques, des PDF et même des fichiers Markdown. L'objectif est de transformer tous ces différents types en un format commun afin que nous puissions les utiliser plus facilement par la suite. Cette partie, appelée analyse syntaxique, est généralement la plus longue et est difficile à implémenter lorsque nous configurons un système comme celui-ci. Mais l'utilisation de Cognita peut être utile, car elle gère déjà pour nous le dur labeur que représente la gestion des pipelines de données.

Après cela, nous avons divisé les données analysées en morceaux uniformes. Mais pourquoi en avons-nous besoin ? Le texte que nous obtenons des fichiers peut être de différentes longueurs. Si nous utilisons directement ces longs textes, nous finirons par ajouter un tas d'informations inutiles. De plus, étant donné que tous les LLM ne peuvent traiter qu'une certaine quantité de texte à la fois, nous ne serons pas en mesure d'inclure tout le contexte important nécessaire à la question. Nous allons donc plutôt décomposer le texte en parties plus petites pour chaque section. Intuitivement, les petits morceaux contiendront des concepts pertinents et seront moins bruyants que les gros morceaux.

Actuellement, nous prenons en charge l'analyse de Markdown, PDF et Texte fichiers. D'autres analyseurs de données peuvent être facilement ajoutés sous backend/modules/analyseurs/ . Une fois qu'un analyseur est ajouté, vous devez l'enregistrer afin qu'il puisse être utilisé par l'application RAG sous backend/modules/parsers/__init__.py Pour enregistrer un analyseur, ajoutez ce qui suit :

register_parser("MyParser", MyParserClass)

Intégrateurs

Une fois que nous avons divisé les données en plus petits morceaux, nous voulons trouver les éléments les plus importants pour une question spécifique. Un moyen rapide et efficace d'y parvenir consiste à utiliser un modèle pré-entraîné (modèle d'intégration) pour convertir nos données et la question en codes spéciaux appelés intégrations. Ensuite, nous comparons les intégrations de chaque bloc de données à celui de la question. En mesurant similitude en cosinus entre ces intégrations, nous pouvons déterminer quels morceaux sont les plus étroitement liés à la question, ce qui nous aide à trouver les meilleurs à utiliser.

Il existe de nombreux modèles pré-entraînés disponibles pour intégrer les données, tels que les modèles d'OpenAI, de Cohere, etc. Les modèles les plus populaires peuvent être découverts via Le benchmark d'intégration de texte massif de HuggingFace (MTEB) classement. Nous fournissons un support pour OpenAI Embeddings, TrueFoundry Embeddings et également les versions actuelles SOTA intégrations (en avril 2024) de pain mixé-ai.

D'autres intégrateurs peuvent être facilement ajoutés sous backend/modules/embedder/ . Une fois qu'un intégrateur est ajouté, vous devez l'enregistrer afin qu'il puisse être utilisé par l'application RAG sous backend/modules/embedders/__init__.py Pour enregistrer un analyseur, ajoutez ce qui suit :

register_embedder("MyEmbedder", MyEmbedderClass)

‍Remarque : N'oubliez pas que les intégrations ne sont pas la seule méthode pour trouver des morceaux importants. Nous pourrions également utiliser un LLM pour cette tâche ! Cependant, les LLM sont beaucoup plus volumineux que les modèles d'intégration et ont une limite quant à la quantité de texte qu'ils peuvent gérer à la fois. C'est pourquoi il est plus intelligent d'utiliser les intégrations pour sélectionner en premier les k meilleurs morceaux. Ensuite, nous pouvons utiliser les LLM sur ces quelques segments afin de déterminer les meilleurs à utiliser comme contexte pour répondre à notre question.

Reclassement

Une fois que l'étape d'intégration trouve des correspondances potentielles, ce qui peut être beaucoup, une étape de reclassement est appliquée. Reclassement pour s'assurer que les meilleurs résultats sont au sommet. Par conséquent, nous pouvons choisir les x meilleurs documents, ce qui rend notre contexte plus concis et les requêtes rapides plus courtes. Nous fournissons le soutien pour SOTA reclassement (en avril 2024) de pain mixé-ai qui est mis en œuvre dans backend/modules/reanker/

Vecteur DB

Une fois que nous avons créé des vecteurs pour les textes, nous les stockons dans ce que l'on appelle une base de données vectorielles. Cette base de données garde une trace de ces vecteurs afin que nous puissions les retrouver rapidement ultérieurement en utilisant différentes méthodes. Les bases de données classiques organisent les données dans des tableaux, tels que des lignes et des colonnes, mais les bases de données vectorielles sont spéciales car elles stockent et trouvent des données en fonction de ces vecteurs. C'est très utile pour des choses comme reconnaître des images, comprendre un langage ou recommander des choses. Par exemple, dans un système de recommandation, chaque article que vous souhaitez recommander (comme un film ou un produit) est transformé en vecteur, avec différentes parties du vecteur représentant différentes caractéristiques de l'article, comme son genre ou son prix. De même, dans le domaine du langage, chaque mot ou document est transformé en vecteur, certaines parties du vecteur représentant les caractéristiques du mot ou du document, comme la fréquence à laquelle le mot est utilisé ou sa signification. Ces bases de données vectorielles sont conçues pour les gérer efficacement. En utilisant différentes méthodes pour mesurer la proximité des vecteurs les uns par rapport aux autres, par exemple leur similitude ou leur distance, nous trouvons les vecteurs les plus proches de la requête utilisateur donnée. Les méthodes les plus courantes pour mesurer cela sont la distance euclidienne, la similarité des cosinus et le produit scalaire.

Il existe plusieurs bases de données vectorielles disponibles sur le marché, telles que Qdrant, SingleStore, Weaviate, etc. Nous prenons actuellement en charge Qdrant et Boutique unique. La classe de base de données du vecteur Qdrant est définie sous /backend/modules/vector_db/qdrant.py, tandis que la classe de base de données vectorielle SingleStore est définie sous /backend/modules/vector_db/singlestore.py

D'autres dbs vectoriels peuvent également être ajoutés dans le vecteur_db dossier et peut être enregistré sous /backend/modules/vector_db/__init__.py

Pour ajouter un support de base de données vectorielle dans Cognita, l'utilisateur doit procéder comme suit :

• Créez la classe qui hérite de Vecteur de base DB (depuis backend.modules.vector_db.base, importez BaseVectorDB) et initialisez-le avec Configuration vectorielle DBConfig (depuis backend.types import VectorDBConfig)
• Mettez en œuvre les méthodes suivantes :
  • créer_collection: Pour initialiser la collection/le projet/la table dans la base de données vectorielle.
  • documents bouleversés: Pour insérer les documents dans la base de données.
  • obtenir_collections: Obtenez toutes les collections présentes dans la base de données.
  • supprimer_collection: Pour supprimer la collection de la base de données.
  • get_vector_store: Pour obtenir la boutique vectorielle de la collection donnée.
  • obtenir_vector_client: Pour obtenir le client vectoriel pour la collection donnée, le cas échéant.
  • liste_point_de_données_vecteurs: Pour répertorier les vecteurs déjà présents dans la base de données qui sont similaires aux documents en cours d'insertion.
  • supprime_données_point_vecteurs: Pour supprimer les vecteurs de la base de données, utilisé pour supprimer les anciens vecteurs du document mis à jour.

Nous montrons maintenant comment ajouter une nouvelle base de données vectorielle au système RAG. Nous prenons l'exemple des deux Qdrant et Boutique unique vecteur dbs.

Intégration de Qdrant

Qdrant est une base de données vectorielle open source et un moteur de recherche vectorielle écrits en Rust. Il fournit un vecteur rapide et évolutif recherche de similarité service avec une API pratique. Pour ajouter la base de données vectorielles Qdrant au système RAG, procédez comme suit :

Dans le fichier .env, vous pouvez ajouter les éléments suivants

VECTOR_DB_CONFIG = '{"url » : "<url_here>«, « provider » : « qdrant"}' # URL Qdrant pour l'instance déployée

VECTOR_DB_CONFIG=' {"provider » :"qdrant », « local » :"true "} ' # Pour une instance Qdrant locale basée sur un fichier sans docker

1. Créer une nouvelle classe Vecteur QDRANTDB dans backend/modules/vector_db/qdrant.py qui hérite de Vecteur de base DB et initialisez-le avec Configuration vectorielle DBConfig

from langchain_community.vectorstores.qdrant import Qdrant
from qdrant_client.http.models import VectorParams, Distance
from qdrant_client import QdrantClient, models

class QdrantVectorDB(BaseVectorDB):
    def __init__(self, config: VectorDBConfig):
        ...
        # Initialize the qdrant client
        self.qdrant_client = QdrantClient(
            url=self.url,
            port=self.port,
            prefer_grpc=self.prefer_grpc,
            prefix=self.prefix,
        )

2. Remplacez le créer_collection méthode pour créer une collection dans Qdrant

 def create_collection(self, collection_name: str, embeddings: Embeddings):
    # Calculate embedding size
    partial_embeddings = embeddings.embed_documents(["Initial document"])
    vector_size = len(partial_embeddings[0])

    # Create collection given the embeding dimension and simalrity metric
    self.qdrant_client.create_collection(
        collection_name=collection_name,
        vectors_config=VectorParams(
            size=vector_size,  # embedding dimension
            distance=Distance.COSINE,
        ),
        replication_factor=3,
    )

    # Create metadata index for the collection
    self.qdrant_client.create_payload_index(
        collection_name=collection_name,
        field_name=f"metadata.{DATA_POINT_FQN_METADATA_KEY}",
        field_schema=models.PayloadSchemaType.KEYWORD,
    )

3. Remplacez le documents bouleversés méthode pour insérer les documents dans la base de données

def upsert_documents(
    self,
    collection_name: str,
    documents,
    embeddings: Embeddings,
    incremental: bool = True,
):
    # Get the data point fqns from the documents
    # fqns uniquely identify the documents
    # This is used to delete the outdated documents from the db
    data_point_fqns = []
    for document in documents:
        if document.metadata.get(DATA_POINT_FQN_METADATA_KEY):
            data_point_fqns.append(
                document.metadata.get(DATA_POINT_FQN_METADATA_KEY)
            )

    # Add Documents
    Qdrant(
        client=self.qdrant_client,
        collection_name=collection_name,
        embeddings=embeddings,
    ).add_documents(documents=documents)

    # Get the record ids that are already present in the db (Existing documents that are modified)
    # These docs should be removed from the db
    # Used when incremental indexing is enabled
    record_ids_to_be_upserted: List[str] = self._get_records_to_be_upserted(
        collection_name=collection_name,
        data_point_fqns=data_point_fqns,
        incremental=incremental,
    )

    # Delete Documents
    if len(record_ids_to_be_upserted):
        for i in range(0, len(record_ids_to_be_upserted), BATCH_SIZE):
            record_ids_to_be_processed = record_ids_to_be_upserted[
                i : i + BATCH_SIZE
            ]
            self.qdrant_client.delete(
                collection_name=collection_name,
                points_selector=models.PointIdsList(
                    points=record_ids_to_be_processed,
                ),
            )

# This method is used to get the records that are already present in the db
def _get_records_to_be_upserted(
    self, collection_name: str,
    data_point_fqns: List[str],
    incremental: bool
):
    if not incremental:
        return []
    # For incremental deletion, we delete the documents with the same document_id
    stop = False
    offset = None
    record_ids_to_be_upserted = []

    # we fetch the records in batches based on the data_point_fqns
    while stop is not True:
        records, next_offset = self.qdrant_client.scroll(
            collection_name=collection_name,
            scroll_filter=models.Filter(
                should=[
                    models.FieldCondition(
                        key=f"metadata.{DATA_POINT_FQN_METADATA_KEY}",
                        match=models.MatchAny(
                            any=data_point_fqns,
                        ),
                    ),
                ]
            ),
            limit=BATCH_SIZE,
            offset=offset,
            with_payload=False,
            with_vectors=False,
        )
        for record in records:
            record_ids_to_be_upserted.append(record.id)
            if len(record_ids_to_be_upserted) > MAX_SCROLL_LIMIT:
                stop = True
                break
        if next_offset is None:
            stop = True
        else:
            offset = next_offset
    return record_ids_to_be_upserted

4. Remplacez le obtenir_collections méthode pour obtenir toutes les collections présentes dans la base de données

def get_collections(self):
    collections = self.qdrant_client.list_collections()
    return [collection.name for collection in collections]

5. Remplacez le supprimer_collection méthode pour supprimer la collection de la base de données

def delete_collection(self, collection_name: str):
    self.qdrant_client.delete_collection(collection_name=collection_name)

6. Remplacez le get_vector_store méthode pour obtenir le magasin vectoriel pour la collection donnée

def get_vector_store(self, collection_name: str, embeddings: Embeddings):
    return Qdrant(
            client=self.qdrant_client,
            embeddings=embeddings,
            collection_name=collection_name,
        )

7. Remplacez le obtenir_vector_client méthode pour obtenir le client vectoriel pour la collection donnée, le cas échéant

def get_vector_client(self):
    return self.qdrant_client

8. Remplacez le liste_point_de_données_vecteurs méthode pour répertorier les vecteurs déjà présents dans la base de données qui sont similaires aux documents en cours d'insertion

def list_data_point_vectors(
    self, collection_name: str,
    data_source_fqn: str,
    batch_size: int
) -> List[DataPointVector]:

    stop = False
    offset = None
    data_point_vectors: List[DataPointVector] = []

    # Get the vectors that match with the doc unique identifier
    # Data point is an object that has vector_id, fqn and hash
    # hash is calculated during parsing and chunking phase
    while stop is not True:
        records, next_offset = self.qdrant_client.scroll(
            collection_name=collection_name,
            limit=batch_size,
            with_payload=[
                f"metadata.{DATA_POINT_FQN_METADATA_KEY}",
                f"metadata.{DATA_POINT_HASH_METADATA_KEY}",
            ],
            # filter the records based on the data_point_fqn
            scroll_filter=models.Filter(
                should=[
                    models.FieldCondition(
                        key=f"metadata.{DATA_POINT_FQN_METADATA_KEY}",
                        match=models.MatchText(
                            text=data_source_fqn,
                        ),
                    ),
                ]
            ),
            with_vectors=False,
            offset=offset,
        )
        for record in records:
            metadata: dict = record.payload.get("metadata")
            if (
                metadata
                and metadata.get(DATA_POINT_FQN_METADATA_KEY)
                and metadata.get(DATA_POINT_HASH_METADATA_KEY)
            ):
                # Create a DataPointVector object
                data_point_vectors.append(
                    DataPointVector(
                        data_point_vector_id=record.id,
                        data_point_fqn=metadata.get(DATA_POINT_FQN_METADATA_KEY),
                        data_point_hash=metadata.get(DATA_POINT_HASH_METADATA_KEY),
                    )
                )
            if len(data_point_vectors) > MAX_SCROLL_LIMIT:
                stop = True
                break
        if next_offset is None:
            stop = True
        else:
            offset = next_offset
    return data_point_vectors

9. Remplacez le supprime_données_point_vecteurs méthode pour supprimer les vecteurs de la base de données, utilisée pour supprimer les anciens vecteurs du document mis à jour

def delete_data_point_vectors(
    self,
    collection_name: str,
    data_point_vectors: List[DataPointVector],
    batch_size: int = BATCH_SIZE,
):

    vectors_to_be_deleted_count = len(data_point_vectors)
    deleted_vectors_count = 0

    # delete the vectors in batches
    for i in range(0, vectors_to_be_deleted_count, batch_size):
        data_point_vectors_to_be_processed = data_point_vectors[i : i + batch_size]
        # Delete the vectors from the db
        self.qdrant_client.delete(
            collection_name=collection_name,
            points_selector=models.PointIdsList(
                points=[
                    document_vector_point.data_point_vector_id
                    for document_vector_point in data_point_vectors_to_be_processed
                ],
            ),
        )
        # Increment the deleted vectors count
        deleted_vectors_count = deleted_vectors_count + len(data_point_vectors_to_be_processed)

Intégration à SingleStore

SingleStore propose de puissantes fonctionnalités de base de données vectorielles parfaitement adaptées aux applications basées sur l'IA, aux chatbots, à la reconnaissance d'images, etc., vous évitant ainsi d'avoir à gérer une base de données vectorielle spécialisée uniquement pour vos charges de travail vectorielles. Contrairement aux bases de données vectorielles traditionnelles, SingleStore stocke les données vectorielles dans des tables relationnelles avec d'autres types de données. La co-localisation de données vectorielles avec des données associées vous permet d'interroger facilement des métadonnées étendues et d'autres attributs de vos données vectorielles avec toute la puissance de SQL.

SingleStore propose un niveau gratuit permettant aux développeurs de démarrer avec leur base de données vectorielles. Vous pouvez créer un compte gratuitement ici. Lors de votre inscription, rendez-vous sur Nuage -> espace de travail -> Créer un utilisateur. Utilisez les informations d'identification pour vous connecter à l'instance SingleStore.

Dans le fichier .env, vous pouvez ajouter les éléments suivants

VECTOR_DB_CONFIG = '{"url » : "<url_here>«, « provider » : « singlestore"}' # url : mysql ://{utilisateur} : {mot de passe} @ {hôte} : {port}/{db}

Pour ajouter la base de données vectorielles SingleStore au système RAG, procédez comme suit :

1. Nous voulons ajouter des colonnes supplémentaires au tableau pour stocker l'identifiant du vecteur. Par conséquent, nous remplaçons SingleStoreDB.

from langchain_community.vectorstores.singlestoredb import SingleStoreDB
import singlestoredb as s2

class SSDB(SingleStoreDB):
    def _create_table(self: SingleStoreDB) -> None:
        """Create table if it doesn't exist."""
        conn = self.connection_pool.connect()
        try:
            cur = conn.cursor()
            # Overriding the default table creation behaviour this adds id as autoinc primary key
            try:
                if self.use_vector_index:
                    index_options = ""
                    if self.vector_index_options and len(self.vector_index_options) > 0:
                        index_options = "INDEX_OPTIONS '{}'".format(
                            json.dumps(self.vector_index_options)
                        )
                    cur.execute(
                        """CREATE TABLE IF NOT EXISTS {}
                        (id BIGINT AUTO_INCREMENT PRIMARY KEY, {} TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci,
                        {} VECTOR({}, F32) NOT NULL, {} JSON,
                        VECTOR INDEX {} ({}) {});""".format(
                            self.table_name,
                            self.content_field,
                            self.vector_field,
                            self.vector_size,
                            self.metadata_field,
                            self.vector_index_name,
                            self.vector_field,
                            index_options,
                        ),
                    )
                else:
                    cur.execute(
                        """CREATE TABLE IF NOT EXISTS {}
                        (id BIGINT AUTO_INCREMENT PRIMARY KEY, {} TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci,
                        {} BLOB, {} JSON);""".format(
                            self.table_name,
                            self.content_field,
                            self.vector_field,
                            self.metadata_field,
                        ),
                    )
            finally:
                cur.close()
        finally:
            conn.close()


    def add_texts(
        self,
        texts: Iterable[str],
        metadatas: Optional[List[dict]] = None,
        embeddings: Optional[List[List[float]]] = None,
        **kwargs: Any,
    ) -> List[str]:
        """Add more texts to the vectorstore.

        Args:
            texts (Iterable[str]): Iterable of strings/text to add to the vectorstore.
            metadatas (Optional[List[dict]], optional): Optional list of metadatas.
                Defaults to None.
            embeddings (Optional[List[List[float]]], optional): Optional pre-generated
                embeddings. Defaults to None.

        Returns:
            List[str]: empty list
        """
        conn = self.connection_pool.connect()
        try:
            cur = conn.cursor()
            try:
                # Write data to singlestore db
                for i, text in enumerate(texts):
                    # Use provided values by default or fallback
                    metadata = metadatas[i] if metadatas else {}
                    embedding = (
                        embeddings[i]
                        if embeddings
                        else self.embedding.embed_documents([text])[0]
                    )
                    # Overriding insert statement to handle autoincrement id
                    cur.execute(
                        "INSERT INTO {} (content, vector, metadata) VALUES (%s, JSON_ARRAY_PACK(%s), %s)".format(
                            self.table_name
                        ),
                        (
                            text,
                            "[{}]".format(",".join(map(str, embedding))),
                            json.dumps(metadata),
                        ),
                    )
                if self.use_vector_index:
                    cur.execute("OPTIMIZE TABLE {} FLUSH;".format(self.table_name))
            finally:
                cur.close()
        finally:
            conn.close()
        return []

2. Créer une nouvelle classe Vector DB à magasin unique dans backend/modules/vector_db/singlestore.py qui hérite de Vecteur de base DB et initialisez-le avec Configuration vectorielle DBConfig

class SingleStoreVectorDB(BaseVectorDB):
    def __init__(self, config: VectorDBConfig):
        # url: mysql://{user}:{password}@{host}:{port}/{db}
        self.host = config.url

3. Remplacez le créer_collection méthode pour créer une collection dans SingleStore

def create_collection(self, collection_name: str, embeddings: Embeddings):
    # Calculate embedding size
    partial_embeddings = embeddings.embed_documents(["Initial document"])
    vector_size = len(partial_embeddings[0])

    # Create collection
    # we keep vector_filed, content_field as default values
    SSDB(
        embedding=embeddings,
        host=self.host,
        table_name=collection_name,
        vector_size=vector_size,
        use_vector_index=True,
    )

4. Remplacez le documents bouleversés méthode pour insérer les documents dans la base de données

def upsert_documents(
    self,
    collection_name: str,
    documents: List[Document],
    embeddings: Embeddings,
    incremental: bool = True,
):
    # Get the data point fqns from the documents
    data_point_fqns = []
    for document in documents:
        if document.metadata.get(DATA_POINT_FQN_METADATA_KEY):
            data_point_fqns.append(
                document.metadata.get(DATA_POINT_FQN_METADATA_KEY)
            )

    try:
        # Add documents to the table
        SSDB.from_documents(
            embedding=embeddings,
            documents=documents,
            table_name=collection_name,
            host=self.host,
        )
        logger.debug(
            f"[SingleStore] Added {len(documents)} documents to collection {collection_name}"
        )
    except Exception as e:
        logger.error(
            f"[SingleStore] Failed to add documents to collection {collection_name}: {e}"
        )

5. Remplacez le obtenir_collections méthode pour obtenir toutes les collections présentes dans la base de données

def get_collections(self) -> List[str]:
    # Create connection to SingleStore
    conn = s2.connect(self.host)
    try:
        cur = conn.cursor()
        try:
            # Get all the tables in the db
            cur.execute("SHOW TABLES")
            return [row[0] for row in cur.fetchall()]
        except Exception as e:
            logger.error(
                f"[SingleStore] Failed to get collections: {e}"
            )
        finally:
            cur.close()
    except Exception as e:
        logger.error(
            f"[SingleStore] Failed to get collections: {e}"
        )
    finally:
        conn.close()

6. Remplacez le supprimer_collection méthode pour supprimer la collection de la base de données

def delete_collection(self, collection_name: str):
    # Create connection to SingleStore
    conn = s2.connect(self.host)
    try:
        cur = conn.cursor()
        try:
            # Drop the table
            cur.execute(f"DROP TABLE {collection_name}")
            logger.debug(f"[SingleStore] Deleted collection {collection_name}")
        except Exception as e:
            logger.error(
                f"[SingleStore] Failed to delete collection {collection_name}: {e}"
            )
        finally:
            cur.close()
    except Exception as e:
        logger.error(
            f"[SingleStore] Failed to delete collection {collection_name}: {e}"
        )
    finally:
        conn.close()

7. Remplacez le get_vector_store méthode pour obtenir le magasin vectoriel pour la collection donnée

def get_vector_store(self, collection_name: str, embeddings: Embeddings):
    return SSDB(
        embedding=embeddings,
        host=self.host,
        table_name=collection_name,
    )

8. Remplacez le obtenir_vector_client méthode pour obtenir le client vectoriel pour la collection donnée, le cas échéant

def get_vector_client(self):
    return s2.connect(self.host)

9. Remplacez le liste_point_de_données_vecteurs méthode pour répertorier les vecteurs déjà présents dans la base de données qui sont similaires aux documents en cours d'insertion

def list_data_point_vectors(
    self,
    collection_name: str,
    data_source_fqn: str,
) -> List[DataPointVector]:
    data_point_vectors: List[DataPointVector] = []
    logger.debug(f"data_source_fqn: {data_source_fqn}")

    # Get all data point vectors from table upto MAX_SCROLL_LIMIT
    conn = s2.connect(self.host)
    try:
        curr = conn.cursor()

        # Remove all data point vectors with the same data_source_fqn
        curr.execute(f"SELECT * FROM {collection_name} WHERE JSON_EXTRACT_JSON(metadata, '{DATA_POINT_FQN_METADATA_KEY}') LIKE '%{data_source_fqn}%' LIMIT {MAX_SCROLL_LIMIT}")

        for record in curr:
            # id, content, vector, metadata
            id, _, _, metadata = record
            if (
                metadata
                and metadata.get(DATA_POINT_FQN_METADATA_KEY)
                and metadata.get(DATA_POINT_HASH_METADATA_KEY)
            ):
                # Create a DataPointVector object
                data_point_vectors.append(
                    DataPointVector(
                        data_point_vector_id=id,
                        data_point_fqn=metadata.get(DATA_POINT_FQN_METADATA_KEY),
                        data_point_hash=metadata.get(DATA_POINT_HASH_METADATA_KEY),
                    )
                )
    except Exception as e:
        logger.error(
            f"[SingleStore] Failed to list data point vectors: {e}"
        )
    finally:
        conn.close()

    logger.debug(
        f"[SingleStore] Listing {len(data_point_vectors)} data point vectors for collection {collection_name}"
    )
    return data_point_vectors

10. Remplacez le supprime_données_point_vecteurs méthode pour supprimer les vecteurs de la base de données, utilisée pour supprimer les anciens vecteurs du document mis à jour

def delete_data_point_vectors(
    self,
    collection_name: str,
    data_point_vectors: List[DataPointVector],
    batch_size: int =BATCH_SIZE,
):
    if len(data_point_vectors) > 0:
        # Create connection to SingleStore
        conn = s2.connect(self.host)

        try:
            vectors_to_be_deleted_count = len(data_point_vectors)
            curr = conn.cursor()

            # Delete the vectors as per the data_point_vector_id
            curr.execute(
                f"DELETE FROM {collection_name} WHERE id in ({', '.join(data_point_vector.data_point_vector_id for data_point_vector in data_point_vectors)})"
            )
            logger.debug(
                f"[SingleStore] Deleted {vectors_to_be_deleted_count} data point vectors"
            )
        except Exception as e:
            logger.error(
                f"[SingleStore] Failed to delete data point vectors: {e}"
            )
        finally:
            conn.close()

Magasin de métadonnées

Il contient les configurations nécessaires qui définissent de manière unique un projet ou une application RAG. Une application RAG peut contenir un ensemble de documents provenant d'une ou de plusieurs sources de données combinées, que nous appelons collection. Les documents provenant de ces sources de données sont indexés dans la base de données vectorielle à l'aide de méthodes de chargement, d'analyse et d'intégration des données. Pour chaque cas d'utilisation de RAG, le magasin de métadonnées contient :

• Nom de la collection
• Nom de la base de données vectorielle associée utilisée
• Sources de données liées
• Configuration de l'analyse pour chaque source de données
• Modèle d'intégration et configuration à utiliser

Nous définissons actuellement deux manières de stocker ces données, l'une localement et autres utilisations Truefoudry. Ces magasins sont définis comme suit : backend/modules/metada_store/

Contrôleurs de requêtes

Une fois les données indexées et stockées dans la base de données vectorielles, il est maintenant temps de combiner toutes les parties pour utiliser notre application. C'est exactement ce que font les contrôleurs de requêtes ! Ils nous aident à trouver la réponse à la requête de l'utilisateur correspondante. Les étapes typiques d'un contrôleur de requêtes sont les suivantes :

1. Les utilisateurs envoient une charge utile de requête contenant la requête, le nom de la collection, la configuration llm, l'invite, le retriever et sa configuration.
2. Sur la base de nom de la collection pertinent base de données vectorielle est corrigé avec sa configuration, comme l'intégrateur utilisé, le type de base de données vectorielle, etc.
3. Sur la base de requête, les documents pertinents sont récupérés à l'aide du retriever à partir de Vector DB.
4. Les documents récupérés proviennent du contexte et avec la requête a.k.a question est donné au LLM pour générer la réponse. Cette étape peut également impliquer un réglage rapide.
5. Si nécessaire, avec la réponse générée, des segments de document pertinents sont également renvoyés dans la réponse.

Remarque : Dans le cas des agents, les étapes intermédiaires peuvent également être diffusées en continu. C'est à l'application concernée de décider.

Les méthodes des contrôleurs de requêtes peuvent être directement exposées en tant qu'API, en ajoutant des décorateurs http aux fonctions respectives.

Pour ajouter votre propre contrôleur de requêtes, procédez comme suit :

1. Chaque cas d'utilisation de RAG doit généralement disposer d'un dossier distinct contenant le contrôleur de requêtes. Supposons que notre dossier est application-2. Par conséquent, nous allons écrire notre contrôleur sous /backend/modules/query_controller/app-2/controller.py
2. Ajouter contrôleur de requêtes décorateur à votre classe et transmettez le nom de votre contrôleur personnalisé en argument
3. Ajoutez des méthodes à ce contrôleur selon vos besoins et utilisez nos décorateurs http comme poste, obtenir, supprimer pour transformer vos méthodes en API

from backend.server.decorator import post
@query_controller("/app-2")
class MyCustomController():
    ...
    @post("/answer")
    def answer(query: str):
        # Write code to express your logic for answer
        # This API will be exposed as POST /my-controller/answer
        ...

4. Importez votre classe de contrôleur personnalisée sur backend/modules/query_controllers/__init__.py

from backend.modules.query_controllers.sample_controller.controller import MyCustomController

Un exemple de contrôleur de requêtes est écrit sous la forme suivante : /backend/modules/query_controller/example/controller.py Veuillez vous référer pour une meilleure compréhension

Cognita - Flux de processus

Un processus Cognita typique comprend deux phases :

1. Indexation des données
2. Génération de réponses

Indexation des données

Cette phase implique le chargement de données à partir de sources, l'analyse des documents présents dans ces sources et leur indexation dans la base de données vectorielle. Pour traiter de grandes quantités de documents rencontrés en production, Cognita va encore plus loin.

1. Cognita regroupe les documents par lots, au lieu de les indexer tous ensemble.
2. Cognita gère les calculs et assure le suivi du hachage des documents. Ainsi, chaque fois qu'un nouveau document est ajouté à la source de données à des fins d'indexation, seuls ces documents sont indexés au lieu d'indexer la source de données complète, ce qui permet de gagner du temps et de réaliser des calculs considérables.
3. Ce mode d'indexation est également appelé INCRÉMENTIEL indexation, il existe également un autre mode pris en charge dans Cognita, à savoir COMPLET indexation. COMPLET l'indexation réingère les données dans la base de données vectorielles indépendamment des données vectorielles présentes pour la collection donnée.

Génération de réponses

La phase de génération de réponse lance un appel au /réponse point de terminaison de votre définition Contrôleur de requêtes et génère la réponse à la requête demandée.

Utilisation de l'interface utilisateur Cognita

Les étapes suivantes vous montreront comment utiliser l'interface utilisateur de Cognita pour interroger des documents :

1. Créer une source de données

• Cliquez sur Sources de données onglet

• Cliquez + Nouvelle source de données
• Le type de source de données peut être un fichier provenant d'un répertoire local, une URL Web, une URL github ou fournissant le FQN de l'artefact Truefoundry.
  • Par exemple : Si Localdir est sélectionné, téléchargez des fichiers depuis votre machine et cliquez sur Soumettre.
• La liste des sources de données créées sera disponible dans l'onglet Sources de données.

2. Créer une collection

• Cliquez sur Collections onglet
• Cliquez + Nouvelle collection

• Entrez le nom de la collection
• Sélectionnez le modèle d'intégration
• Ajoutez la source de données créée précédemment et la configuration nécessaire
• Cliquez Procédé pour créer la collection et indexer les données.

3. Dès que vous créez la collection, l'ingestion des données commence. Vous pouvez consulter son statut en sélectionnant votre collection dans l'onglet Collections. Vous pouvez également ajouter des sources de données supplémentaires ultérieurement et les indexer dans la collection.

4. Génération de réponses

• Sélectionnez la collection
• Sélectionnez le LLM et sa configuration
• Sélectionnez le récupérateur de documents
• Rédigez l'invite ou utilisez l'invite par défaut
• Posez la question

Commencez dès maintenant !

Réservez un pDémo personnalisée ou inscrivez-vous dès aujourd'hui pour commencer à créer vos cas d'utilisation RAG.