OpenSearch:Créer un système de questions-réponses sur une base de connaissances à l’aide du RAG

AI Search Open Platform propose un pipeline de développement RAG complet, couvrant le prétraitement des données, la récupération et la génération de réponses. Les services algorithmiques sont composables et exposés via des API. Téléchargez le code fourni, remplacez la clé API, l'endpoint ainsi que les informations relatives à la base de connaissances comme décrit dans cette rubrique pour créer rapidement une application de recherche conversationnelle sur votre propre base de connaissances.

Fonctionnement

La génération augmentée par récupération (RAG) associe la récupération d'informations aux grands modèles de langage (LLM) afin d'améliorer la précision et la pertinence du contenu généré. Un système RAG récupère les informations les plus pertinentes depuis une base de connaissances externe et les fournit comme contexte au LLM. Cette approche permet au modèle de générer des réponses précises fondées sur des données actualisées ou spécifiques à un domaine, plutôt que de s'appuyer uniquement sur ses données d'entraînement.

Cas d'utilisation

La recherche conversationnelle sur une base de connaissances répond à des besoins tels que la récupération de connaissances en entreprise et les questions-réponses spécifiques à un domaine. En appliquant le RAG et les LLM à vos documents, le système comprend des requêtes complexes en langage naturel et aide les utilisateurs à trouver rapidement des informations dans divers formats (PDF, Word, tableaux et images).

Dans l'interface de recherche conversationnelle, un utilisateur soumet une question en langage naturel. Le système renvoie une réponse structurée et suggère des questions de suivi pour approfondir l'exploration.

Prérequis

• Le service AI Search Open Platform est activé. Pour plus d'informations, consultez la rubrique Activation du service.

• Obtenez un endpoint de service et des identifiants d'authentification. Pour plus d'informations, consultez les rubriques Obtenir un endpoint de service et Gérer les clés API.

  AI Search Open Platform prend en charge les endpoints publics et VPC. Les appels interrégions sont possibles via les endpoints VPC. Actuellement, l'accès VPC est disponible dans les régions Chine (Shanghai), Chine (Hangzhou), Chine (Shenzhen), Chine (Pékin), Chine (Zhangjiakou) et Chine (Qingdao).

  Sur la page API Keys, un message en haut indique : « Les clés API sont utilisées pour les autorisations d'appel de service. Veuillez les conserver en toute sécurité. Si une clé API est compromise, désactivez-la immédiatement. Vous pouvez avoir jusqu'à 10 clés API activées simultanément. » La section Access Domain affiche le Public API Domain et le Private API Domain, qui prennent tous deux en charge l'accès HTTPS.

• Créez un cluster Alibaba Cloud Elasticsearch version 8.5 ou ultérieure. Pour plus d'informations, consultez la rubrique Créer un cluster Alibaba Cloud Elasticsearch. Lors de l'accès au cluster via le réseau public ou un VPC, vous devez ajouter l'adresse IP de votre appareil à la liste d'autorisation d'adresses IP du cluster. Pour plus d'informations, consultez la rubrique Configurer une liste d'autorisation d'adresses IP publique ou privée pour un cluster Elasticsearch.

• Vous disposez d'un environnement Python 3.7 ou ultérieur avec les packages aiohttp 3.8.6 et elasticsearch 8.14 installés.

Créer le pipeline de développement RAG

Étape 1 : Sélectionner les services et télécharger le code

Sélectionnez les services algorithmiques et le framework de développement pour votre pipeline RAG en fonction de votre base de connaissances et de vos besoins métier. Cette rubrique utilise le SDK Python à titre d'exemple.

1. Connectez-vous à la console AI Search Open Platform.

2. Sélectionnez la région Chine (Shanghai), basculez vers AI Search Open Platform, puis sélectionnez votre espace de travail cible.

   Remarque

   • AI Search Open Platform est disponible uniquement dans les régions Chine (Shanghai) et Allemagne (Francfort).

   • Les utilisateurs des régions Chine (Hangzhou), Chine (Shenzhen), Chine (Pékin), Chine (Zhangjiakou) et Chine (Qingdao) peuvent utiliser un endpoint VPC pour appeler les services AI Search Open Platform entre régions.

3. Dans le volet de navigation de gauche, cliquez sur Scene Center. Sur la carte RAG Scene-Knowledge Base Online Q & A., cliquez sur Enter.

4. Dans les listes déroulantes, sélectionnez les services nécessaires selon vos besoins métier. Vous pouvez consulter les détails de chaque service sous l'onglet Service Details.

   Remarque

   • Lorsque vous appelez un service algorithmique dans le pipeline RAG via une API, vous devez fournir l'ID du service (service_id). Par exemple, l'ID du service d'analyse de contenu de document est ops-document-analyze-001.

   • Lorsque vous changez de service dans la liste, le service_id dans le code généré se met à jour automatiquement. Après le téléchargement du code, vous pouvez toujours modifier le service_id pour appeler un autre service.

   Étape	Description du service
   Analyse de contenu de document	Service d'analyse de contenu de document (ops-document-analyze-001) : service polyvalent qui extrait les structures logiques telles que les titres et les paragraphes à partir de documents non structurés (texte, tableaux et images) et les produit dans un format structuré.
   Analyse de contenu d'image	Service de compréhension de contenu d'image (ops-image-analyze-vlm-001) : utilise un grand modèle multimodal pour analyser, comprendre et reconnaître le texte des images. Le texte extrait peut servir à la recherche d'images et aux scénarios de questions-réponses. Service de reconnaissance optique de caractères (OCR) d'images (ops-image-analyze-ocr-001) : utilise l'OCR pour reconnaître le texte dans les images. Le texte analysé peut servir à la recherche d'images et aux scénarios de questions-réponses.
   Découpage de document	Service de découpage de document (ops-document-split-001) : service de découpage de texte polyvalent qui divise les données structurées aux formats HTML, Markdown et TXT selon les paragraphes du document, la sémantique du texte ou des règles spécifiées. Il prend également en charge l'extraction du code, des images et des tableaux sous forme de texte enrichi à partir des documents.
   Intégration de texte	Service d'intégration de texte OpenSearch-001 (ops-text-embedding-001) : propose une intégration de texte multilingue (plus de 40 langues). La longueur d'entrée maximale est de 300 tokens et la dimension du vecteur de sortie est de 1536. Service d'intégration de texte général OpenSearch-002 (ops-text-embedding-002) : propose une intégration de texte multilingue (plus de 100 langues). La longueur d'entrée maximale est de 8 192 tokens et la dimension du vecteur de sortie est de 1024. Service d'intégration de texte chinois OpenSearch-001 (ops-text-embedding-zh-001) : propose une intégration de texte chinois. La longueur d'entrée maximale est de 1 024 tokens et la dimension du vecteur de sortie est de 768. Service d'intégration de texte anglais OpenSearch-001 (ops-text-embedding-en-001) : propose une intégration de texte anglais. La longueur d'entrée maximale est de 512 tokens et la dimension du vecteur de sortie est de 768.
   Intégration de texte clairsemée	Convertit le texte en une représentation vectorielle clairsemée. Les vecteurs clairsemés utilisent moins d'espace de stockage et représentent généralement les mots-clés et les fréquences des termes. Combinez-les avec des vecteurs denses pour une recherche hybride afin d'améliorer les performances de récupération. Service d'intégration de texte clairsemée OpenSearch (ops-text-sparse-embedding-001) : propose une intégration de texte clairsemée multilingue (plus de 100 langues). La longueur d'entrée maximale est de 8 192 tokens.
   Analyse de requête	Service d'analyse de requête 001 (ops-query-analyze-001) : utilise un grand modèle de langage pour comprendre l'intention de la requête utilisateur et l'étendre avec des questions similaires.
   Moteur de recherche	Alibaba Cloud Elasticsearch : service cloud entièrement géré basé sur Elasticsearch open source, 100 % compatible avec les fonctionnalités open source, offrant une expérience prête à l'emploi et facturée à l'utilisation. Remarque Si vous choisissez Alibaba Cloud Elasticsearch comme moteur de recherche, le service d'intégration de texte clairsemée n'est pas disponible en raison de problèmes de compatibilité. Nous vous recommandons d'utiliser les services d'intégration de texte à la place. OpenSearch Vector Search Edition : moteur de recherche vectoriel distribué à grande échelle développé par Alibaba, qui prend en charge divers algorithmes de recherche vectorielle avec une haute précision et une indexation rentable à grande échelle. Les index prennent en charge la mise à l'échelle horizontale, les builds en flux continu, les requêtes en temps réel et les mises à jour dynamiques des données. Remarque Si vous devez utiliser OpenSearch Vector Search Edition, vous pouvez remplacer la configuration du moteur et le code dans le pipeline RAG.
   Service de reclassement	Modèle de reclassement BGE (ops-bge-reranker-larger) : évalue et trie les documents par pertinence par rapport à la requête, du plus pertinent au moins pertinent, et produit les résultats de notation.
   Grand modèle de langage	OpenSearch-Qwen-Turbo (ops-qwen-turbo) : basé sur le grand modèle de langage Qwen-Turbo, ce service est affiné par apprentissage supervisé pour améliorer l'augmentation de la récupération et réduire les réponses nuisibles. Qwen-Turbo (qwen-turbo) : grand modèle de langage de la série Qwen qui prend en charge diverses langues, dont le chinois et l'anglais. Pour plus d'informations, consultez la rubrique Présentation des LLM de la série Qwen. Qwen-Plus (qwen-plus) : version améliorée du grand modèle de langage Qwen-Turbo qui prend en charge diverses langues, dont le chinois et l'anglais. Pour plus d'informations, consultez la rubrique Présentation des LLM de la série Qwen. Qwen-Max (qwen-max) : modèle de langage ultra-grand à mille milliards de paramètres de la série Qwen qui prend en charge diverses langues, dont le chinois et l'anglais. Pour plus d'informations, consultez la rubrique Présentation des LLM de la série Qwen.

5. Après avoir sélectionné vos services, cliquez sur After the configuration is completed, enter the code query pour afficher et télécharger le code.

   Le code est structuré en deux parties qui reflètent le flux d'exécution du pipeline RAG : le traitement hors ligne des documents et la recherche conversationnelle en ligne.

   Processus	Fonction	Description
   Traitement hors ligne des documents	Traite les documents, ce qui inclut l'analyse, l'extraction d'images, le découpage, l'intégration et l'écriture des résultats dans un index Elasticsearch.	La fonction principale document_pipeline_execute exécute le workflow suivant. Vous pouvez fournir des documents via une URL ou un encodage Base64. Analyse de document. Pour plus de détails sur l'API, consultez la rubrique API d'analyse de document. Appelez l'API d'analyse asynchrone de document pour extraire le contenu à partir d'une URL de document ou décoder le contenu à partir d'un fichier encodé en Base64. Utilisez la fonction create_async_extraction_task pour créer une tâche d'analyse et la fonction poll_task_result pour interroger l'état d'achèvement de la tâche. Extraction d'images. Pour plus de détails sur l'API, consultez la rubrique API d'extraction de contenu d'image. Appelez l'API d'analyse asynchrone d'images pour extraire le contenu à partir d'une URL d'image ou le décoder à partir d'un fichier encodé en Base64. Utilisez la fonction create_image_analyze_task pour créer une tâche d'analyse d'image et la fonction get_image_analyze_task_status pour obtenir son état. Découpage de document. Pour plus de détails sur l'API, consultez la rubrique API de découpage de document. Appelez l'API de découpage de document pour diviser le document analysé selon une stratégie spécifiée. Utilisez la fonction document_split pour le découpage de document et l'analyse de contenu de texte enrichi. Intégration de texte. Pour plus de détails sur l'API, consultez la rubrique API d'intégration de texte. Appelez l'API d'intégration de texte pour créer des représentations vectorielles du texte découpé. Utilisez la fonction text_embedding pour calculer le vecteur d'intégration pour chaque fragment. Écriture dans Elasticsearch. Pour plus de détails sur le service, consultez la rubrique Utiliser la fonctionnalité de recherche k-nearest neighbor (kNN) d'Elasticsearch. Créez une configuration d'index Elasticsearch qui spécifie le champ vectoriel embedding et le champ de contenu de document content. Important Lorsque vous créez un index Elasticsearch, tout index existant portant le même nom est supprimé. Pour éviter une perte accidentelle de données, modifiez le nom de l'index dans le code. Utilisez la fonction helpers.async_bulk pour écrire en masse les résultats vectorisés dans l'index Elasticsearch.
   Recherche conversationnelle en ligne	Traite les requêtes utilisateur en ligne, ce qui inclut la génération de vecteurs de requête, l'analyse de requête, la récupération de fragments de document pertinents, le reclassement des résultats de recherche et la génération d'une réponse finale.	La fonction principale query_pipeline_execute exécute le workflow suivant pour traiter une requête utilisateur et renvoyer une réponse. Vectorisation de la requête. Pour plus de détails sur l'API, consultez la rubrique API d'intégration de texte. Appelez l'API d'intégration de texte pour convertir la requête utilisateur en vecteur. Utilisez la fonction text_embedding pour générer le vecteur de requête. Appelez le service d'analyse de requête. Pour plus de détails, consultez la rubrique API d'analyse de requête. Ce service identifie l'intention de l'utilisateur et génère des questions similaires en analysant l'historique des conversations. Recherche de fragments intégrés. Pour plus de détails sur le service, consultez la rubrique Utiliser la fonctionnalité de recherche k-nearest neighbor (kNN) d'Elasticsearch. Utilisez Elasticsearch pour récupérer les fragments de document de l'index qui sont similaires au vecteur de requête. Utilisez l'API search d'AsyncElasticsearch combinée à une requête kNN pour effectuer une recherche de similarité. Appelez le service de reclassement. Pour plus de détails, consultez la rubrique API de reclassement. Appelez l'API du service de reclassement pour noter et trier les fragments récupérés. Utilisez la fonction documents_ranking pour noter et trier les documents en fonction de la requête utilisateur. Générez une réponse avec le grand modèle de langage. Pour plus de détails sur l'API, consultez la rubrique API de génération de réponse. Appelez le service LLM en utilisant la fonction llm_call avec les résultats de récupération et la requête utilisateur pour générer la réponse finale.

   Sous Code Query, sélectionnez Document processing flow et Online Q & A Process, puis cliquez sur Copy Code ou Download File pour enregistrer le code localement.

Étape 2 : Configurer et tester le pipeline

Après avoir téléchargé le code dans deux fichiers locaux, par exemple offline.py et online.py, configurez les paramètres clés suivants.

# AI Search Open Platform 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"}

Après avoir configuré les paramètres, exécutez d'abord le script offline.py, puis le script online.py dans un environnement Python 3.7 ou ultérieur pour tester les résultats.

Si le document de la base de connaissances est Présentation d'AI Search Open Platform, posez la question suivante : Que peut faire AI Search Open Platform ?

Vous devriez voir la sortie suivante :

• Résultat du traitement hors ligne des documents

  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}

• Résultat de la recherche conversationnelle en ligne

  /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 AI Search Open Platform do?
  Final answer from the large model:  AI Search Open Platform provides intelligent search services that power the core search functions for Alibaba's businesses, including Taobao and Tmall, and offers intelligent search solutions to external clients across various industries. It features industry-specific query semantic understanding and machine learning ranking algorithms to help developers build high-quality intelligent search services.
  The platform is suitable for a wide range 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
  AI Search Open Platform focuses on intelligent search and Retrieval-Augmented Generation (RAG) scenarios, providing component-based services and flexible calling mechanisms. It has built-in services for document parsing, document chunking, text embedding, retrieval, reranking, and large language models, enabling a one-stop, flexible development experience for AI search applications.
  Process finished with exit code 0

• Fichiers source

  offline.py

  # RAG offline pipeline - Elasticsearch engine
  # Environment requirements:
  # Python 3.7 or later
  # Elasticsearch cluster 8.5 or later. If using Alibaba Cloud Elasticsearch, you must activate the service and configure the IP address whitelist beforehand. See https://www.alibabacloud.com/help/en/elasticsearch/latest/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"}
  
  # Elasticsearch configuration
  es_host = 'http://es-cn-xxx.public.elasticsearch.aliyuncs.com:9200'
  es_auth = ('elastic', 'xxx')
  
  # Input document URL. The example uses the product documentation for AI Search Open Platform.
  document_url = "https://www.alibabacloud.com/help/en/open-search/search-platform/product-overview/introduction-to-search-platform"
  
  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 it exists.
      if await es.indices.exists(index=index_name):
          await es.indices.delete(index=index_name)
  
      # Create a vector index. Specify the `emb` field as dense_vector, `content` as text, and `source_doc` as keyword.
      index_mappings = {
          "mappings": {
              "properties": {
                  "emb": {
                      "type": "dense_vector",
                      "index": True,
                      "similarity": "cosine",
                      "dims": 1536  # Modify the dimension based on the output of the text embedding model.
                  },
                  "content": {
                      "type": "text"
                  },
                  "source_doc": {
                      "type": "keyword"
                  }
              }
          }
      }
      await es.indices.create(index=index_name, body=index_mappings)
  
      # Bulk-upload the embedding results to the newly created index.
      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 successful upload.
      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):
  
      # Initialize the AI Search Open 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 chunking
      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 chunking results. For image chunks, the text content is extracted 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 per request.
      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 to the Elasticsearch storage engine.
      await write_to_es(doc_list)
  
  if __name__ == "__main__":
      # Run the asynchronous task.
      #    import nest_asyncio # If running in a Jupyter notebook, uncomment these two lines.
      #    nest_asyncio.apply() # If running 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 calling method
            

  online.py

  # RAG online pipeline - Elasticsearch engine
  # Environment requirements:
  # Python 3.7 or later
  # Elasticsearch cluster 8.5 or later. If using Alibaba Cloud Elasticsearch, you must activate the service and configure the IP address whitelist beforehand. See https://www.alibabacloud.com/help/en/elasticsearch/latest/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
  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"
  }
  
  # Elasticsearch configuration
  es_host = 'http://es-cn-xxx.public.elasticsearch.aliyuncs.com:9200'
  es_auth = ('elastic', 'xxx')
  
  # User query:
  user_query = "What can AI Search Open 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
  
  # Initialize the AI Search Open 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'
      # Vectorize the query.
      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 (ef_search)
      }
  
      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 conversational search pipeline. The input is the user's 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: Document retrieval
      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 retrieved 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 large language model to generate an answer.
      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 provided information contains multiple independent documents, each enclosed in <article> and </article> tags. Information:\n'''{docs}'''
                                           \n\nBased on the information provided above, answer the user's question in a detailed and organized manner. Ensure your answer fully addresses the question and correctly uses the provided information. If the information is insufficient to answer the question, state "The question cannot be answered based on the provided information." Do not use any information outside of the provided context. Ensure that every statement in your answer is supported by the context. Please answer in English.
                                           \nQuestion: '''{user_query}'''""""")
      ]
      response = await ops_client.get_text_generation_async(workspace_name, service_id_config["llm"],
                                                            GetTextGenerationRequest(messages=messages))
      print("Final answer from the large model: ", response.body.result.text)
  
  if __name__ == "__main__":
      # Run the asynchronous task.
      #    import nest_asyncio # If running in a Jupyter notebook, uncomment these two lines.
      #    nest_asyncio.apply() # If running in a Jupyter notebook, uncomment these two lines.
      asyncio.run(query_pipeline_execute())
            

FAQ

Lors de l'exécution du code, vous pouvez voir un message « Unclosed connector » dû à des ressources non libérées à temps. Vous pouvez ignorer ce message en toute sécurité.