Tutorial: Criar um pipeline de indexação para RAG no Azure AI Search

Artigo
11/19/2024

Saiba como criar um pipeline de indexação automatizado para uma solução RAG no Azure AI Search. A automação da indexação é feita por meio de um indexador que impulsiona a indexação e a execução do conjunto de habilidades, fornecendo fragmentação e vetorização de dados integradas de forma única ou recorrente para atualizações incrementais.

Neste tutorial:

Forneça o esquema de índice do tutorial anterior
Criar uma conexão de fonte de dados
Criar um indexador
Crie um conjunto de habilidades que fragmente, vetorize e reconheça entidades
Execute o indexador e verifique os resultados

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

Gorjeta

Você pode usar o assistente Importar e vetorizar dados para criar seu pipeline. Experimente alguns inícios rápidos: Pesquisa de imagens e Pesquisa vetorial.

Pré-requisitos

Visual Studio Code com a extensão Python e o pacote Jupyter. Para obter mais informações, consulte Python no Visual Studio Code.
Conta de uso geral do Armazenamento do Azure. Este exercício carrega arquivos PDF no armazenamento de blob para indexação automatizada.
Azure AI Search, camada Básica ou superior para identidade gerenciada e classificação semântica. Escolha uma região que seja compartilhada com o Azure OpenAI e os Serviços de IA do Azure.
Azure OpenAI, com uma implantação de incorporação de texto 3-grande, na mesma região do Azure AI Search. Para obter mais informações sobre a incorporação de modelos usados em soluções RAG, consulte Escolher modelos de incorporação para RAG no Azure AI Search.
Conta multisserviço do Serviço Azure AI, na mesma região que o Azure AI Search. Este recurso é usado para a habilidade de Reconhecimento de Entidade que deteta locais em seu conteúdo.

Transferir o exemplo

Baixe um bloco de anotações Jupyter do GitHub para enviar as solicitações para o Azure AI Search. Para obter mais informações, consulte Baixando arquivos do GitHub.

Fornecer o esquema de índice

Abra ou crie um bloco de anotações Jupyter (.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, você estará pronto para começar com os componentes do pipeline de indexação.

Vamos começar com o esquema de índice do tutorial anterior. Organiza-se em torno de pedaços vetorizados e não vetorizados. Ele inclui um locations campo que armazena 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")

Criar uma conexão de fonte de dados

Nesta etapa, configure os dados de exemplo 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 carrega 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 documentos para indexadores de 16 MB por chamada de API e também dos limites de dados de enriquecimento de IA. Para simplificar, omitimos vetorização de imagem para este exercício.

Entre no portal do Azure e localize sua conta de Armazenamento do Azure.
Crie um contêiner e carregue os PDFs do earth_book_2019_text_pages.
Certifique-se de que o Azure AI Search tem permissões de Leitor de Dados de Blob de Armazenamento no recurso.

Em seguida, no Visual Studio Code, defina uma fonte de dados de indexador que forneça 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 o Azure AI Search para a conexão, a cadeia de conexão incluirá um ResourceId= sufixo. Deve ser semelhante ao seguinte exemplo: "ResourceId=/subscriptions/FAKE-SUBCRIPTION=ID/resourceGroups/FAKE-RESOURCE-GROUP/providers/Microsoft.Storage/storageAccounts/FAKE-ACCOUNT;"

Criar um conjunto de competências

As habilidades são a base para a fragmentação e vetorização de dados integrados. No mínimo, você deseja uma habilidade de Divisão de Texto para fragmentar seu conteúdo e uma habilidade de incorporação que crie representações vetoriais de seu conteúdo em partes.

Neste conjunto de habilidades, uma habilidade extra é usada para criar dados estruturados no índice. A habilidade de Reconhecimento de Entidade é usada para identificar locais, que podem variar de nomes próprios a referências genéricas, como "oceano" ou "montanha". Ter dados estruturados oferece 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. O Azure AI Search usa a chave para fins de cobrança e ela é necessária, a menos que suas cargas de trabalho permaneçam abaixo do limite gratuito. Você também pode usar uma conexão sem chave se estiver usando a API de visualização ou os pacotes beta mais recentes. Para obter mais informações, consulte Anexar um recurso multisserviço da 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")

Criar e executar o indexador

Os indexadores são o componente que coloca todos os processos em movimento. 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 Blob, execute as habilidades, incluindo fragmentação e vetorização, e carregue o índice.

O indexador leva vários minutos para ser executado. Quando terminar, você pode passar 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.')

Executar uma consulta para verificar os resultados

Envie uma consulta para confirmar se o índice está operacional. Esta solicitação converte a cadeia de texto "what's NASA's website?" em um vetor para uma pesquisa vetorial. Os resultados consistem nos campos na instrução select, alguns dos quais são impressos como saída.

Não há chat ou IA generativa neste momento. Os resultados são conteúdo literal do seu í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']}")

Esta consulta devolve uma única correspondência (top=1) que consiste num pedaço determinado pelo motor de pesquisa como sendo o mais relevante. 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

Experimente mais algumas consultas para ter uma noção do que o mecanismo de pesquisa retorna diretamente para que você possa compará-lo com uma resposta habilitada para 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 ser semelhantes aos resultados a seguir, que são levemente editados para concisão. A saída é copiada do bloco de anotações, que trunca a resposta ao que você vê neste exemplo. Você pode expandir a saída da célula para revisar 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, é mais fácil identificar como os blocos são retornados literalmente e como a pesquisa de palavras-chave e semelhanças identifica as principais correspondências. Este pedaço específico definitivamente tem informações sobre a 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, vamos ver como conectar o Azure AI Search a um LLM para pesquisa conversacional.

Próximo passo

Converse com os seus dados

Partilhar via