OpenSearch:Crie um sistema de perguntas e respostas com base de conhecimento online usando RAG

Para cenários de perguntas e respostas com base de conhecimento online, o OpenSearch LLM-Based Conversational Search (AI Search Open Platform) oferece uma solução completa para construir um pipeline de desenvolvimento de geração aumentada por recuperação (RAG). Esse pipeline é composto por três módulos: pré-processamento de dados, service de recuperação e geração de resumos de perguntas e respostas. A AI Search Open Platform disponibiliza os serviços de algoritmos de cada módulo como componentes. Selecione flexivelmente os serviços utilizados em cada módulo, como análise de documentos, classificação e resumo de perguntas e respostas, e gere código de desenvolvimento rapidamente. A AI Search Open Platform expõe seus serviços como APIs. Baixe o código para seu ambiente local e substitua a API_KEY, o endpoint da API, a base de conhecimento local e outras informações seguindo as etapas deste tópico. Assim, você constrói rapidamente uma aplicação de perguntas e respostas com base de conhecimento baseada no pipeline de desenvolvimento RAG.

Como funciona

A Geração Aumentada por Recuperação (RAG) é um método de IA que combina tecnologias de recuperação e geração para melhorar a relevância, a precisão e a diversidade do conteúdo gerado por um modelo. Ao processar uma tarefa de geração, o RAG recupera primeiro os segmentos mais relevantes para a entrada a partir de um grande volume de dados externos ou de uma base de conhecimento. Em seguida, o RAG envia as informações recuperadas, juntamente com a entrada original, para um modelo de linguagem grande (LLM) como prompt ou contexto, orientando o modelo a gerar uma resposta mais precisa e abrangente. Esse método permite que o modelo não dependa apenas de seus parâmetros internos e dados de treinamento ao gerar respostas, mas também utilize informações externas atualizadas ou específicas do domínio para aumentar a precisão das respostas.

Cenários

O sistema de perguntas e respostas com base de conhecimento online é comum em cenários de negócios como recuperação e sumarização de bases de conhecimento corporativas internas e atendimento online em domínios verticais. Com base nos documentos profissionais da base de conhecimento do cliente, as tecnologias RAG e LLM compreendem e respondem a consultas complexas em linguagem natural. Isso ajuda os clientes empresariais a usar linguagem natural para recuperar rapidamente as informações necessárias em documentos PDF, Word, planilhas e imagens.

Na interface conversacional de perguntas e respostas, após o usuário enviar uma pergunta em linguagem natural, o sistema retorna uma resposta em formato de etapas estruturadas e fornece perguntas recomendadas abaixo da resposta para exploração adicional.

Pré-requisitos

• A AI Search Open Platform está ativada. Para mais informações, consulte Activate the service.

• Obtenha o endpoint do service e as informações de autenticação de identidade. Para mais informações, consulte Obtain a service endpoint e Obtain an API key.

  A AI Search Open Platform permite chamar serviços pela Internet ou usando um endpoint de virtual private cloud (VPC), além de suportar chamadas de service entre regiões via VPCs. Atualmente, usuários nas regiões China (Shanghai), China (Hangzhou), China (Shenzhen), China (Beijing), China (Zhangjiakou) e China (Qingdao) podem chamar os serviços da AI Search Open Platform usando endpoints VPC.

  Na página API Keys, a seguinte mensagem aparece na parte superior: "API keys are used for service invocation permissions. Keep them secure. If an API key is leaked, disable it immediately. Up to 10 API keys can be enabled at the same time." A seção Access Domain exibe o Public API Domain e o Private API Domain. Ambos suportam acesso HTTPS.

• Crie uma instância do Alibaba Cloud Elasticsearch (ES) versão 8,5 ou posterior. Para mais informações, consulte Create an Alibaba Cloud Elasticsearch cluster. Para acessar uma instância do Alibaba Cloud ES pela Internet ou por uma rede interna, adicione o endereço IP do dispositivo que acessa a instância à lista de permissões de acesso da instância. Para mais informações, consulte Manage IP address whitelists.

• Instale o Python 3.7 ou posterior e importe as dependências de pacotes Python aiohttp 3.8.6 e elasticsearch 8.14 para o ambiente de desenvolvimento.

Construa um pipeline de desenvolvimento RAG

Etapa 1: Selecione os serviços e baixe o código

Selecione os serviços de algoritmo e o framework de desenvolvimento para usar no pipeline RAG com base na sua base de conhecimento e nos requisitos de negócios. Este tópico usa o framework de desenvolvimento Python SDK como exemplo para descrever como construir um pipeline RAG.

1. Faça login no console da AI Search Open Platform.

2. Selecione a região China (Shanghai), mude para AI Search Open Platform e alterne para o workspace desejado.

   Nota

   • A AI Search Open Platform está disponível apenas nas regiões China (Shanghai) e Germany (Frankfurt).

   • Usuários nas regiões China (Hangzhou), China (Shenzhen), China (Beijing), China (Zhangjiakou) e China (Qingdao) podem usar um endpoint VPC para chamar serviços da AI Search Open Platform entre regiões.

3. No painel de navegação à esquerda, selecione Scene Center e clique em Enter no lado direito de RAG Scene-Knowledge Base Online Q & A..

4. Escolha os serviços necessários nas listas suspensas com base nas informações do service e nas características do seu negócio. Visualize os detalhes dos serviços na página Service Details.

   Nota

   • Ao chamar um service de algoritmo no pipeline RAG usando uma API, especifique o ID do service (service_id). Por exemplo, o ID do service de análise de conteúdo de documento é ops-document-analyze-001.

   • Após alternar para um service diferente na lista de serviços, o service_id no código gerado é atualizado automaticamente. Depois de baixar o código para o ambiente local, ainda é possível alterar o service_id para chamar o service correspondente.

   Estágio	Descrição do service
   Análise de conteúdo de documento	Serviço de análise de conteúdo de documento (ops-document-analyze-001): fornece um service de análise de documentos de uso geral. Extrai estruturas hierárquicas lógicas, como títulos e parágrafos, de documentos não estruturados (texto, tabelas, imagens, entre outros) e gera a saída em formato estruturado.
   Análise de conteúdo de imagem	Serviço de compreensão de conteúdo de imagem (ops-image-analyze-vlm-001): utiliza um modelo grande multimodal para analisar e compreender o conteúdo da imagem e reconhecer texto em imagens. O texto analisado pode ser usado em cenários de recuperação de imagens e perguntas e respostas. Serviço de reconhecimento de texto em imagem (ops-image-analyze-ocr-001): usa reconhecimento óptico de caracteres (OCR) para identificar texto em imagens. O texto extraído pode ser aplicado em cenários de recuperação de imagens e perguntas e respostas.
   Divisão de documentos	Serviço de divisão de documentos (ops-document-split-001): oferece um service de divisão de texto de uso geral. Divide dados estruturados nos formatos HTML, Markdown e TXT com base em parágrafos do documento, semântica do texto ou regras especificadas. Também extrai códigos, imagens e tabelas de documentos como rich text.
   Embedding de texto	OpenSearch text embedding service-001 (ops-text-embedding-001): fornece um service de embedding de texto para mais de 40 idiomas. O comprimento máximo do texto de entrada é 300, e o vetor de saída possui 1.536 dimensões. OpenSearch general text embedding service-002 (ops-text-embedding-002): oferece um service de embedding de texto para mais de 100 idiomas. O comprimento máximo do texto de entrada é 8.192, e o vetor de saída tem 1.024 dimensões. OpenSearch text embedding service-Chinese-001 (ops-text-embedding-zh-001): disponibiliza um service de embedding de texto para textos em chinês. O comprimento máximo do texto de entrada é 1.024, e o vetor de saída possui 768 dimensões. OpenSearch text embedding service-English-001 (ops-text-embedding-en-001): provê um service de embedding de texto para textos em inglês. O comprimento máximo do texto de entrada é 512, e o vetor de saída tem 768 dimensões.
   Embedding esparso de texto	Oferece um service que converte dados de texto em vetores esparsos. Vetores esparsos exigem menos espaço de armazenamento e são comumente usados para representar palavras-chave e informações de frequência de termos. Podem ser usados junto com vetores densos para recuperação híbrida, melhorando o desempenho da recuperação. OpenSearch text sparse embedding service (ops-text-sparse-embedding-001): fornece um service de embedding de texto para mais de 100 idiomas. O comprimento máximo do texto de entrada é 8.192.
   Análise de consulta	O Query analysis service 001 (ops-query-analyze-001) fornece um service de análise de consulta de uso geral. Utiliza um LLM para compreender a intenção das consultas do usuário e expandi-las com perguntas semelhantes.
   Motor de busca	Alibaba Cloud Elasticsearch: um service de cloud totalmente gerenciado baseado no Elasticsearch open source. É 100% compatível com recursos open source e está pronto para uso imediato com faturamento pagamento conforme o uso. Nota Se você selecionar o Alibaba Cloud Elasticsearch como service de motor de busca, o service de embedding esparso de texto ficará indisponível devido a problemas de compatibilidade. Recomendamos o uso dos serviços de embedding de texto. O OpenSearch Vector Search Edition é um motor de busca vetorial distribuído de grande escala desenvolvido pela alibaba. Suporta múltiplos algoritmos de recuperação vetorial e entrega excelente desempenho com alta precisão. Constrói índices e realiza recuperações em grande escala com alta relação custo-benefício. Além disso, os índices suportam dimensionamento horizontal e mesclagem, construção de índices em streaming, consultas em tempo real sobre dados recém-adicionados e atualizações dinâmicas de dados em tempo real. Nota Para utilizar a solução de engine OpenSearch Vector Search Edition, substitua a configuração da engine e o código no pipeline RAG.
   Serviço de classificação	Modelo de reranking BGE (ops-bge-reranker-larger): um service de pontuação de documentos de uso geral. Classifica os documentos em ordem decrescente de pontuação com base na relevância entre a consulta e o conteúdo do documento, e gera os resultados da pontuação.
   LLM	OpenSearch-Qwen-Turbo (ops-qwen-turbo): utiliza o LLM qwen-turbo como modelo base e aplica ajuste fino supervisionado para aprimorar o aumento por recuperação e reduzir conteúdo nocivo. Qwen-Turbo (qwen-turbo): um modelo de linguagem de ultra-grande escala Qwen que suporta entrada em diferentes idiomas, como chinês e inglês. Para mais informações, consulte Qwen Large Language Model Overview. Qwen-Plus (qwen-plus): uma versão aprimorada do modelo de linguagem de ultra-grande escala Qwen que aceita entradas em diferentes idiomas, como chinês e inglês. Para mais informações, consulte Qwen Large Language Model Overview. Qwen-Max (qwen-max): um modelo de linguagem de ultra-grande escala Qwen com centenas de bilhões de parâmetros que suporta entrada em diferentes idiomas, como chinês e inglês. Para mais informações, consulte Qwen Large Language Model Overview.

5. Após selecionar os serviços, clique em After the configuration is completed, enter the code query para visualizar e baixar o código.

   Com base no fluxo de execução quando uma aplicação chama o pipeline RAG, o código divide-se em duas partes: processamento offline de documentos e processamento online de perguntas e respostas do usuário.

   Processo	Finalidade	Descrição
   Processamento offline de documentos	Processa documentos, incluindo análise de documentos, extração de imagens, divisão, embedding e gravação dos resultados do processamento de documentos em um índice ES.	A função principal document_pipeline_execute executa o seguinte processo. Forneça os documentos a serem processados usando URLs de documentos ou codificação Base64. Análise de documento. Para informações sobre como chamar o service, consulte Document parsing API. Chama a operação assíncrona de análise de documento para extrair o conteúdo de um documento a partir de uma URL, ou decodifica o conteúdo de um arquivo codificado em Base64. Utiliza a função create_async_extraction_task para criar uma tarefa de análise e a função poll_task_result para verificar periodicamente o status de conclusão da tarefa. Extração de imagem. Para informações sobre como chamar o service, consulte Image content extraction API. Chama a operação assíncrona de análise de imagem para extrair o conteúdo de uma imagem a partir de uma URL, ou decodifica o conteúdo de um arquivo codificado em Base64. Usa a função create_image_analyze_task para criar uma tarefa de análise de imagem e a função get_image_analyze_task_status para obter o status da tarefa de análise de imagem. Divisão de documento. Para informações sobre como chamar o service, consulte Document splitting API. Chama a operação de divisão de documento para dividir o documento analisado com base na estratégia especificada. Utiliza a função document_split para dividir documentos. Isso inclui a divisão de documentos e a análise de conteúdo rich text. Embedding de texto. Para informações sobre como chamar o service, consulte Text embedding API. Chama a operação de representação vetorial densa de texto para converter o texto dividido em vetores. Usa a função text_embedding para calcular o vetor de embedding de cada chunk. Gravação de dados no ES. Para informações sobre como chamar o service, consulte Use the kNN search feature of Elasticsearch. Cria a configuração do índice ES, que especifica o campo vetorial embedding e o campo de conteúdo do documento content. Importante Quando um índice ES é criado, o sistema exclui qualquer índice existente com o mesmo nome. Para evitar que um índice existente com o mesmo nome seja excluído acidentalmente, altere o nome do índice no código. Utiliza a função helpers.async_bulk para gravar os resultados de embedding no índice ES em lotes.
   Processamento online de perguntas e respostas	Processa consultas online de usuários, incluindo geração de vetores de consulta, análise de consultas, recuperação de chunks de documentos relevantes, classificação dos resultados de recuperação e geração de respostas com base nos resultados recuperados.	A função principal query_pipeline_execute executa o seguinte processo para lidar com consultas de usuários e gerar respostas. Embedding de consulta. Para informações sobre como chamar o service, consulte Text embedding API. Chama a operação de representação vetorial densa de texto para converter consultas de usuários em vetores. Usa a função text_embedding para gerar vetores de consulta. Chame o service de análise de consulta. Para mais informações, consulte Query analysis API. Chama a operação de análise de consulta para identificar a intenção das perguntas do usuário e gerar perguntas semelhantes analisando mensagens históricas. Busca de chunks de embedding. Para informações sobre como chamar o service, consulte Use the kNN search feature of Elasticsearch. Usa o ES para recuperar chunks de documentos no índice que sejam semelhantes ao vetor de consulta. Utiliza a operação search do AsyncElasticsearch juntamente com uma consulta KNN para realizar a recuperação por similaridade. Chame o service de classificação. Para mais informações, consulte Ranking API. Chama a operação do service de classificação para pontuar e classificar os chunks relevantes recuperados. Usa a função documents_ranking para pontuar e classificar o conteúdo do documento com base na consulta do usuário. Chame um LLM para gerar uma resposta. Para informações sobre como chamar o service, consulte Answer generation API. Chama o service LLM e usa a função llm_call para gerar a resposta final com base nos resultados da recuperação e na consulta do usuário.

   Em Code Query, selecione Document processing flow e Online Q & A Process sucessivamente, e clique em Copy Code ou Download File para baixar o código para o seu ambiente local.

Etapa 2: Adapte o ambiente local e teste o pipeline de desenvolvimento RAG

Após baixar o código para dois arquivos locais, como online.py e offline.py, configure os parâmetros principais no código.

# Search development workbench configuration
api_key = "xxx"
host = "http://xxx.platform-cn-shanghai.opensearch.aliyuncs.com"
workspace_name = "default"
# Service ID configuration
service_id_config = {"extract": "ops-document-analyze-001", "split": "ops-document-split-001", "emb": "ops-text-embedding-001"}

Após configurar os parâmetros, execute o arquivo de processamento offline de documentos offline.py e, em seguida, o arquivo de processamento online de perguntas e respostas online.py em um ambiente Python 3.7 ou posterior para verificar se os resultados estão corretos.

Por exemplo, se o documento da base de conhecimento for AI Search Open Platform overview, faça a seguinte pergunta sobre o documento: O que a AI Search Open Platform pode fazer?

Os seguintes resultados são retornados:

• Resultado do processamento offline de documentos

  image analyze :https://img.alicdn.com/imgextra/i2/O1CN01bYc1m81RrcSAyOjMu_!!6000000002165-54-tps-60-60.apng
      https://img.alicdn.com/imgextra/i2/O1CN01bYc1m81RrcSAyOjMu_!!6000000002165-54-tps-60-60.apng is not analyzable.
      image analyze :https://help-static-aliyun-doc.aliyuncs.com/assets/img/zh-CN/3873436171/p802381.png
      image analyze :https://help-static-aliyun-doc.aliyuncs.com/assets/img/zh-CN/0650850271/p819277.png
      image analyze :https://help-static-aliyun-doc.aliyuncs.com/assets/img/zh-CN/0650850271/p819277.png
      image analyze ://gw.alicdn.com/tfs/TB1GxwdSXXXXXa.aXXXXXXXXXXX-65-70.gif
          https://gw.alicdn.com/tfs/TB1GxwdSXXXXXa.aXXXXXXXXXXX-65-70.gif is not analyzable.
      image analyze ://img.alicdn.com/tfs/TB1..50QpXXXX7XpXXXXXXXXXX-40-40.png
      image analyze :https://img.alicdn.com/tfs/TB1A0dINW6qK1RjSZFmXXX0PFXa-258-258.jpg
      image analyze ://gw.alicdn.com/tfs/TB1GxwdSXXXXXa.aXXXXXXXXXXX-65-70.gif
          https://gw.alicdn.com/tfs/TB1GxwdSXXXXXa.aXXXXXXXXXXX-65-70.gif is not analyzable.
      image analyze ://img.alicdn.com/tfs/TB1..50QpXXXX7XpXXXXXXXXXX-40-40.png
      image analyze ://gw.alicdn.com/tfs/TB1GxwdSXXXXXa.aXXXXXXXXXXX-65-70.gif
          https://gw.alicdn.com/tfs/TB1GxwdSXXXXXa.aXXXXXXXXXXX-65-70.gif is not analyzable.
      image analyze ://img.alicdn.com/tfs/TB1..50QpXXXX7XpXXXXXXXXXX-40-40.png
  text-embedding done
  OS write response:  {"status":"OK","code":200}

• Resultado do processamento online de perguntas e respostas

  /opt/miniconda3/envs/QA-pytest-base-lib1/bin/python /Users/liu/codeRepos/QA-pytest-base-lib/rag/case/SDK/python_sdk_es_zx.py
  query analysis rewrite result:What can the OpenSearch search development platform do?
  LLM final answer:  The OpenSearch search development platform provides intelligent search services. It supports the search services of core Alibaba Group businesses, including Taobao and Tmall, and provides intelligent search solutions for external customers in various industries. It offers industry-specific query semantic understanding and machine learning ranking algorithms, which help developers build intelligent search services with good performance. In addition, OpenSearch also
  The platform is suitable for a variety of scenarios, including but not limited to:
  - E-commerce and retail intelligent search
  - Content and news search
  - Gaming industry search
  - Healthcare industry search
  - Financial industry search
  The OpenSearch search development workbench focuses on intelligent search and RAG (Retrieval-Augmented Generation) scenarios, and provides componentized services and flexible invocation mechanisms. It has built-in services such as document parsing, document splitting, text embedding, retrieval, ranking, and LLMs, enabling one-stop, flexible AI search business development. With these services, users can easily work with RAG and search pipelines,
  Process finished with exit code 0

• View source code files

  offline.py

  # RAG offline pipeline - Elasticsearch engine
  # Environment requirements:
  # Python version: 3.7 or later
  # ES cluster version: 8.9 or later. If you use Alibaba Cloud Elasticsearch, you must activate it in advance and configure an IP address whitelist for access: https://www.alibabacloud.com/help/zh/es/user-guide/configure-a-public-or-private-ip-address-whitelist-for-an-elasticsearch-cluster
  # Package requirements:
  # pip install alibabacloud_searchplat20240529
  # pip install elasticsearch
  # AI Search Open Platform configuration
  aisearch_endpoint = "xxx.platform-cn-shanghai.opensearch.aliyuncs.com"
  api_key = "OS-xxx"
  workspace_name = "default"
  service_id_config = {"extract": "ops-document-analyze-001",
                       "split": "ops-document-split-001",
                       "text_embedding": "ops-text-embedding-001",
                       "text_sparse_embedding": "ops-text-sparse-embedding-001",
                       "image_analyze": "ops-image-analyze-ocr-001"}
  # ES configuration
  es_host = 'http://es-cn-xxx.public.elasticsearch.aliyuncs.com:9200'
  es_auth = ('elastic', 'xxx')
  # Input document URL. The sample document is the product description document of AI Search Open Platform
  document_url = "https://www.alibabacloud.com/help/zh/open-search/search-platform/product-overview/introduction-to-search-platform?spm=a2c4g.11186623.0.0.7ab93526WDzQ8z"
  import asyncio
  from typing import List
  from elasticsearch import AsyncElasticsearch
  from elasticsearch import helpers
  from alibabacloud_tea_openapi.models import Config
  from alibabacloud_searchplat20240529.client import Client
  from alibabacloud_searchplat20240529.models import GetDocumentSplitRequest, CreateDocumentAnalyzeTaskRequest, \
      CreateDocumentAnalyzeTaskRequestDocument, GetDocumentAnalyzeTaskStatusRequest, \
      GetDocumentSplitRequestDocument, GetTextEmbeddingRequest, GetTextEmbeddingResponseBodyResultEmbeddings, \
      GetTextSparseEmbeddingRequest, GetTextSparseEmbeddingResponseBodyResultSparseEmbeddings, \
      CreateImageAnalyzeTaskRequestDocument, CreateImageAnalyzeTaskRequest, CreateImageAnalyzeTaskResponse, \
      GetImageAnalyzeTaskStatusRequest, GetImageAnalyzeTaskStatusResponse
  async def poll_task_result(ops_client, task_id, service_id, interval=5):
      while True:
          request = GetDocumentAnalyzeTaskStatusRequest(task_id=task_id)
          response = await ops_client.get_document_analyze_task_status_async(workspace_name, service_id, request)
          status = response.body.result.status
          if status == "PENDING":
              await asyncio.sleep(interval)
          elif status == "SUCCESS":
              return response
          else:
              raise Exception("document analyze task failed")
  def is_analyzable_url(url:str):
      if not url:
          return False
      image_extensions = {'.jpg', '.jpeg', '.png', '.bmp', '.tiff'}
      return url.lower().endswith(tuple(image_extensions))
  async def image_analyze(ops_client, url):
      try:
          print("image analyze :" + url)
          if url.startswith("//"):
              url = "https:" + url
          if not is_analyzable_url(url):
              print(url + " is not analyzable.")
              return url
          image_analyze_service_id = service_id_config["image_analyze"]
          document = CreateImageAnalyzeTaskRequestDocument(
              url=url,
          )
          request = CreateImageAnalyzeTaskRequest(document=document)
          response: CreateImageAnalyzeTaskResponse = ops_client.create_image_analyze_task(workspace_name, image_analyze_service_id, request)
          task_id = response.body.result.task_id
          while True:
              request = GetImageAnalyzeTaskStatusRequest(task_id=task_id)
              response: GetImageAnalyzeTaskStatusResponse = ops_client.get_image_analyze_task_status(workspace_name, image_analyze_service_id, request)
              status = response.body.result.status
              if status == "PENDING":
                  await asyncio.sleep(5)
              elif status == "SUCCESS":
                  return url + response.body.result.data.content
              else:
                  print("image analyze error: " + response.body.result.error)
                  return url
      except Exception as e:
          print(f"image analyze Exception : {e}")
  def chunk_list(lst, chunk_size):
      for i in range(0, len(lst), chunk_size):
          yield lst[i:i + chunk_size]
  async def write_to_es(doc_list):
      es = AsyncElasticsearch(
          [es_host],
          basic_auth=es_auth,
          verify_certs=False,  # Do not verify SSL certificates
          request_timeout=30,
          max_retries=10,
          retry_on_timeout=True
      )
      index_name = 'dense_vertex_index'
      # Delete the existing index
      if await es.indices.exists(index=index_name):
          await es.indices.delete(index=index_name)
      # Create a vector index. Set the embedding field to dense_vector, the content field to text, and the source field to keyword
      index_mappings = {
          "mappings": {
              "properties": {
                  "emb": {
                      "type": "dense_vector",
                      "index": True,
                      "similarity": "cosine",
                      "dims": 1536  # Modify based on the output dimensions of the embedding model
                  },
                  "content": {
                      "type": "text"
                  },
                  "source_doc": {
                      "type": "keyword"
                  }
              }
          }
      }
      await es.indices.create(index=index_name, body=index_mappings)
      # Upload the embedding results to the index created in the previous step
      actions = []
      for i, doc in enumerate(doc_list):
          action = {
              "_index": index_name,
              "_id": doc['id'],
              "_source": {
                  "emb": doc['embedding'],
                  "content": doc['content'],
                  "source_doc": document_url
              }
          }
          actions.append(action)
      try:
          await helpers.async_bulk(es, actions)
      except Exception as e:
          for error in e.errors:
              print(error)
      # Confirm that the upload is successful
      await asyncio.sleep(2)
      query = {
          "query": {
              "ids": {
                  "values": [doc_list[0]["id"]]
              }
          }
      }
      res = await es.search(index=index_name, body=query)
      if len(res['hits']['hits']) > 0:
          print("ES write success")
      await es.close()
  async def document_pipeline_execute(document_url: str = None, document_base64: str = None, file_name: str = None):
      # Create the OpenSearch development platform client
      config = Config(bearer_token=api_key, endpoint=aisearch_endpoint, protocol="http")
      ops_client = Client(config=config)
      # Step 1: Document parsing
      document_analyze_request = CreateDocumentAnalyzeTaskRequest(
          document=CreateDocumentAnalyzeTaskRequestDocument(url=document_url, content=document_base64,
                                                            file_name=file_name, file_type='html'))
      document_analyze_response = await ops_client.create_document_analyze_task_async(workspace_name=workspace_name,
                                                                                      service_id=service_id_config[
                                                                                          "extract"],
                                                                                      request=document_analyze_request)
      print("document-analyze task_id:" + document_analyze_response.body.result.task_id)
      extraction_result = await poll_task_result(ops_client, document_analyze_response.body.result.task_id,
                                                 service_id_config["extract"])
      print("document-analyze done")
      document_content = extraction_result.body.result.data.content
      content_type = extraction_result.body.result.data.content_type
      # Step 2: Document splitting
      document_split_request = GetDocumentSplitRequest(
          GetDocumentSplitRequestDocument(content=document_content, content_type=content_type))
      document_split_result = await ops_client.get_document_split_async(workspace_name, service_id_config["split"],
                                                                        document_split_request)
      print("document-split done, chunks count: " + str(len(document_split_result.body.result.chunks))
            + " rich text count:" + str(len(document_split_result.body.result.rich_texts)))
      # Step 3: Text embedding
      # Extract the splitting results. Text content is extracted from image chunks by using the image analysis service
      doc_list = ([{"id": chunk.meta.get("id"), "content": chunk.content} for chunk in
                   document_split_result.body.result.chunks]
                  + [{"id": chunk.meta.get("id"), "content": chunk.content} for chunk in
                     document_split_result.body.result.rich_texts if chunk.meta.get("type") != "image"]
                  + [{"id": chunk.meta.get("id"), "content": await image_analyze(ops_client, chunk.content)} for chunk in
                     document_split_result.body.result.rich_texts if chunk.meta.get("type") == "image"]
                  )
      chunk_size = 32  # A maximum of 32 embeddings can be calculated at a time
      all_text_embeddings: List[GetTextEmbeddingResponseBodyResultEmbeddings] = []
      for chunk in chunk_list([text["content"] for text in doc_list], chunk_size):
          response = await ops_client.get_text_embedding_async(workspace_name, service_id_config["text_embedding"],
                                                               GetTextEmbeddingRequest(chunk))
          all_text_embeddings.extend(response.body.result.embeddings)
      all_text_sparse_embeddings: List[GetTextSparseEmbeddingResponseBodyResultSparseEmbeddings] = []
      for chunk in chunk_list([text["content"] for text in doc_list], chunk_size):
          response = await ops_client.get_text_sparse_embedding_async(workspace_name,
                                                                      service_id_config["text_sparse_embedding"],
                                                                      GetTextSparseEmbeddingRequest(chunk,
                                                                                                    input_type="document",
                                                                                                    return_token=True))
          all_text_sparse_embeddings.extend(response.body.result.sparse_embeddings)
      for i in range(len(doc_list)):
          doc_list[i]["embedding"] = all_text_embeddings[i].embedding
          doc_list[i]["sparse_embedding"] = all_text_sparse_embeddings[i].embedding
      print("text-embedding done")
      # Step 4: Write data to the Elasticsearch storage engine
      await write_to_es(doc_list)
  if __name__ == "__main__":
      # Run the asynchronous task
      #    import nest_asyncio # If you run the code in a Jupyter notebook, uncomment these two lines
      #    nest_asyncio.apply() # If you run the code in a Jupyter notebook, uncomment these two lines
      asyncio.run(document_pipeline_execute(document_url))
      # asyncio.run(document_pipeline_execute(document_base64="eHh4eHh4eHg...", file_name="attention.pdf")) # Alternative invocation method
            

  online.py

  # RAG online pipeline - Elasticsearch engine
  # Environment requirements:
  # Python version: 3.7 or later
  # ES cluster version: 8.9 or later (if you use Alibaba Cloud Elasticsearch, you must activate it in advance and configure an IP address whitelist for access: https://www.alibabacloud.com/help/zh/es/user-guide/configure-a-public-or-private-ip-address-whitelist-for-an-elasticsearch-cluster)
  # Package requirements:
  # pip install alibabacloud_searchplat20240529
  # pip install elasticsearch
  # OpenSearch search development workbench configuration
  api_key = "OS-xxx"
  aisearch_endpoint = "xxx.platform-cn-shanghai.opensearch.aliyuncs.com"
  workspace_name = "default"
  service_id_config = {
      "rank": "ops-bge-reranker-larger",
      "text_embedding": "ops-text-embedding-001",
      "text_sparse_embedding": "ops-text-sparse-embedding-001",
      "llm": "ops-qwen-turbo",
      "query_analyze": "ops-query-analyze-001"
  }
  # ES configuration
  es_host = 'http://es-cn-xxx.public.elasticsearch.aliyuncs.com:9200'
  es_auth = ('elastic', 'xxx')
  # User query:
  user_query = "What can the OpenSearch search development platform do?"
  import asyncio
  from elasticsearch import AsyncElasticsearch
  from alibabacloud_tea_openapi.models import Config
  from alibabacloud_searchplat20240529.client import Client
  from alibabacloud_searchplat20240529.models import GetTextEmbeddingRequest,  \
      GetDocumentRankRequest, GetTextGenerationRequest, GetTextGenerationRequestMessages, \
      GetQueryAnalysisRequest
  # Create the OpenSearch development platform client
  config = Config(bearer_token=api_key, endpoint=aisearch_endpoint, protocol="http")
  ops_client = Client(config=config)
  async def es_retrieve(query):
      es = AsyncElasticsearch(
          [es_host],
          basic_auth=es_auth,
          verify_certs=False,
          request_timeout=30,
          max_retries=10,
          retry_on_timeout=True
      )
      index_name = 'dense_vertex_index'
      # Query embedding
      query_emb_result = await ops_client.get_text_embedding_async(workspace_name, service_id_config["text_embedding"],
                                                                   GetTextEmbeddingRequest(input=[query],
                                                                                           input_type="query"))
      query_emb = query_emb_result.body.result.embeddings[0].embedding
      query = {
          "field": "emb",
          "query_vector": query_emb,
          "k": 5,  # Number of document chunks to return
          "num_candidates": 100  # HNSW search parameter efsearch
      }
      res = await es.search(index=index_name, knn=query)
      search_results = [item['_source']['content'] for item in res['hits']['hits']]
      await es.close()
      return search_results
  # Online Q&A pipeline. The input is the user question
  async def query_pipeline_execute():
      # Step 1: Query analysis
      query_analyze_response = ops_client.get_query_analysis(workspace_name, service_id_config['query_analyze'],
                                                             GetQueryAnalysisRequest(query=user_query))
      print("query analysis rewrite result:" + query_analyze_response.body.result.query)
      # Step 2: Retrieve documents
      all_query_results = []
      user_query_results = await es_retrieve(user_query)
      all_query_results.extend(user_query_results)
      rewrite_query_results = await es_retrieve(query_analyze_response.body.result.query)
      all_query_results.extend(rewrite_query_results)
      for extend_query in query_analyze_response.body.result.queries:
          extend_query_result = await es_retrieve(extend_query)
          all_query_results.extend(extend_query_result)
      # Deduplicate all retrieval results
      remove_duplicate_results = list(set(all_query_results))
      # Step 3: Rerank the retrieved documents
      rerank_top_k = 8
      score_results = await ops_client.get_document_rank_async(workspace_name, service_id_config["rank"],GetDocumentRankRequest(remove_duplicate_results, user_query))
      rerank_results = [remove_duplicate_results[item.index] for item in score_results.body.result.scores[:rerank_top_k]]
      # Step 4: Call the LLM to answer the question
      docs = '\n'.join([f"<article>{s}</article>" for s in rerank_results])
      messages = [
          GetTextGenerationRequestMessages(role="system", content="You are a helpful assistant."),
          GetTextGenerationRequestMessages(role="user",
                                           content=f"""The known information contains multiple independent documents. Each document is enclosed between <article> and </article>. The known information is as follows:\n'''{docs}'''
                                           \n\nBased on the preceding known information, answer the user's question in a detailed and well-organized manner. Make sure that the answer fully addresses the question and correctly uses the known information. If the information is insufficient to answer the question, say "The question cannot be answered based on the known information."
                                           Do not use content that is not in the known information to generate the answer. Make sure that every statement in the answer is supported by corresponding content in the preceding known information. Answer in Chinese.
                                           \nThe question is: '''{user_query}'''""""")
      ]
      response = await ops_client.get_text_generation_async(workspace_name, service_id_config["llm"],
                                                            GetTextGenerationRequest(messages=messages))
      print("Final answer from the LLM: ", response.body.result.text)
  if __name__ == "__main__":
      #    import nest_asyncio # If you run the code in a Jupyter notebook, uncomment these two lines
      #    nest_asyncio.apply() # If you run the code in a Jupyter notebook, uncomment these two lines
      asyncio.run(query_pipeline_execute())
            

FAQ

Durante a execução do código, uma mensagem Unclosed connector pode aparecer porque os recursos não foram liberados em tempo hábil. Ignore essa mensagem.