Cognita: Construindo aplicações RAG de código aberto, modulares, para Produção

Suponha que haja uma equipe A designada para desenvolver um aplicativo RAG para use-case-1, então há a equipe B que está desenvolvendo um aplicativo RAG para use-case-2, e depois há a equipe C, que está apenas planejando seu próximo caso de uso de aplicativo RAG. Você já desejou que a construção de pipelines RAG em várias equipes fosse fácil? Cada equipe não precisaria começar do zero, mas sim de uma forma modular onde cada equipe pudesse usar a mesma funcionalidade base e desenvolver efetivamente seus próprios aplicativos sobre ela sem qualquer interferência?

Não se preocupe!! É por isso que criamos Cognita. Embora o RAG seja inegavelmente impressionante, o processo de criar um aplicativo funcional com ele pode ser assustador. Há uma quantidade significativa a ser compreendida sobre as práticas de implementação e desenvolvimento, desde a seleção dos modelos de IA apropriados para o caso de uso específico até a organização eficaz dos dados para obter os insights desejados. Embora ferramentas como LangChain e LlamaIndex existam para simplificar o processo de design de protótipos, ainda não havia um modelo RAG de código aberto acessível e pronto para uso que incorporasse as melhores práticas e oferecesse suporte modular, permitindo que qualquer pessoa o utilizasse de forma rápida e fácil.

Vantagens do Cognita:

1. Um repositório central reutilizável de parsers, carregadores, embedders e recuperadores.
2. Capacidade para usuários não técnicos interagirem com a interface do usuário - Carregar documentos e realizar Perguntas e Respostas (QnA) usando módulos construídos pela equipe de desenvolvimento.
3. Totalmente baseado em API - o que permite a integração com outros sistemas.

Visão Geral

Ao aprofundarmo-nos no funcionamento interno do Cognita, nosso objetivo era encontrar um equilíbrio entre personalização total e adaptabilidade, garantindo ao mesmo tempo a facilidade de uso desde o primeiro momento. Dado o rápido avanço em RAG e IA, era imperativo para nós projetar o Cognita com escalabilidade em mente, permitindo a integração perfeita de novas descobertas e diversos casos de uso. Isso nos levou a dividir o processo RAG em etapas modulares distintas (conforme mostrado no diagrama acima, a ser discutido nas seções subsequentes), facilitando a manutenção do sistema, a adição de novas funcionalidades, como a interoperabilidade com outras bibliotecas de IA, e permitindo que os usuários adaptem a plataforma às suas necessidades específicas. Nosso foco permanece em fornecer aos usuários uma ferramenta robusta que não apenas atenda às suas necessidades atuais, mas também evolua junto com a tecnologia, incluindo mudanças arquitetônicas mais amplas, como MCP vs RAG, garantindo valor a longo prazo.

Componentes

Cognita é projetado em torno de sete módulos diferentes, cada um personalizável e controlável para atender a diferentes necessidades:

1. Carregadores de Dados
2. Parsers
3. Embedders
4. Rerankers
5. Bancos de Dados Vetoriais
6. Armazenamento de Metadados
7. Controladores de Consulta

Carregadores de Dados

Estes carregam os dados de diferentes fontes, como diretórios locais, buckets S3, bancos de dados, Truefoundry artefatos, etc. Atualmente, o Cognita suporta o carregamento de dados de diretórios locais, URLs da web, repositórios Github e artefatos Truefoundry. Mais carregadores de dados podem ser facilmente adicionados em backend/modules/dataloaders/ . Uma vez que um carregador de dados é adicionado, você precisa registrá-lo para que possa ser usado pela aplicação RAG em backend/modules/dataloaders/__init__.py Para registrar um carregador de dados, adicione o seguinte:

register_dataloader("MyDataLoader", MyDataLoaderClass)

Analisadores

Nesta etapa, lidamos com diferentes tipos de dados, como arquivos de texto comuns, PDFs e até mesmo arquivos Markdown. O objetivo é transformar todos esses tipos diferentes em um formato comum para que possamos trabalhar com eles mais facilmente mais tarde. Esta parte, chamada de análise (parsing), geralmente é a mais demorada e difícil de implementar ao configurar um sistema como este. Mas usar o Cognita pode ajudar, pois ele já lida com o trabalho árduo de gerenciar pipelines de dados para nós.

Depois disso, dividimos os dados analisados em blocos uniformes. Mas por que precisamos disso? O texto que obtemos dos arquivos pode ter comprimentos diferentes. Se usarmos esses textos longos diretamente, acabaremos adicionando um monte de informações desnecessárias. Além disso, como todos os LLMs só conseguem lidar com uma certa quantidade de texto por vez, não seremos capazes de incluir todo o contexto importante necessário para a pergunta. Então, em vez disso, vamos dividir o texto em partes menores para cada seção. Intuitivamente, blocos menores conterão conceitos relevantes e serão menos "barulhentos" em comparação com blocos maiores.

Atualmente, suportamos a análise para Markdown, PDF e Texto arquivos. Mais analisadores de dados podem ser facilmente adicionados em backend/modules/parsers/ . Uma vez que um analisador é adicionado, você precisa registrá-lo para que possa ser usado pela aplicação RAG em backend/modules/parsers/__init__.py Para registrar um parser, adicione o seguinte:

register_parser("MyParser", MyParserClass)

Geradores de Embeddings

Depois de dividir os dados em partes menores, queremos encontrar os trechos mais importantes para uma pergunta específica. Uma maneira rápida e eficaz de fazer isso é usando um modelo pré-treinado (modelo de embedding) para converter nossos dados e a pergunta em códigos especiais chamados embeddings. Em seguida, comparamos os embeddings de cada trecho de dados com o da pergunta. Ao medir a similaridade de cosseno entre esses embeddings, podemos descobrir quais trechos estão mais intimamente relacionados à pergunta, ajudando-nos a encontrar os melhores para usar.

Existem muitos modelos pré-treinados disponíveis para incorporar os dados, como modelos da OpenAI, Cohere, etc. Os mais populares podem ser descobertos através do Massive Text Embedding Benchmark (MTEB) da HuggingFace leaderboard. Oferecemos suporte para OpenAI Embeddings, TrueFoundry Embeddings e também para os embeddings SOTA (a partir de abril de 2024) da mixedbread-ai.

Mais geradores de embeddings podem ser facilmente adicionados em backend/modules/embedder/ . Uma vez que um gerador de embeddings é adicionado, você precisa registrá-lo para que possa ser usado pela aplicação RAG em backend/modules/embedders/__init__.py Para registrar um parser, adicione o seguinte:

register_embedder("MyEmbedder", MyEmbedderClass)

‍Nota: Lembre-se, embeddings não são o único método para encontrar segmentos importantes. Poderíamos também usar um LLM para esta tarefa! No entanto, LLMs são muito maiores que os modelos de embedding e têm um limite na quantidade de texto que podem processar de uma vez. É por isso que é mais inteligente usar embeddings para selecionar os k melhores segmentos primeiro. Depois, podemos usar LLMs nesses poucos segmentos para descobrir os melhores a serem usados como contexto para responder à nossa pergunta.

Rerankers

Uma vez que a etapa de embedding encontra algumas correspondências potenciais, que podem ser muitas, uma etapa de reordenação (reranking) é aplicada. O reranking garante que os melhores resultados estejam no topo. Como resultado, podemos escolher os x melhores documentos, tornando nosso contexto mais conciso e a consulta do prompt mais curta. Oferecemos suporte para SOTA reranker (a partir de abril de 2024) da mixedbread-ai que é implementado em backend/modules/reranker/

VectorDB

Depois de criarmos vetores para textos, nós os armazenamos em algo chamado banco de dados vetorial. Este banco de dados rastreia esses vetores para que possamos encontrá-los rapidamente mais tarde usando diferentes métodos. Bancos de dados regulares organizam dados em tabelas, como linhas e colunas, mas os bancos de dados vetoriais são especiais porque armazenam e encontram dados com base nesses vetores. Isso é super útil para coisas como reconhecimento de imagens, compreensão de linguagem ou recomendação de itens. Por exemplo, em um sistema de recomendação, cada item que você pode querer recomendar (como um filme ou um produto) é transformado em um vetor, com diferentes partes do vetor representando diferentes características do item, como seu gênero ou preço. Da mesma forma, em processamento de linguagem, cada palavra ou documento é transformado em um vetor, com partes do vetor representando características da palavra ou documento, como a frequência com que a palavra é usada ou o que ela significa. Esses bancos de dados vetoriais são projetados para lidar com isso de forma eficiente. Usando diferentes maneiras de medir a proximidade dos vetores entre si, como o quão semelhantes são ou o quão distantes estão, encontramos os vetores mais próximos da consulta do usuário fornecida. As formas mais comuns de medir isso são Distância Euclidiana, Similaridade de Cosseno e Produto Escalar.

Existem vários bancos de dados vetoriais disponíveis no mercado, como Qdrant, SingleStore, Weaviate, etc. Atualmente, oferecemos suporte para Qdrant e SingleStore. A classe do banco de dados vetorial Qdrant é definida em /backend/modules/vector_db/qdrant.py, enquanto a classe do banco de dados vetorial SingleStore é definida em /backend/modules/vector_db/singlestore.py

Outros VectorDBs também podem ser adicionados em vector_db pasta e pode ser registrado em /backend/modules/vector_db/__init__.py

Para adicionar qualquer suporte a DB vetorial no Cognita, o usuário precisa fazer o seguinte:

• Crie a classe que herda de BaseVectorDB (from backend.modules.vector_db.base import BaseVectorDB) e inicialize-a com VectorDBConfig (from backend.types import VectorDBConfig)
• Implemente os seguintes métodos:
  • create_collection: Para inicializar a coleção/projeto/tabela no DB vetorial.
  • upsert_documents: Para inserir os documentos no DB.
  • get_collections: Obter todas as coleções presentes na base de dados.
  • delete_collection: Para excluir a coleção da base de dados.
  • get_vector_store: Para obter o armazenamento de vetores para a coleção especificada.
  • get_vector_client: Para obter o cliente de vetores para a coleção especificada, se houver.
  • list_data_point_vectors: Para listar os vetores já presentes na base de dados que são semelhantes aos documentos que estão sendo inseridos.
  • delete_data_point_vectors: Para excluir os vetores da base de dados, usado para remover vetores antigos do documento atualizado.

Mostramos agora como podemos adicionar uma nova base de dados de vetores ao sistema RAG. Usamos como exemplo tanto Qdrant e SingleStore bancos de dados vetoriais.

Integração Qdrant

Qdrant é um Banco de Dados Vetorial de Código Aberto e um Mecanismo de Busca Vetorial escrito em Rust. Ele oferece busca vetorial rápida e escalável por similaridade com uma API conveniente. Para adicionar o banco de dados vetorial Qdrant ao sistema RAG, siga os passos abaixo:

No arquivo .env você pode adicionar o seguinte

VECTOR_DB_CONFIG = '{"url": "<url_here>", "provider": "qdrant"}' # URL do Qdrant para instância implantada

VECTOR_DB_CONFIG='{"provider":"qdrant","local":"true"}' # Para uma instância Qdrant local baseada em arquivo, sem docker

1. Crie uma nova classe QdrantVectorDB em backend/modules/vector_db/qdrant.py que herda de BaseVectorDB e inicialize-a com 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. Sobrescreva o create_collection método para criar uma coleção no 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. Sobrescreva o upsert_documents método para inserir os documentos na base de dados

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. Sobrescreva o get_collections método para obter todas as coleções presentes na base de dados

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

5. Sobrescreva o delete_collection método para eliminar a coleção da base de dados

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

6. Sobrescreva o get_vector_store método para obter o armazenamento de vetores para a coleção especificada

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

7. Sobrescreva o get_vector_client método para obter o cliente de vetor para a coleção especificada, se existir

def get_vector_client(self):
    return self.qdrant_client

8. Sobrescreva o list_data_point_vectors método para listar vetores já presentes no banco de dados que são semelhantes aos documentos sendo inseridos

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. Sobrescreva o delete_data_point_vectors método para excluir os vetores do banco de dados, usado para remover vetores antigos do documento atualizado

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)

Integração com SingleStore

SingleStore oferece uma funcionalidade poderosa de banco de dados vetorial, perfeitamente adequada para aplicações baseadas em IA, chatbots, reconhecimento de imagem e muito mais, eliminando a necessidade de você operar um banco de dados vetorial especializado apenas para suas cargas de trabalho de vetores. Ao contrário dos bancos de dados vetoriais tradicionais, o SingleStore armazena dados vetoriais em tabelas relacionais juntamente com outros tipos de dados. A co-localização de dados vetoriais com dados relacionados permite que você consulte facilmente metadados estendidos e outros atributos de seus dados vetoriais com todo o poder do SQL.

SingleStore oferece um nível gratuito para desenvolvedores começarem a usar seu banco de dados vetorial. Você pode se inscrever para uma conta gratuita aqui. Após o cadastro, vá para Cloud -> workspace -> Criar Usuário. Use as credenciais para se conectar à instância do SingleStore.

No arquivo .env, você pode adicionar o seguinte

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

Para adicionar o banco de dados vetorial SingleStore ao sistema RAG, siga os passos abaixo:

1. Queremos adicionar colunas extras à tabela para armazenar o ID do vetor, portanto, sobrescrevemos o 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. Crie uma nova classe SingleStoreVectorDB em backend/modules/vector_db/singlestore.py que herda de BaseVectorDB e inicialize-a com VectorDBConfig

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

3. Sobrescreva o create_collection método para criar uma coleção no 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. Sobrescreva o upsert_documents método para inserir os documentos no banco de dados

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. Sobrescreva o get_collections método para obter todas as coleções presentes no banco de dados

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. Sobrescreva o delete_collection método para excluir a coleção do banco de dados

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. Sobrescrever o get_vector_store método para obter o armazenamento de vetores para a coleção especificada

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

8. Sobrescrever o get_vector_client método para obter o cliente de vetores para a coleção especificada, se houver

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

9. Sobrescrever o list_data_point_vectors método para listar vetores já presentes no banco de dados que são semelhantes aos documentos a serem inseridos

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. Sobrescrever o delete_data_point_vectors método para excluir os vetores do banco de dados, usado para remover vetores antigos do documento atualizado

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

Repositório de Metadados

Isto contém as configurações necessárias que definem de forma única um projeto ou aplicativo RAG. Um aplicativo RAG pode conter um conjunto de documentos de uma ou mais fontes de dados combinados, que denominamos como coleção. Os documentos dessas fontes de dados são indexados no banco de dados de vetores usando métodos de carregamento de dados + análise + incorporação. Para cada caso de uso RAG, o repositório de metadados contém:

• Nome da coleção
• Nome do Banco de Dados de Vetores associado utilizado
• Fontes de Dados Vinculadas
• Configuração de Análise para cada fonte de dados
• Modelo de Embedding e sua configuração a ser utilizada

Atualmente, definimos duas formas de armazenar esses dados, uma localmente e outra utilizando Truefoudry. Esses armazenamentos são definidos em - backend/modules/metada_store/

Controladores de Consulta

Uma vez que os dados são indexados e armazenados no db vetorial, agora é hora de combinar todas as partes para usar nosso aplicativo. Os Controladores de Consulta fazem exatamente isso! Eles nos ajudam a recuperar a resposta para a consulta do usuário correspondente. As etapas típicas de um controlador de consulta são as seguintes:

1. O usuário envia um payload da requisição que contém a consulta, nome da coleção, configuração do llm, prompt, retriever e sua configuração.
2. Com base no nome da coleção o db vetorial relevante é buscado com sua configuração como o embedder utilizado, tipo de db vetorial, etc.
3. Com base na consulta, documentos relevantes são recuperados utilizando o retriever do db vetorial.
4. Os documentos recuperados formam o contexto e, juntamente com a consulta, também conhecida como pergunta é fornecida ao LLM para gerar a resposta. Esta etapa também pode envolver o ajuste de prompts.
5. Se necessário, juntamente com a resposta gerada, fragmentos de documentos relevantes também são retornados na resposta.

Nota: No caso de agentes, as etapas intermediárias também podem ser transmitidas. Cabe à aplicação específica decidir.

Os métodos do controlador de consulta podem ser expostos diretamente como uma API, adicionando decoradores http às funções respetivas.

Para adicionar o seu próprio controlador de consulta, siga os seguintes passos:

1. Cada caso de uso RAG deve tipicamente ter uma pasta separada que contenha o controlador de consulta. Assuma que a nossa pasta é app-2. Assim, escreveremos o nosso controlador em /backend/modules/query_controller/app-2/controller.py
2. Adicione query_controller decorador à sua classe e passe o nome do seu controlador personalizado como argumento
3. Adicione métodos a este controlador conforme as suas necessidades e utilize os nossos decoradores http como post, get, delete para tornar seus métodos uma 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. Importe sua classe de controlador personalizada em backend/modules/query_controllers/__init__.py

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

Um controlador de consulta de exemplo está escrito em: /backend/modules/query_controller/example/controller.py Para melhor compreensão, consulte

Cognita - Fluxo do Processo

Um processo típico do Cognita consiste em duas fases:

1. Indexação de dados
2. Geração de Resposta

Indexação de Dados

Esta fase envolve o carregamento de dados de fontes, a análise dos documentos presentes nessas fontes e a indexação deles no banco de dados vetorial. Para lidar com grandes quantidades de documentos encontrados em produção, o Cognita vai um passo além.

1. O Cognita agrupa documentos em lotes, em vez de indexá-los todos de uma vez.
2. O Cognita calcula e rastreia o hash dos documentos para que, sempre que um novo documento for adicionado à fonte de dados para indexação, apenas esses documentos sejam indexados, em vez de indexar a fonte de dados completa. Isso economiza tempo e recursos computacionais drasticamente.
3. Este modo de indexação também é conhecido como INCREMENTAL indexação, existe também outro modo suportado no Cognita que é a COMPLETA indexação. A indexação COMPLETA re-ingere os dados para o banco de dados vetorial, independentemente de quaisquer dados vetoriais presentes para a coleção fornecida.

Geração de Resposta

A fase de geração de resposta faz uma chamada para o /answer endpoint do seu QueryController definido e gera a resposta para a consulta solicitada.

Usando a UI do Cognita

Os passos seguintes demonstram como usar a UI do Cognita para consultar documentos:

1. Criar Fonte de Dados

• Clique na aba Fontes de Dados

• Clique em + Nova Fonte de Dados
• O tipo de fonte de dados pode ser arquivos de diretório local, URL da web, URL do GitHub ou fornecendo o FQN do artefato Truefoundry.
  • Ex: Se Diretório local for selecionado, carregue os arquivos da sua máquina e clique em Enviar.
• A lista de fontes de dados criadas estará disponível na aba Fontes de Dados.

2. Criar Coleção

• Clique em Coleções aba
• Clique em + Nova Coleção

• Insira o Nome da Coleção
• Selecione o Modelo de Embedding
• Adicione a fonte de dados criada anteriormente e a configuração necessária
• Clique em Processar para criar a coleção e indexar os dados.

3. Assim que você cria a coleção, a ingestão de dados começa. Você pode ver o status dela selecionando sua coleção na aba de coleções. Você também pode adicionar fontes de dados adicionais mais tarde e indexá-las na coleção.

4. Geração de resposta

• Selecione a coleção
• Selecione o LLM e sua configuração
• Selecione o recuperador de documentos
• Escreva o prompt ou use o prompt padrão
• Faça a consulta

Comece agora!

Agende uma demonstração personalizada ou cadastre-se hoje para começar a construir seus casos de uso RAG.