Tutorial: criar um pipeline de indexação para o RAG na Pesquisa de IA do Azure

Artigo
11/19/2024

Aprenda a criar um pipeline de indexação automatizado para uma solução RAG usando a Pesquisa de IA do Azure. A automação da indexação é feita através de um indexador que gerencia a indexação e a execução do conjunto de habilidades, oferecendo chunking (divisão em partes) e vetorização de dados integradas de forma única ou recorrente para atualizações incrementais.

Neste tutorial, você:

Forneça o esquema de índices do tutorial anterior
Crie uma conexão de fonte de dados
Criar um indexador
Criar um conjunto de habilidades que agrupa, vetoriza e reconhece entidades
Execute o indexador e verifique os resultados

Se você não tiver uma assinatura do Azure, crie uma conta gratuita antes de começar.

Dica

Você pode utilizar o assistente de importação e vetorização de dados para criar seu pipeline. Experimente alguns exemplos de guias de início de rápido: Pesquisa de imagens e Busca em vetores.

Pré-requisitos

Visual Studio Code com a extensão do Python e o pacote do Jupyter. Para mais informações, Python no Visual Studio Code.
Conta de uso geral do Armazenamento do Microsoft Azure. Este exercício faz o upload de arquivos PDF para o armazenamento de blobs para realizar uma indexação automatizada.
A Pesquisa de IA do Azure requer nível Básico ou superior para obter uma identidade gerenciada e realizar classificação semântica. Escolha uma região compartilhada com o OpenAI do Azure e os Serviços de IA do Azure.
OpenAI do Azure com uma implantação de text-embedding-3-large, na mesma região que a Pesquisa de IA do Azure. Para mais informações sobre modelos de incorporação usados em soluções RAG, confira o artigo Escolha modelos de incorporação para RAG na Pesquisa de IA do Azure.
A conta multissserviço do Serviço de IA do Azure precisa estar na mesma região que a Pesquisa de IA do Azure. Esse recurso é usado para a habilidade de Reconhecimento de Entidades que detecta locais no seu conteúdo.

Baixar o exemplo

No GitHub, baixe um notebook do Jupyter para enviar as solicitações para a Pesquisa de IA do Azure. Para obter mais informações, consulte Baixando arquivos do GitHub.

Forneça o esquema de índices

Abra ou crie um Jupyter Notebook (.ipynb) no Visual Studio Code para conter os scripts que compõem o pipeline. As etapas iniciais instalam pacotes e coletam variáveis para as conexões. Depois de concluir as etapas de configuração, prepare-se para começar com os componentes do pipeline de indexação.

Vamos começar com o esquema de índices do tutorial anterior. Ele está organizado em torno de partes vetorizadas e não vetorizadas. Também inclui um campolocations que armazena o conteúdo gerado por IA criado pelo conjunto de habilidades.

from azure.identity import DefaultAzureCredential
from azure.identity import get_bearer_token_provider
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchField,
    SearchFieldDataType,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    AzureOpenAIVectorizer,
    AzureOpenAIVectorizerParameters,
    SearchIndex
)

credential = DefaultAzureCredential()

# Create a search index  
index_name = "py-rag-tutorial-idx"
index_client = SearchIndexClient(endpoint=AZURE_SEARCH_SERVICE, credential=credential)  
fields = [
    SearchField(name="parent_id", type=SearchFieldDataType.String),  
    SearchField(name="title", type=SearchFieldDataType.String),
    SearchField(name="locations", type=SearchFieldDataType.Collection(SearchFieldDataType.String), filterable=True),
    SearchField(name="chunk_id", type=SearchFieldDataType.String, key=True, sortable=True, filterable=True, facetable=True, analyzer_name="keyword"),  
    SearchField(name="chunk", type=SearchFieldDataType.String, sortable=False, filterable=False, facetable=False),  
    SearchField(name="text_vector", type=SearchFieldDataType.Collection(SearchFieldDataType.Single), vector_search_dimensions=1024, vector_search_profile_name="myHnswProfile")
    ]  
  
# Configure the vector search configuration  
vector_search = VectorSearch(  
    algorithms=[  
        HnswAlgorithmConfiguration(name="myHnsw"),
    ],  
    profiles=[  
        VectorSearchProfile(  
            name="myHnswProfile",  
            algorithm_configuration_name="myHnsw",  
            vectorizer_name="myOpenAI",  
        )
    ],  
    vectorizers=[  
        AzureOpenAIVectorizer(  
            vectorizer_name="myOpenAI",  
            kind="azureOpenAI",  
            parameters=AzureOpenAIVectorizerParameters(  
                resource_url=AZURE_OPENAI_ACCOUNT,  
                deployment_name="text-embedding-3-large",
                model_name="text-embedding-3-large"
            ),
        ),  
    ], 
)  
  
# Create the search index
index = SearchIndex(name=index_name, fields=fields, vector_search=vector_search)  
result = index_client.create_or_update_index(index)  
print(f"{result.name} created")

Crie uma conexão de fonte de dados

Nesta etapa, configure os dados de amostra e uma conexão com o Armazenamento de Blobs do Azure. O indexador recupera PDFs de um contêiner. Você cria o contêiner e faz upload dos arquivos nesta etapa.

O ebook original é grande, com mais de 100 páginas e 35 MB de tamanho. Dividimos em PDFs menores, um por página de texto, para ficar abaixo do limite de documento para indexadores de 16 MB por chamada à API e também os limites de dados de enriquecimento de IA. Para simplificar, omitimos a vetorização de imagem para este exercício.

Entre no portal do Azure e encontre sua conta de Armazenamento do Microsoft Azure.
Crie um contêiner e carregue os PDFs de earth_book_2019_text_pages.
Verifique se a Pesquisa de IA do Azure tem Leitor de Dados de Blob de Armazenamento como permissões no recurso.

No Visual Studio Code, defina uma fonte de dados do indexador com informações de conexão durante a indexação.

from azure.search.documents.indexes import SearchIndexerClient
from azure.search.documents.indexes.models import (
    SearchIndexerDataContainer,
    SearchIndexerDataSourceConnection
)

# Create a data source 
indexer_client = SearchIndexerClient(endpoint=AZURE_SEARCH_SERVICE, credential=credential)
container = SearchIndexerDataContainer(name="nasa-ebooks-pdfs-all")
data_source_connection = SearchIndexerDataSourceConnection(
    name="py-rag-tutorial-ds",
    type="azureblob",
    connection_string=AZURE_STORAGE_CONNECTION,
    container=container
)
data_source = indexer_client.create_or_update_data_source_connection(data_source_connection)

print(f"Data source '{data_source.name}' created or updated")

Se você configurar uma identidade gerenciada para a Pesquisa de IA do Azure para a conexão, a cadeia de conexão incluirá um sufixo ResourceId=. Ela deve ser parecida com o seguinte exemplo: "ResourceId=/subscriptions/FAKE-SUBCRIPTION=ID/resourceGroups/FAKE-RESOURCE-GROUP/providers/Microsoft.Storage/storageAccounts/FAKE-ACCOUNT;"

Criar um conjunto de habilidades

As habilidades são essenciais para o chunking de dados e vetorização integrados. No mínimo, você precisará de uma habilidade de Divisão de Texto para dividir seu conteúdo em partes e uma habilidade de incorporação para criar representações vetoriais do conteúdo dividido.

Neste conjunto de habilidades, uma habilidade extra é usada para criar dados estruturados no índice. A habilidade de Reconhecimento de Entidades identifica locais, que podem variar de nomes próprios a referências genéricas, como "oceano" ou "montanha". Dados estruturados oferecem mais opções para criar consultas interessantes e aumentar a relevância.

O AZURE_AI_MULTISERVICE_KEY é necessário mesmo se você estiver usando o controle de acesso baseado em função. A Pesquisa de IA do Azure usa a chave para fins de cobrança e é necessária, a menos que suas cargas de trabalho permaneçam abaixo do limite gratuito. Você também poderá ter uma conexão sem chave se estiver usando a API de versão prévia mais recente ou pacotes beta. Para obter mais informações, confira Anexar um recurso de vários serviços de IA do Azure a um conjunto de habilidades.

from azure.search.documents.indexes.models import (
    SplitSkill,
    InputFieldMappingEntry,
    OutputFieldMappingEntry,
    AzureOpenAIEmbeddingSkill,
    EntityRecognitionSkill,
    SearchIndexerIndexProjection,
    SearchIndexerIndexProjectionSelector,
    SearchIndexerIndexProjectionsParameters,
    IndexProjectionMode,
    SearchIndexerSkillset,
    CognitiveServicesAccountKey
)

# Create a skillset  
skillset_name = "py-rag-tutorial-ss"

split_skill = SplitSkill(  
    description="Split skill to chunk documents",  
    text_split_mode="pages",  
    context="/document",  
    maximum_page_length=2000,  
    page_overlap_length=500,  
    inputs=[  
        InputFieldMappingEntry(name="text", source="/document/content"),  
    ],  
    outputs=[  
        OutputFieldMappingEntry(name="textItems", target_name="pages")  
    ],  
)  
  
embedding_skill = AzureOpenAIEmbeddingSkill(  
    description="Skill to generate embeddings via Azure OpenAI",  
    context="/document/pages/*",  
    resource_url=AZURE_OPENAI_ACCOUNT,  
    deployment_name="text-embedding-3-large",  
    model_name="text-embedding-3-large",
    dimensions=1536,
    inputs=[  
        InputFieldMappingEntry(name="text", source="/document/pages/*"),  
    ],  
    outputs=[  
        OutputFieldMappingEntry(name="embedding", target_name="text_vector")  
    ],  
)

entity_skill = EntityRecognitionSkill(
    description="Skill to recognize entities in text",
    context="/document/pages/*",
    categories=["Location"],
    default_language_code="en",
    inputs=[
        InputFieldMappingEntry(name="text", source="/document/pages/*")
    ],
    outputs=[
        OutputFieldMappingEntry(name="locations", target_name="locations")
    ]
)
  
index_projections = SearchIndexerIndexProjection(  
    selectors=[  
        SearchIndexerIndexProjectionSelector(  
            target_index_name=index_name,  
            parent_key_field_name="parent_id",  
            source_context="/document/pages/*",  
            mappings=[  
                InputFieldMappingEntry(name="chunk", source="/document/pages/*"),  
                InputFieldMappingEntry(name="text_vector", source="/document/pages/*/text_vector"),
                InputFieldMappingEntry(name="locations", source="/document/pages/*/locations"),  
                InputFieldMappingEntry(name="title", source="/document/metadata_storage_name"),  
            ],  
        ),  
    ],  
    parameters=SearchIndexerIndexProjectionsParameters(  
        projection_mode=IndexProjectionMode.SKIP_INDEXING_PARENT_DOCUMENTS  
    ),  
) 

cognitive_services_account = CognitiveServicesAccountKey(key=AZURE_AI_MULTISERVICE_KEY)

skills = [split_skill, embedding_skill, entity_skill]

skillset = SearchIndexerSkillset(  
    name=skillset_name,  
    description="Skillset to chunk documents and generating embeddings",  
    skills=skills,  
    index_projection=index_projections,
    cognitive_services_account=cognitive_services_account
)
  
client = SearchIndexerClient(endpoint=AZURE_SEARCH_SERVICE, credential=credential)  
client.create_or_update_skillset(skillset)  
print(f"{skillset.name} created")

Crie e execute o indexador

Indexadores são o componente que iniciam todo o processo. Você pode criar um indexador em um estado desabilitado, mas o padrão é executá-lo imediatamente. Neste tutorial, crie e execute o indexador para recuperar os dados do Armazenamento de Blobs, executar as habilidades, como chunking e vetorização, e carregar o índice.

O indexador leva vários minutos para ser executado. Quando terminar, passe para a etapa final: consultar seu índice.

from azure.search.documents.indexes.models import (
    SearchIndexer,
    FieldMapping
)

# Create an indexer  
indexer_name = "py-rag-tutorial-idxr" 

indexer_parameters = None

indexer = SearchIndexer(  
    name=indexer_name,  
    description="Indexer to index documents and generate embeddings",  
    skillset_name=skillset_name,  
    target_index_name=index_name,  
    data_source_name=data_source.name,
    # Map the metadata_storage_name field to the title field in the index to display the PDF title in the search results  
    field_mappings=[FieldMapping(source_field_name="metadata_storage_name", target_field_name="title")],
    parameters=indexer_parameters
)  

# Create and run the indexer  
indexer_client = SearchIndexerClient(endpoint=AZURE_SEARCH_SERVICE, credential=credential)  
indexer_result = indexer_client.create_or_update_indexer(indexer)  

print(f' {indexer_name} is created and running. Give the indexer a few minutes before running a query.')

Execute uma consulta para verificar os resultados

Envie uma consulta para confirmar que seu índice está funcionando. Essa solicitação converte a cadeia de texto "what's NASA's website?" em um vetor para uma busca em vetores. Os resultados incluirão os campos na instrução select, alguns dos quais serão exibidos como saída.

Não há chat ou IA generativa neste momento. Os resultados são conteúdo verbatim do índice de pesquisa.

from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizableTextQuery

# Vector Search using text-to-vector conversion of the querystring
query = "what's NASA's website?"  

search_client = SearchClient(endpoint=AZURE_SEARCH_SERVICE, credential=credential, index_name=index_name)
vector_query = VectorizableTextQuery(text=query, k_nearest_neighbors=50, fields="text_vector")
  
results = search_client.search(  
    search_text=query,  
    vector_queries= [vector_query],
    select=["chunk"],
    top=1
)  
  
for result in results:  
    print(f"Score: {result['@search.score']}")
    print(f"Chunk: {result['chunk']}")

Essa consulta deve retornar uma única correspondência (top=1) consistindo na única parte considerada mais relevante pelo mecanismo de pesquisa. Os resultados da consulta devem ser semelhantes ao exemplo a seguir:

Score: 0.01666666753590107
Chunk: national Aeronautics and Space Administration

earth Science

NASA Headquarters 

300 E Street SW 

Washington, DC 20546

www.nasa.gov

np-2018-05-2546-hQ

Tente algumas consultas para ter uma noção do que o mecanismo de pesquisa retorna diretamente e compare com uma resposta habilitada por LLM. Execute novamente o script anterior com esta consulta: "patagonia geography" e defina top como 3 para retornar mais de uma resposta.

Os resultados desta segunda consulta devem parecer com os seguintes, que foram levemente editados facilitar a compreensão. A saída é copiada do notebook, que trunca a resposta ao que você vê neste exemplo. Você pode expandir a saída da célula para examinar a resposta completa.

Score: 0.03306011110544205
Chunk: 

Swirling Bloom off Patagonia
Argentina

Interesting art often springs out of the convergence of different ideas and influences. 
And so it is with nature. 

Off the coast of Argentina, two strong ocean currents converge and often stir up a colorful 
brew, as shown in this Aqua image from 

December 2010. 

This milky green and blue bloom formed on the continental shelf off of Patagonia, where warmer, 
saltier waters from the subtropics 

meet colder, fresher waters flowing from the south. Where these currents collide, turbulent 
eddies and swirls form, pulling nutrients 

up from the deep ocean. The nearby Rio de la Plata also deposits nitrogen- and iron-laden 
sediment into the sea. Add in some 
...

while others terminate in water. The San Rafael and San Quintín glaciers (shown at the right) 
are the icefield’s largest. Both have 

been receding rapidly in the past 30 years.

Com este exemplo, fica mais fácil ver como as partes são retornadas de forma literal e como a pesquisa por palavra-chave e similaridade encontra as correspondências mais relevantes. Essa parte específica definitivamente tem informações sobre Patagônia e geografia, mas não é exatamente relevante para a consulta. O classificador semântico promoveria partes mais relevantes para uma resposta melhor, mas como próxima etapa, vejamos como conectar a Pesquisa de IA do Azure a um LLM para pesquisa conversacional.

Próxima etapa

Converse com seus dados

Compartilhar via