Como sincronizo automaticamente meus dados com o Amazon Bedrock?

Breve descrição

As organizações que usam uma abordagem baseada na geração aumentada via recuperação (RAG) em aplicações de IA devem manter sua base de conhecimento sincronizada com seus dados. Para automatizar as atualizações de dados, é possível usar a API StartIngestionJob.

Pré-requisitos:

• Uma conta da AWS com as permissões apropriadas.
• Familiaridade com o AWS SDK para a linguagem de programação preferida.

Observação: se você receber erros ao executar comandos da AWS Command Line Interface (AWS CLI), consulte Solucionar erros da AWS CLI. Além disso, verifique se você está usando a versão mais recente da AWS CLI.

Resolução

Usar a API IngestionJob

Conclua as etapas a seguir:

1. Configure seu AWS SDK com sua linguagem de programação preferida. Ou, se usar a AWS CLI, configure a AWS CLI com suas credenciais.

2. Para encontrar o ID da base de conhecimento, execute o comando list-knowledge-bases da AWS CLI:

   aws bedrock-agent list-knowledge-bases --region your-region-name

   Observação: substitua your-region-name pela sua região da AWS.

3. Para encontrar o ID de fonte de dados, execute o comando list-data-sources da AWS CLI:

   aws bedrock-agent list-data-sources --knowledge-base-id your-knowledge-base-id --region your-region-name

   Observação: substitua your-region-name pela sua região da AWS e your-knowledge-base-id pelo ID da base de conhecimento.

4. Execute a API StartIngestionJob:

   SDK_BedrockAgent_Client.StartIngestionJob(
     --knowledge-base-id your-knowledge-base-id
     --data-source-id your-data-source-id)

   Observação: a API pode parecer diferente dependendo da linguagem do programa que você usa. Ou pode ser diferente se usar a AWS CLI. Veja a seguir um exemplo que usa Python e boto3:

   import boto3from botocore.exceptions import ClientError
   def start_ingestion_job(knowledge_base_id, data_source_id):
       bedrock = boto3.client('bedrock-agent', region_name='your-region')
       try:
           response = bedrock.start_ingestion_job(
               knowledgeBaseId=knowledge_base_id,
               dataSourceId=data_source_id        )
           return response    except ClientError as e:
           print(f"An error occurred: {e}")
           return None
   
   --# Usage
   knowledge_base_id = 'your-knowledge-base-id'
   data_source_id = 'your-data-source-id'
   job_response = start_ingestion_job(knowledge_base_id, data_source_id)
   
   if job_response:
       print(f"Ingestion job started successfully. Job ID: {job_response['ingestionJob']['ingestionJobId']}")
   else:
       print("Failed to start ingestion job.")

5. Na saída, observe o ingestionJobId.

6. Para verificar o status da tarefa de ingestão, execute a API GetIngestionJob:

   SDK_BedrockAgent_Client.GetIngestionJob(
     --knowledge-base-id your-knowledge-base-id
     --data-source-id your-data-source-id
     --ingestion-job-id your-ingestion-job-id)

   Observação: a API pode parecer diferente dependendo da linguagem do programa que você usa. Ou pode ser diferente se usar a AWS CLI. Veja a seguir um exemplo que usa Python e boto3:

   def check_ingestion_job_status(knowledge_base_id, data_source_id, ingestion_job_id):
       bedrock = boto3.client('bedrock-agent', region_name='your-region')
       try:
           response = bedrock.get_ingestion_job(
               knowledgeBaseId=knowledge_base_id,
               dataSourceId = data_source_id,
               ingestionJobId=ingestion_job_id        )
           return response['ingestionJob']['status']
       except ClientError as e:
           print(f"An error occurred: {e}")        return None
   
   --# Usage
   if job_response:
       status = check_ingestion_job_status(knowledge_base_id, data_source_id, ingestion_job_id)
       print(f"Current ingestion job status: {status}")

Usar pseudocódigo para enviar os dados para a base de conhecimento

Use um pseudocódigo para atualizar os dados de todas as fontes de dados disponíveis em sua base de conhecimento.

Exemplo:

Function StartJob(knowledgeBaseId, dataSourceId)
    Try        job = BedrockAgentService.StartIngestionJob(knowledgeBaseId, dataSourceId)
        Return job
    Catch Error        LogError("Failed to start ingestion job for data source: " + dataSourceId)
        Return null
Function GetIngestionJobStatus(knowledgeBaseId, dataSourceId, ingestionJobId)
    Try        jobStatus = BedrockAgentService.GetIngestionJob(knowledgeBaseId, dataSourceId, ingestionJobId)
        Return jobStatus
    Catch Error        LogError("Failed to get status for job: " + ingestionJobId)
        Return null
Function RunIngestionJobs(knowledgeBaseId)
    dataSources = BedrockAgentService.ListDataSources(knowledgeBaseId)

    For Each dataSource in dataSources        job = StartJob(knowledgeBaseId, dataSource.Id)

        If job is not null Then            LogInfo("Job started successfully for data source: " + dataSource.Id)

            While job.Status is not (Completed or Failed or Stopped)
                Wait for short interval                job = GetIngestionJobStatus(knowledgeBaseId, dataSource.Id, job.Id)

                If job is null Then                    Break While loop
                        If job is not null Then                LogInfo("Job completed with status: " + job.Status)
            Else                LogError("Job monitoring failed for data source: " + dataSource.Id)
        Else            LogError("Failed to start job for data source: " + dataSource.Id)

Main    knowledgeBaseId = "<your-knowledge-base-id"
    RunIngestionJobs(knowledgeBaseId)

O código define três funções principais:

• A função StartJob usa a API StartIngestionJob para iniciar uma tarefa de ingestão para a base de conhecimento e a fonte de dados que você fornece.
• A função GetIngestionJobStatus obtém o status atual da tarefa de ingestão fornecida.
• A função RunIngestionJobs inicia e monitora as tarefas de ingestão das fontes de dados na base de conhecimento fornecida.

Observação: se uma operação falhar, revise as mensagens de erro para mais detalhes.

Seguir as práticas recomendadas

Para reduzir os problemas ao sincronizar seus dados com o Amazon Bedrock, adote as seguintes práticas recomendadas:

• Para gerenciar problemas com chamadas de API, implemente o tratamento de erros em todo o processo.
• Para verificar periodicamente o status de uma tarefa, use um mecanismo de pesquisa. Para mais informações, consulte Pesquisa sobre a situação do trabalho na Lambda e AWS Batch.
• Mantenha logs detalhados do processo de ingestão para fins de solução de problemas e auditoria.
• Siga as práticas recomendadas de segurança da AWS.
• Analise os custos de ingestão e armazenamento de dados.