Cognita: Modulaire open-source RAG-applicaties bouwen voor productie

Stel dat team A is aangewezen om een RAG-applicatie te ontwikkelen voor use-case-1, dat team B een RAG-applicatie ontwikkelt voor use-case-2, en dat team C net bezig is met de planning van hun komende RAG-applicatie. Heeft u ooit gewenst dat het bouwen van RAG-pipelines voor meerdere teams eenvoudiger was? Dat niet elk team vanaf nul hoeft te beginnen, maar dat er een modulaire manier is waarop elk team dezelfde basisfunctionaliteit kan gebruiken en hier effectief eigen apps bovenop kan bouwen zonder elkaar te hinderen?

Geen zorgen! Dit is precies waarom we Cognita hebben gemaakt. Hoewel RAG onmiskenbaar indrukwekkend is, kan het proces om er een functionele applicatie mee te maken uitdagend zijn. Er komt heel wat kijken bij de implementatie en best practices voor ontwikkeling, variërend van het selecteren van de juiste AI-modellen voor de specifieke use-case tot het effectief organiseren van data om de gewenste inzichten te verkrijgen. Hoewel tools zoals LangChain en LlamaIndex bestaan om het ontwerpen van prototypes te vereenvoudigen, was er nog geen toegankelijk, gebruiksklaar open-source RAG-template dat best practices bevat en modulaire ondersteuning biedt, zodat iedereen het snel en eenvoudig kan gebruiken.

Voordelen van Cognita:

1. Een centrale, herbruikbare repository van parsers, loaders, embedders en retrievers.
2. Mogelijkheid voor niet-technische gebruikers om met de UI te werken - documenten uploaden en Q&A uitvoeren met behulp van modules die door het ontwikkelteam zijn gebouwd.
3. Volledig API-gestuurd - wat integratie met andere systemen mogelijk maakt.

Overzicht

Bij het ontwikkelen van de interne werking van Cognita was ons doel om een balans te vinden tussen volledige aanpasbaarheid en flexibiliteit, terwijl we direct uit de doos gebruiksvriendelijkheid garanderen. Gezien het snelle tempo van ontwikkelingen in RAG en AI, was het voor ons essentieel om Cognita te ontwerpen met schaalbaarheid in het achterhoofd. Dit maakt een naadloze integratie van nieuwe doorbraken en uiteenlopende use-cases mogelijk. Hierdoor hebben we het RAG-proces opgedeeld in afzonderlijke modulaire stappen (zoals getoond in het bovenstaande diagram, te bespreken in de volgende secties). Dit vergemakkelijkt het onderhoud van het systeem, de toevoeging van nieuwe functionaliteiten zoals interoperabiliteit met andere AI-bibliotheken, en stelt gebruikers in staat het platform af te stemmen op hun specifieke vereisten. Onze focus blijft gericht op het bieden van een robuuste tool die niet alleen aan de huidige behoeften van gebruikers voldoet, maar ook mee-evolueert met de technologie, inclusief bredere architecturale verschuivingen zoals MCP vs RAG, wat langetermijnwaarde garandeert.

Componenten

Cognita is ontworpen rond zeven verschillende modules, die elk aanpasbaar en bestuurbaar zijn om aan verschillende behoeften te voldoen:

1. Data Loaders
2. Parsers
3. Embedders
4. Rerankers
5. Vector DB's
6. Metadata Store
7. Query Controllers

Data Loaders

Deze laden de data uit verschillende bronnen zoals lokale mappen, S3-buckets, databases, Truefoundry-artifacts, enz. Cognita ondersteunt momenteel het laden van data uit een lokale map, web-URL, GitHub-repository en Truefoundry-artifacts. Meer data loaders kunnen eenvoudig worden toegevoegd onder backend/modules/dataloaders/ . Zodra een dataloader is toegevoegd, moet u deze registreren zodat deze door de RAG-applicatie kan worden gebruikt onder backend/modules/dataloaders/__init__.py Voeg het volgende toe om een dataloader te registreren:

register_dataloader("MyDataLoader", MyDataLoaderClass)

Parsers

In deze stap verwerken we verschillende soorten data, zoals reguliere tekstbestanden, PDF's en zelfs Markdown-bestanden. Het doel is om al deze verschillende typen om te zetten naar één gemeenschappelijk formaat, zodat we er later gemakkelijker mee kunnen werken. Dit onderdeel, parsing genoemd, duurt meestal het langst en is lastig te implementeren wanneer we een dergelijk systeem opzetten. Het gebruik van Cognita helpt hierbij, omdat het het zware werk van het beheren van datapipelines al voor ons afhandelt.

Hierna splitsen we de geparsede data op in uniforme chunks. Maar waarom is dit nodig? De tekst die we uit de bestanden halen, kan verschillende lengtes hebben. Als we deze lange teksten direct gebruiken, voegen we uiteindelijk een hoop onnodige informatie toe. Omdat alle LLM's bovendien slechts een beperkte hoeveelheid tekst tegelijk kunnen verwerken, kunnen we niet alle belangrijke context opnemen die nodig is voor de vraag. Daarom splitsen we de tekst op in kleinere delen voor elke sectie. Intuitief zullen kleinere chunks relevantere concepten bevatten en minder ruis veroorzaken in vergelijking met grotere chunks.

Momenteel ondersteunen we parsing voor Markdown-, PDF - en Text -bestanden. Meer data parsers kunnen eenvoudig worden toegevoegd onder backend/modules/parsers/ . Zodra een parser is toegevoegd, moet u deze registreren zodat deze door de RAG-applicatie kan worden gebruikt onder backend/modules/parsers/__init__.py Voeg het volgende toe om een parser te registreren:

register_parser("MyParser", MyParserClass)

Embedders

Zodra we de data in kleinere stukken hebben opgedeeld, willen we de belangrijkste chunks voor een specifieke vraag vinden. Een snelle en effectieve manier om dit te doen is door een voorgeconditioneerd model (embedding-model) te gebruiken om onze data en de vraag om te zetten in speciale codes genaamd embeddings. Vervolgens vergelijken we de embeddings van elke chunk data met die van de vraag. Door de cosine similarity tussen deze embeddings te meten, kunnen we bepalen welke chunks het meest verwant zijn aan de vraag, wat ons helpt de beste te selecteren voor gebruik.

Er zijn veel voorgeconditioneerde modellen beschikbaar om de data te embedden, zoals modellen van OpenAI, Cohere, enz. De populairste kunnen worden ontdekt via HuggingFace's Massive Text Embedding Benchmark (MTEB) leaderboard. We bieden ondersteuning voor OpenAI Embeddings, TrueFoundry Embeddings en ook de huidige SOTA embeddings (per april 2024) van mixedbread-ai.

Meer embedders kunnen eenvoudig worden toegevoegd onder backend/modules/embedder/ . Zodra een embedder is toegevoegd, moet u deze registreren zodat deze door de RAG-applicatie kan worden gebruikt onder backend/modules/embedders/__init__.py Voeg het volgende toe om een parser te registreren:

register_embedder("MyEmbedder", MyEmbedderClass)

‍Opmerking: Onthoud dat embeddings niet de enige methode zijn om belangrijke chunks te vinden. We zouden hier ook een LLM voor kunnen gebruiken! LLM's zijn echter veel groter dan embedding-modellen en hebben een limiet op de hoeveelheid tekst die ze tegelijk kunnen verwerken. Daarom is het slimmer om eerst embeddings te gebruiken om de top k chunks te selecteren. Vervolgens kunnen we LLM's toepassen op deze minder talrijke chunks om te bepalen welke het beste als context gebruikt kunnen worden om onze vraag te beantwoorden.

Rerankers

Zodra de embedding-stap potentiële matches vindt (wat er veel kunnen zijn), wordt er een reranking-stap toegepast. Reranking zorgt ervoor dat de beste resultaten bovenaan staan. Hierdoor kunnen we de top x documenten kiezen, wat onze context beknopter maakt en de prompt-query korter. We bieden ondersteuning voor de SOTA reranker (per april 2024) van mixedbread-ai die is geïmplementeerd onder backend/modules/reranker/

VectorDB

Zodra we vectoren voor teksten hebben gemaakt, slaan we ze op in een zogenaamde vector database. Deze database houdt deze vectoren bij, zodat we ze later snel kunnen terugvinden via verschillende methoden. Reguliere databases organiseren data in tabellen met rijen en kolommen, maar vector databases zijn speciaal omdat ze data opslaan en zoeken op basis van deze vectoren. Dit is uitermate handig voor zaken zoals afbeeldingsherkenning, taalbegrip of aanbevelingssystemen. In een aanbevelingssysteem wordt elk item dat u wilt aanbevelen (zoals een film of een product) bijvoorbeeld omgezet in een vector, waarbij verschillende onderdelen van de vector verschillende kenmerken van het item vertegenwoordigen, zoals het genre of de prijs. Evenzo wordt bij taaltoepassingen elk woord of document omgezet in een vector, waarbij delen van de vector kenmerken van het woord of document vertegenwoordigen, zoals hoe vaak het woord wordt gebruikt of wat het betekent. Deze vector databases zijn ontworpen om dit efficiënt af te handelen. Door verschillende manieren te gebruiken om te meten hoe dicht vectoren bij elkaar liggen, zoals hun gelijkenis of afstand, vinden we de vectoren die het dichtst bij de gegeven gebruikersquery liggen. De meest gebruikelijke meetmethoden zijn Euclidean Distance, Cosine Similarity en Dot Product.

Er zijn diverse vector databases beschikbaar op de markt, zoals Qdrant, SingleStore, Weaviate, enz. We ondersteunen momenteel Qdrant en SingleStore. De Qdrant vector db-klasse is gedefinieerd onder /backend/modules/vector_db/qdrant.py, terwijl de SingleStore vector db-klasse is gedefinieerd onder /backend/modules/vector_db/singlestore.py

Andere vector db's kunnen ook worden toegevoegd in de map vector_db en kunnen worden geregistreerd onder /backend/modules/vector_db/__init__.py

Om ondersteuning voor een vector DB toe te voegen in Cognita, moet de gebruiker het volgende doen:

• Maak de klasse aan die overerft van BaseVectorDB (from backend.modules.vector_db.base import BaseVectorDB) en initialiseer deze met VectorDBConfig (from backend.types import VectorDBConfig)
• Implementeer de volgende methoden:
  • create_collection: Om de collectie/het project/de tabel te initialiseren in de vector db.
  • upsert_documents: Om de documenten in te voegen in de db.
  • get_collections: Haal alle collecties op die aanwezig zijn in de db.
  • delete_collection: Om de collectie te verwijderen uit de db.
  • get_vector_store: Om de vector store op te halen voor de gegeven collectie.
  • get_vector_client: Om de vector client op te halen voor de gegeven collectie, indien aanwezig.
  • list_data_point_vectors: Om reeds aanwezige vectoren in de db op te sommen die vergelijkbaar zijn met de documenten die worden ingevoegd.
  • delete_data_point_vectors: Om de vectoren uit de db te verwijderen, gebruikt om oude vectoren van het bijgewerkte document te verwijderen.

We laten nu zien hoe we een nieuwe vector db kunnen toevoegen aan het RAG-systeem. We nemen het voorbeeld van zowel Qdrant - als SingleStore -vector db's.

Qdrant Integratie

Qdrant is een Open-Source Vector Database en Vector Search Engine geschreven in Rust. Het biedt een snelle en schaalbare vector similarity search dienst met een handige API. Volg onderstaande stappen om de Qdrant vector db toe te voegen aan het RAG-systeem:

In het .env-bestand kunt u het volgende toevoegen

VECTOR_DB_CONFIG = '{"url": "<url_here>", "provider": "qdrant"}' # Qdrant URL for deployed instance

VECTOR_DB_CONFIG='{"provider":"qdrant","local":"true"}' # For local file based Qdrant instance without docker

1. Maak een nieuwe klasse QdrantVectorDB aan in backend/modules/vector_db/qdrant.py die overerft van BaseVectorDB en initialiseer deze met VectorDBConfig

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. Overschrijf de methode create_collection om een collectie aan te maken in 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. Overschrijf de methode upsert_documents om de documenten in de db in te voegen

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. Overschrijf de methode get_collections om alle collecties op te halen die aanwezig zijn in de db

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

5. Overschrijf de methode delete_collection om de collectie te verwijderen uit de db

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

6. Overschrijf de methode get_vector_store om de vector store op te halen voor de gegeven collectie

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

7. Overschrijf de methode get_vector_client om de vector client op te halen voor de gegeven collectie, indien aanwezig

def get_vector_client(self):
    return self.qdrant_client

8. Overschrijf de methode list_data_point_vectors om al aanwezige vectoren in de db op te sommen die vergelijkbaar zijn met de documenten die worden ingevoegd

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. Overschrijf de methode delete_data_point_vectors om de vectoren uit de db te verwijderen, gebruikt om oude vectoren van het bijgewerkte document te verwijderen

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)

SingleStore Integratie

SingleStore biedt krachtige vector database-functionaliteit die uitermate geschikt is voor op AI gebaseerde applicaties, chatbots, afbeeldingsherkenning en meer, waardoor het niet meer nodig is om een gespecialiseerde vector database te draaien puur voor uw vector-workloads. In tegenstelling tot traditionele vector databases slaat SingleStore vectordata op in relationele tabellen naast andere typen data. Door vectordata samen te brengen met gerelateerde data kunt u eenvoudig uitgebreide metadata en andere attributen van uw vectordata bevragen met de volle kracht van SQL.

SingleStore biedt een gratis tier voor ontwikkelaars om aan de slag te gaan met hun vector database. U kunt zich hier aanmelden voor een gratis account. Ga na het aanmelden naar Cloud -> workspace -> Create User. Gebruik de inloggegevens om verbinding te maken met de SingleStore-instantie.

In het .env-bestand kunt u het volgende toevoegen

VECTOR_DB_CONFIG = '{"url": "<url_hier>", "provider": "singlestore"}' # url: mysql://{user}:{password}@{host}:{port}/{db}

Volg de onderstaande stappen om de SingleStore vector db toe te voegen aan het RAG-systeem:

1. We willen extra kolommen aan de tabel toevoegen om het vector-ID op te slaan, daarom overschrijven we 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. Maak een nieuwe klasse SingleStoreVectorDB aan in backend/modules/vector_db/singlestore.py die overerft van BaseVectorDB en initialiseer deze met VectorDBConfig

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

3. Overschrijf de methode create_collection om een collectie aan te maken in 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. Overschrijf de methode upsert_documents om de documenten in te voegen in de db

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. Overschrijf de methode get_collections om alle collecties op te halen die aanwezig zijn in de db

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. Overschrijf de methode delete_collection om de collectie te verwijderen uit de db

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. Overschrijf de methode get_vector_store om de vector store op te halen voor de gegeven collectie

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

8. Overschrijf de methode get_vector_client om de vector client op te halen voor de gegeven collectie, indien aanwezig

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

9. Overschrijf de methode list_data_point_vectors om al aanwezige vectoren in de db op te sommen die vergelijkbaar zijn met de documenten die worden ingevoegd

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. Overschrijf de methode delete_data_point_vectors om de vectoren uit de db te verwijderen, gebruikt om oude vectoren van het bijgewerkte document te verwijderen

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()

Metadata Store

Dit bevat de noodzakelijke configuraties die een project of RAG-app uniek definiëren. Een RAG-app kan een verzameling documenten bevatten uit een of meer gecombineerde databronnen, wat we aanduiden als collection. De documenten uit deze databronnen worden geïndexeerd in de vector db met behulp van methoden voor data loading + parsing + embedding. Voor elke RAG-use-case bevat de metadata store:

• Naam van de collectie
• Naam van de geassocieerde gebruikte Vector DB
• Gekoppelde databronnen
• Parsing-configuratie voor elke databron
• Te gebruiken embedding-model en de configuratie daarvan

We definiëren momenteel twee manieren om deze data op te slaan: één locally en de andere met behulp van Truefoudry. Deze stores zijn gedefinieerd onder - backend/modules/metada_store/

Query Controllers

Zodra de data is geïndexeerd en opgeslagen in de vector db, is het tijd om alle onderdelen samen te voegen om onze app te gebruiken. Query Controllers doen precies dat! Ze helpen ons het antwoord op te halen voor de bijbehorende gebruikersquery. De stappen van een typische query controller zijn als volgt:

1. De gebruiker stuurt een request-payload die de query, collectienaam, LLM-configuratie, prompt, retriever en de configuratie daarvan bevat.
2. Op basis van de collection name wordt de relevante vector db opgehaald met de bijbehorende configuratie zoals het gebruikte embedding-model, vector db-type, enz.
3. Op basis van de query worden relevante documenten opgehaald uit de vector db met behulp van de retriever .
4. De opgehaalde documenten vormen de context en worden samen met de query (oftewel question) aan de LLM gegeven om het antwoord te genereren. Deze stap kan ook prompt tuning omvatten.
5. Indien nodig worden samen met het gegenereerde antwoord ook relevante document-chunks geretourneerd in de response.

Opmerking: In het geval van agents kunnen de tussenstappen ook gestreamd worden. Het is aan de specifieke app om dit te bepalen.

Query controller-methoden kunnen direct worden blootgesteld als API door HTTP-decorators toe te voegen aan de respectievelijke functies.

Voer de volgende stappen uit om uw eigen query controller toe te voegen:

1. Elke RAG-use-case moet doorgaans een eigen map hebben die de query controller bevat. Stel dat onze map app-2 is. We schrijven onze controller dan onder /backend/modules/query_controller/app-2/controller.py
2. Voeg de query_controller decorator toe aan uw klasse en geef de naam van uw aangepaste controller door als argument
3. Voeg methoden toe aan deze controller naar uw behoeften en gebruik onze HTTP-decorators zoals post, get, delete om van uw methoden een API te maken

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. Importeer uw aangepaste controller-klasse op backend/modules/query_controllers/__init__.py

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

Een voorbeeld van een query controller is geschreven onder: /backend/modules/query_controller/example/controller.py Raadpleeg dit voor een beter begrip

Cognita - Procesverloop

Een typisch Cognita-proces bestaat uit twee fasen:

1. Dataindexering
2. Antwoordgeneratie

Dataindexering

Deze fase omvat het laden van data uit bronnen, het parsen van de documenten in deze bronnen en het indexeren ervan in de vector db. Om grote hoeveelheden documenten in productieomgevingen te verwerken, gaat Cognita nog een stap verder.

1. Cognita groepeert documenten in batches, in plaats van ze allemaal tegelijk te indexeren.
2. Cognita berekent en houdt de hash van documenten bij, zodat wanneer een nieuw document wordt toegevoegd aan de databron voor indexering, alleen die documenten worden geïndexeerd in plaats van de volledige databron. Dit bespaart aanzienlijk veel tijd en rekenkracht.
3. Deze manier van indexeren wordt ook wel INCREMENTAL indexering genoemd. Er is ook een andere modus die in Cognita wordt ondersteund, namelijk FULL indexering. FULL indexering voert de data opnieuw in de vector db in, ongeacht de aanwezige vectordata voor de gegven collectie.

Antwoordgeneratie

De fase van antwoordgeneratie voert een call uit naar het /answer endpoint van uw gedefinieerde QueryController en genereert het antwoord voor de aangevraagde query.

De Cognita UI gebruiken

De volgende stappen laten zien hoe u de Cognita UI kunt gebruiken om documenten te bevragen:

1. Databron aanmaken

• Klik op het tabblad Data Sources

• Klik op + New Datasource
• Het type databron kan bestaan uit bestanden uit een lokale map, een web-URL, een GitHub-URL of het opgeven van een Truefoundry artifact FQN.
  • Bijv.: als Localdir is geselecteerd, uploadt u bestanden van uw apparaat en klikt u op Submit.
• De lijst met aangemaakte databronnen is beschikbaar in het tabblad Data Sources.

2. Collectie aanmaken

• Klik op het tabblad Collections
• Klik op + New Collection

• Voer de naam van de collectie in
• Selecteer het Embedding Model
• Voeg de eerder aangemaakte databron en de benodigde configuratie toe
• Klik op Process om de collectie aan te maken en de data te indexeren.

3. Zodra u de collectie aanmaakt, begint de data-ingestie. U kunt de status bekijken door uw collectie te selecteren in het tabblad Collections. U kunt later ook aanvullende databronnen toevoegen en deze in de collectie indexeren.

4. Antwoordgeneratie

• Selecteer de collectie
• Selecteer het LLM en de configuratie daarvan
• Selecteer de document retriever
• Schrijf de prompt of gebruik de standaard prompt
• Stel de query

Start vandaag nog!

Boek een gepersonaliseerde demo of meld u vandaag nog aan om te starten met het bouwen van uw RAG-use-cases.