Cognita: Entwicklung modularer Open-Source-RAG-Anwendungen für die Produktion

Angenommen, es gibt eine Mannschaft A beauftragt mit der Entwicklung einer RAG-Anwendung für Anwendungsfall-1, dann ist da Mannschaft B das entwickelt eine RAG-Anwendung für Anwendungsfall-2, und dann ist da Mannschaft C, das plant gerade für ihren bevorstehenden Anwendungsfall der RAG-Anwendung. Haben Sie sich gewünscht, dass der Aufbau von RAG-Pipelines für mehrere Teams einfach hätte sein sollen? Jedes Team muss nicht bei Null anfangen, sondern eine modulare Methode verwenden, bei der jedes Team die gleichen Basisfunktionen nutzen und darauf aufbauend effektiv eigene Apps entwickeln kann, ohne dass es zu Störungen kommt?

Mach dir keine Sorgen!! Deshalb haben wir geschaffen Cognita. RAG ist zwar unbestreitbar beeindruckend, aber der Prozess, damit eine funktionale Anwendung zu erstellen, kann entmutigend sein. In Bezug auf Implementierungs- und Entwicklungspraktiken gibt es eine Menge zu verstehen, angefangen bei der Auswahl der geeigneten KI-Modelle für den spezifischen Anwendungsfall bis hin zur effektiven Organisation von Daten, um die gewünschten Erkenntnisse zu gewinnen. Während Tools wie Lang-Kette und Llamaindex Um den Prototyp-Designprozess zu vereinfachen, gibt es noch keine zugängliche, gebrauchsfertige Open-Source-RAG-Vorlage, die bewährte Verfahren enthält und modulare Unterstützung bietet, sodass jeder sie schnell und einfach verwenden kann.

Vorteile von Cognita:

1. Ein zentrales wiederverwendbares Repository für Parser, Loader, Embedder und Retriever.
2. Möglichkeit für Benutzer ohne technische Kenntnisse, mit der Benutzeroberfläche zu spielen — Laden Sie Dokumente hoch und führen Sie QnA mithilfe von Modulen durch, die vom Entwicklungsteam erstellt wurden.
3. Vollständig API-gesteuert — was die Integration mit anderen Systemen ermöglicht.

Überblick

Eintauchen in das Innenleben von Cognita, unser Ziel war es, ein Gleichgewicht zwischen vollständiger Anpassung und Anpassungsfähigkeit zu finden und gleichzeitig die Benutzerfreundlichkeit von Anfang an zu gewährleisten. Angesichts der rasanten Fortschritte bei RAG und KI war es für uns unerlässlich, Cognita unter Berücksichtigung der Skalierbarkeit zu entwickeln, um eine nahtlose Integration neuer Durchbrüche und vielfältiger Anwendungsfälle zu ermöglichen. Dies veranlasste uns, den RAG-Prozess in verschiedene modulare Schritte zu unterteilen (wie in der obigen Abbildung dargestellt, die in den nachfolgenden Abschnitten erörtert wird), was eine einfachere Systemwartung ermöglichte, neue Funktionen wie die Interoperabilität mit anderen KI-Bibliotheken hinzufügte und es den Benutzern ermöglichte, die Plattform an ihre spezifischen Anforderungen anzupassen. Unser Fokus liegt weiterhin darauf, den Benutzern ein robustes Tool zur Verfügung zu stellen, das nicht nur ihren aktuellen Bedürfnissen entspricht, sondern sich auch parallel zur Technologie weiterentwickelt, einschließlich umfassenderer architektonischer Veränderungen wie MCP gegen RAG, um einen langfristigen Wert zu gewährleisten.

Komponenten

Cognita besteht aus sieben verschiedenen Modulen, die jeweils an unterschiedliche Bedürfnisse angepasst und gesteuert werden können:

1. Datenlader
2. Parser
3. Einbetter
4. Reranker
5. Vektor-DBs
6. Metadatenspeicher
7. Controller abfragen

Datenlader

Diese laden die Daten aus verschiedenen Quellen wie lokalen Verzeichnissen, S3-Buckets, Datenbanken, Echte Gießerei Artefakte usw. Cognita unterstützt derzeit das Laden von Daten aus einem lokalen Verzeichnis, einer Web-URL, einem Github-Repository und Truefoundry-Artefakten. Weitere Datenlader können einfach hinzugefügt werden unter Backend/Module/Datenlader/ . Sobald ein Dataloader hinzugefügt wurde, müssen Sie ihn registrieren, damit er von der RAG-Anwendung unter verwendet werden kann backend/modules/dataloaders/__init__.py Um einen Dataloader zu registrieren, fügen Sie Folgendes hinzu:

register_dataloader("MyDataLoader", MyDataLoaderClass)

Parser

In diesem Schritt befassen wir uns mit verschiedenen Arten von Daten, wie normalen Textdateien, PDFs und sogar Markdown-Dateien. Ziel ist es, all diese verschiedenen Typen in ein gemeinsames Format umzuwandeln, damit wir später einfacher mit ihnen arbeiten können. Dieser Teil, der als Parsen bezeichnet wird, dauert normalerweise am längsten und ist schwierig zu implementieren, wenn wir ein solches System einrichten. Aber die Verwendung von Cognita kann helfen, da es uns bereits die harte Arbeit der Verwaltung von Datenpipelines abnimmt.

Nachdem wir das gepostet haben, teilen wir die analysierten Daten in einheitliche Blöcke auf. Aber warum brauchen wir das? Der Text, den wir aus den Dateien erhalten, kann unterschiedlich lang sein. Wenn wir diese langen Texte direkt verwenden, fügen wir am Ende eine Menge unnötiger Informationen hinzu. Da alle LLMs außerdem nur eine bestimmte Textmenge gleichzeitig verarbeiten können, können wir nicht den gesamten wichtigen Kontext angeben, der für die Frage benötigt wird. Stattdessen werden wir den Text für jeden Abschnitt in kleinere Teile zerlegen. Intuitiv gesehen enthalten kleinere Abschnitte relevante Konzepte und sind im Vergleich zu größeren Abschnitten weniger laut.

Derzeit unterstützen wir das Parsen für Abschlag, PDF und Text Dateien. Weitere Datenparser können einfach hinzugefügt werden unter Backend/Module/Parser/ . Sobald ein Parser hinzugefügt wurde, müssen Sie ihn registrieren, damit er von der RAG-Anwendung unter verwendet werden kann backend/modules/parsers/__init__.py Um einen Parser zu registrieren, fügen Sie Folgendes hinzu:

register_parser("MyParser", MyParserClass)

Einbetter

Sobald wir die Daten in kleinere Teile aufgeteilt haben, wollen wir die wichtigsten Teile für eine bestimmte Frage finden. Eine schnelle und effektive Methode, dies zu tun, besteht darin, ein vortrainiertes Modell (Einbettungsmodell) zu verwenden, um unsere Daten und die Frage in spezielle Codes umzuwandeln, die als Einbettungen bezeichnet werden. Dann vergleichen wir die Einbettungen der einzelnen Datenblöcke mit denen für die Frage. Durch die Messung der Kosinusähnlichkeit Zwischen diesen Einbettungen können wir herausfinden, welche Chunks am engsten mit der Frage zusammenhängen, was uns hilft, die besten zu verwenden.

Zum Einbetten der Daten stehen viele vortrainierte Modelle zur Verfügung, z. B. Modelle von OpenAI, Cohere usw. Die beliebtesten Modelle finden Sie unter Der Benchmark für massive Texteinbettung (MTEB) von HuggingFace Bestenliste. Wir bieten Unterstützung für OpenAI Embeddings, TrueFoundry Embeddings und auch aktuelle SOTA Einbettungen (Stand April 2024) von gemischtes Brot - Ai.

Weitere Einbetter können einfach hinzugefügt werden unter Backend/Module/Embedder/ . Sobald ein Embedder hinzugefügt wurde, müssen Sie ihn registrieren, damit er von der RAG-Anwendung unter verwendet werden kann backend/modules/embedders/__init__.py Um einen Parser zu registrieren, fügen Sie Folgendes hinzu:

register_embedder("MyEmbedder", MyEmbedderClass)

‍Hinweis: Denken Sie daran, dass Einbettungen nicht die einzige Methode sind, um wichtige Chunks zu finden. Wir könnten für diese Aufgabe auch ein LLM verwenden! LLMs sind jedoch viel größer als Einbettungsmodelle und haben eine Begrenzung der Textmenge, die sie gleichzeitig verarbeiten können. Deshalb ist es klüger, Einbettungen zu verwenden, um zuerst die besten K Chunks auszuwählen. Dann können wir LLMs für diese weniger Chunks verwenden, um herauszufinden, welche Chunks am besten als Kontext für die Beantwortung unserer Frage verwendet werden können.

Reranker

Sobald beim Einbettungsschritt einige potenzielle Übereinstimmungen gefunden wurden, was eine Menge sein kann, wird ein Schritt zur Neubewertung durchgeführt. Eine Neubewertung stellt sicher, dass die besten Ergebnisse ganz oben stehen. Dadurch können wir die obersten x Dokumente auswählen, wodurch unser Kontext übersichtlicher wird und die Abfrage kürzer wird. Wir bieten Unterstützung für SOTA reranker (Stand April 2024) von gemischtes Brot - Ai was implementiert ist unter Backend/Module/Reranker/

Vektor DB

Sobald wir Vektoren für Texte erstellt haben, speichern wir sie in einer sogenannten Vektordatenbank. Diese Datenbank verfolgt diese Vektoren, sodass wir sie später mit verschiedenen Methoden schnell finden können. Reguläre Datenbanken organisieren Daten in Tabellen, wie Zeilen und Spalten, aber Vektordatenbanken sind etwas Besonderes, da sie Daten speichern und finden, die auf diesen Vektoren basieren. Das ist sehr nützlich, um Bilder zu erkennen, Sprache zu verstehen oder Dinge zu empfehlen. In einem Empfehlungssystem wird beispielsweise jeder Artikel, den du empfehlen möchtest (wie ein Film oder ein Produkt), in einen Vektor umgewandelt, wobei verschiedene Teile des Vektors verschiedene Merkmale des Artikels repräsentieren, wie zum Beispiel sein Genre oder seinen Preis. Ähnlich verhält es sich bei Sprachkenntnissen: Jedes Wort oder Dokument wird in einen Vektor umgewandelt, wobei Teile des Vektors Merkmale des Wortes oder Dokuments darstellen, z. B. wie oft das Wort verwendet wird oder was es bedeutet. Diese Vektordatenbanken sind so konzipiert, dass sie diese effizient verarbeiten. Mithilfe verschiedener Methoden, um zu messen, wie nah Vektoren beieinander sind, z. B. wie ähnlich sie sind oder wie weit sie voneinander entfernt sind, finden wir Vektoren, die der angegebenen Benutzerabfrage am nächsten kommen. Die gängigsten Methoden, dies zu messen, sind die euklidische Entfernung, die Kosinusähnlichkeit und das Punktprodukt.

Es gibt verschiedene verfügbare Vektordatenbanken auf dem Markt, wie Qdrant, SingleStore, Weaviate usw. Wir unterstützen derzeit Adrant und Einzelner Laden. Die Klasse Qdrant Vector DB ist definiert unter /backend/modules/vector_db/qdrant.py, während die SingleStore Vector DB-Klasse definiert ist unter /backend/modules/vector_db/singlestore.py

Andere Vektor-DBs können auch hinzugefügt werden in der vektor_db Ordner und kann registriert werden unter /backend/modules/vector_db/__init__.py

Um Vektor-DB-Unterstützung in Cognita hinzuzufügen, muss der Benutzer Folgendes tun:

• Erstellen Sie die Klasse, die erbt von BasisvektorDB (aus backend.modules.vector_db.base importiere BaseVectorDB) und initialisieren Sie es mit Vector DBConfig (aus backend.types importiere VectorDBConfig)
• Implementieren Sie die folgenden Methoden:
  • Sammlung erstellen: Um die Sammlung/das Projekt/die Tabelle in Vector DB zu initialisieren.
  • upsert_documents: Um die Dokumente in die Datenbank einzufügen.
  • get_collections: Ruft alle in der Datenbank vorhandenen Sammlungen ab.
  • löschen_Sammlung: Um die Sammlung aus der Datenbank zu löschen.
  • get_vector_store: Um den Vektorspeicher für die angegebene Sammlung abzurufen.
  • get_vector_client: Um den Vektor-Client für die angegebene Sammlung abzurufen, falls vorhanden.
  • list_daten_punkt_vektoren: Um bereits vorhandene Vektoren in der Datenbank aufzulisten, die den einzufügenden Dokumenten ähnlich sind.
  • Datenpunkt-Vektoren löschen: Um die Vektoren aus der Datenbank zu löschen, werden alte Vektoren des aktualisierten Dokuments entfernt.

Wir zeigen nun, wie wir dem RAG-System eine neue Vektor-DB hinzufügen können. Wir nehmen ein Beispiel für beide Adrant und Einzelner Laden Vektor-dbs.

Adrant-Integration

Adrant ist eine Open-Source-Vektordatenbank und Vektor-Suchmaschine, die in Rust geschrieben wurde. Es bietet einen schnellen und skalierbaren Vektor Ähnlichkeitssuche Service mit praktischer API. Gehen Sie wie folgt vor, um die Qdrant-Vektordatenbank zum RAG-System hinzuzufügen:

In der ENV-Datei können Sie Folgendes hinzufügen

VECTOR_DB_CONFIG = '{"url“: "<url_here>„, „provider“: „qdrant"}' # Qdrant-URL für die bereitgestellte Instanz

VECTOR_DB_CONFIG=' {"Anbieter“ :"qdrant“, "local“ :"true "} ' # Für lokale dateibasierte Qdrant-Instanz ohne Docker

1. Eine neue Klasse erstellen QDRANT Vector DB in backend/modules/vector_db/qdrant.py das erbt von BasisvektorDB und initialisiere es mit Vector 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. Überschreiben Sie die Sammlung erstellen Methode zum Erstellen einer Sammlung in Adrant

 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. Überschreiben Sie die upsert_documents Methode zum Einfügen der Dokumente in die Datenbank

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. Überschreiben Sie die get_collections Methode, um alle in der Datenbank vorhandenen Sammlungen abzurufen

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

5. Überschreiben Sie die löschen_Sammlung Methode zum Löschen der Sammlung aus der Datenbank

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

6. Überschreiben Sie die get_vector_store Methode, um den Vektorspeicher für die angegebene Sammlung zu erhalten

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

7. Überschreiben Sie die get_vector_client Methode, um den Vektor-Client für die angegebene Sammlung abzurufen, falls vorhanden

def get_vector_client(self):
    return self.qdrant_client

8. Überschreiben Sie die list_daten_punkt_vektoren Methode, um bereits vorhandene Vektoren in der Datenbank aufzulisten, die den einzufügenden Dokumenten ähnlich sind

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. Überschreiben Sie die Datenpunkt-Vektoren löschen Methode zum Löschen der Vektoren aus der Datenbank, die verwendet wird, um alte Vektoren des aktualisierten Dokuments zu entfernen

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-Integration

SingleStore bietet leistungsstarke Vektordatenbankfunktionen, die sich perfekt für KI-basierte Anwendungen, Chatbots, Bilderkennung und mehr eignen, sodass Sie keine spezielle Vektordatenbank ausschließlich für Ihre Vektor-Workloads betreiben müssen. Im Gegensatz zu herkömmlichen Vektordatenbanken speichert SingleStore Vektordaten zusammen mit anderen Datentypen in relationalen Tabellen. Durch die Verknüpfung von Vektordaten mit verwandten Daten können Sie problemlos erweiterte Metadaten und andere Attribute Ihrer Vektordaten mit der vollen Leistungsfähigkeit von SQL abfragen.

SingleStore bietet Entwicklern eine kostenlose Stufe, um mit ihrer Vektordatenbank zu beginnen. Sie können sich für ein kostenloses Konto anmelden hier. Gehen Sie nach der Anmeldung zu Wolke -> Arbeitsplatz -> Benutzer erstellen. Verwenden Sie die Anmeldeinformationen, um eine Verbindung zur SingleStore-Instanz herzustellen.

In der ENV-Datei können Sie Folgendes hinzufügen

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

Gehen Sie wie folgt vor, um SingleStore vector db zum RAG-System hinzuzufügen:

1. Wir möchten der Tabelle zusätzliche Spalten hinzufügen, um die Vektor-ID zu speichern, daher überschreiben wir 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. Eine neue Klasse erstellen SingleStore-Vector DB in backend/modules/vector_db/singlestore.py das erbt von BasisvektorDB und initialisiere es mit Vector DBConfig

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

3. Überschreiben Sie die Sammlung erstellen Methode zum Erstellen einer Sammlung 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. Überschreiben Sie die upsert_documents Methode zum Einfügen der Dokumente in die Datenbank

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. Überschreiben Sie die get_collections Methode, um alle in der Datenbank vorhandenen Sammlungen abzurufen

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. Überschreiben Sie die löschen_Sammlung Methode zum Löschen der Sammlung aus der Datenbank

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. Überschreiben Sie die get_vector_store Methode, um den Vektorspeicher für die angegebene Sammlung zu erhalten

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

8. Überschreiben Sie die get_vector_client Methode, um den Vektor-Client für die angegebene Sammlung abzurufen, falls vorhanden

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

9. Überschreiben Sie die list_daten_punkt_vektoren Methode, um bereits vorhandene Vektoren in der Datenbank aufzulisten, die den einzufügenden Dokumenten ähnlich sind

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. Überschreiben Sie die Datenpunkt-Vektoren löschen Methode zum Löschen der Vektoren aus der Datenbank, die verwendet wird, um alte Vektoren des aktualisierten Dokuments zu entfernen

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

Metadatenspeicher

Dies enthält die notwendigen Konfigurationen, die ein Projekt oder eine RAG-App eindeutig definieren. Eine RAG-App kann eine Reihe von Dokumenten aus einer oder mehreren Datenquellen zusammen enthalten, die wir als Kollektion. Die Dokumente aus diesen Datenquellen werden mithilfe von Methoden zum Laden, Analysieren und Einbetten von Daten in die Vektordatenbank indexiert. Für jeden RAG-Anwendungsfall enthält der Metadatenspeicher:

• Name der Sammlung
• Name der verwendeten zugehörigen Vector-DB
• Verknüpfte Datenquellen
• Konfiguration für jede Datenquelle analysieren
• Einbettungsmodell und dessen zu verwendende Konfiguration

Wir definieren derzeit zwei Möglichkeiten, diese Daten zu speichern, eine örtlich und andere Verwendungen Echtes Foudry. Diese Geschäfte sind definiert unter - Backend/Module/metada_store/

Controller abfragen

Sobald die Daten indexiert und in Vector DB gespeichert sind, ist es jetzt an der Zeit, alle Teile miteinander zu kombinieren, um unsere App zu verwenden. Query Controller machen das einfach! Sie helfen uns, die Antwort auf die entsprechende Benutzeranfrage abzurufen. Ein typischer Abfrage-Controller sieht wie folgt aus:

1. Benutzer senden eine Anforderungsnutzlast, die die Abfrage, den Sammlungsnamen, die LLM-Konfiguration, die Aufforderung, den Abruf und die zugehörige Konfiguration enthält.
2. Basierend auf dem Name der Sammlung relevant Vektor-db wird mit seiner Konfiguration wie verwendeter Embedder, Vektor-DB-Typ usw. abgerufen
3. Basierend auf dem abfragen, relevante Dokumente werden mit dem abgerufen Retriever von Vector DB.
4. Die abgerufenen Dokumente bilden den Kontext und zusammen mit der Abfrage a.k.a Frage wird gegeben an LLM um die Antwort zu generieren. Dieser Schritt kann auch eine schnelle Abstimmung beinhalten.
5. Falls erforderlich, werden zusammen mit der generierten Antwort auch relevante Dokumentblöcke in der Antwort zurückgegeben.

Hinweis: Bei Agenten können die Zwischenschritte auch gestreamt werden. Es liegt an der jeweiligen App, das zu entscheiden.

Abfrage-Controller-Methoden können direkt als API bereitgestellt werden, indem den jeweiligen Funktionen HTTP-Dekoratoren hinzugefügt werden.

Gehen Sie wie folgt vor, um Ihren eigenen Abfrage-Controller hinzuzufügen:

1. Jeder RAG-Anwendungsfall sollte normalerweise einen separaten Ordner haben, der den Abfrage-Controller enthält. Angenommen, unser Ordner ist App 2. Daher schreiben wir unseren Controller unter /backend/modules/query_controller/app-2/controller.py
2. Hinzufügen Query-Controller Decorator an Ihre Klasse und übergeben Sie den Namen Ihres benutzerdefinierten Controllers als Argument
3. Fügen Sie diesem Controller Methoden nach Ihren Bedürfnissen hinzu und verwenden Sie unsere HTTP-Dekoratoren wie Post, erhalten, löschen um Ihre Methoden zu einer API zu machen

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. Importiere deine benutzerdefinierte Controller-Klasse unter backend/modules/query_controllers/__init__.py

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

Ein Beispiel für einen Abfrage-Controller ist geschrieben unter: /backend/modules/query_controller/example/controller.py Bitte beziehen Sie sich zum besseren Verständnis

Cognita - Prozessablauf

Ein typischer Cognita-Prozess besteht aus zwei Phasen:

1. Indizierung von Daten
2. Generierung von Antworten

Datenindizierung

In dieser Phase werden Daten aus Quellen geladen, die in diesen Quellen vorhandenen Dokumente analysiert und in der Vektordatenbank indexiert. Cognita geht noch einen Schritt weiter, um große Mengen an Dokumenten verarbeiten zu können, die bei der Produktion anfallen.

1. Cognita gruppiert Dokumente stapelweise, anstatt sie alle zusammen zu indexieren.
2. Cognita berechnet und verfolgt den Dokument-Hash, sodass jedes Mal, wenn der Datenquelle ein neues Dokument zur Indexierung hinzugefügt wird, nur diese Dokumente indexiert werden, anstatt die gesamte Datenquelle zu indexieren. Dies spart Zeit und Rechenleistung erheblich.
3. Diese Art der Indizierung wird auch bezeichnet als INKREMENTELL Indizierung, es gibt auch einen anderen Modus, der in Cognita unterstützt wird, nämlich VOLL Indizierung. VOLL Bei der Indizierung werden die Daten erneut in den Vektor db aufgenommen, unabhängig davon, ob für die angegebene Sammlung Vektordaten vorhanden sind.

Generierung von Antworten

In der Phase der Antwortgenerierung wird der angerufen /antwort Endpunkt Ihres definierten Abfrage-Controller und generiert die Antwort für die angeforderte Anfrage.

Verwenden der Cognita-Benutzeroberfläche

Die folgenden Schritte zeigen, wie Sie die Cognita-Benutzeroberfläche zum Abfragen von Dokumenten verwenden:

1. Datenquelle erstellen

• Klicken Sie auf Datenquellen Registerkarte

• Klicken Sie + Neue Datenquelle
• Der Datenquellentyp kann entweder Dateien aus dem lokalen Verzeichnis, eine Web-URL, eine Github-URL oder die Bereitstellung von Truefoundry-Artefakt-FQN sein.
  • ZB: Wenn Lokales Verzeichnis ist ausgewählt, laden Sie Dateien von Ihrem Computer hoch und klicken Sie auf Einreichen.
• Die Liste der erstellten Datenquellen wird auf der Registerkarte Datenquellen verfügbar sein.

2. Sammlung erstellen

• Klicken Sie auf Sammlungen Registerkarte
• Klicken Sie + Neue Kollektion

• Namen der Sammlung eingeben
• Wählen Sie das Einbettungsmodell
• Fügen Sie eine zuvor erstellte Datenquelle und die erforderliche Konfiguration hinzu
• Klicken Sie Prozess um die Sammlung zu erstellen und die Daten zu indexieren.

3. Sobald Sie die Sammlung erstellt haben, beginnt die Datenaufnahme. Sie können ihren Status einsehen, indem Sie Ihre Sammlung im Tab Sammlungen auswählen. Sie können später auch weitere Datenquellen hinzufügen und sie in der Sammlung indizieren.

4. Generierung von Antworten

• Wählen Sie die Sammlung
• Wählen Sie das LLM und seine Konfiguration
• Wählen Sie den Dokumentabruder aus
• Schreiben Sie die Aufforderung oder verwenden Sie die Standardaufforderung
• Stellen Sie die Anfrage

Fangen Sie jetzt an!

Buche ein pPersonalisierte Demo oder melden Sie sich an heute, um mit der Erstellung Ihrer RAG-Anwendungsfälle zu beginnen.