OpenSearch:Build a RAG-based conversational search

How it works

Retrieval-Augmented Generation (RAG) combines information retrieval with large language models (LLMs) to improve the accuracy and relevance of generated content. A RAG system retrieves the most relevant information from an external knowledge base and provides it as context to the LLM, enabling the model to generate precise answers grounded in up-to-date or domain-specific data rather than relying solely on its training data.

Use cases

Conversational search over a knowledge base suits use cases such as enterprise knowledge retrieval and domain-specific Q&A. By applying RAG and LLMs to your knowledge base documents, the system can understand complex natural language queries and help users quickly find information across document formats including PDF, Word, tables, and images.

In the conversational search interface, a user submits a question in natural language. The system returns a structured answer and suggests follow-up questions for further exploration.

Prerequisites

• The AI Search Open Platform service is activated. For more information, see Activate the service.

• Obtain a service endpoint and authentication credentials. For more information, see Obtain a service endpoint and Manage API keys.

  AI Search Open Platform supports both public and VPC endpoints. Cross-region calls are supported through VPC endpoints. Currently, VPC access is available in the China (Shanghai), China (Hangzhou), China (Shenzhen), China (Beijing), China (Zhangjiakou), and China (Qingdao) regions.

  On the API Keys page, a message at the top reads, "API keys are used for service call permissions. Please keep them secure. If an API key is compromised, disable it immediately. You can have up to 10 API keys enabled at a time." The Access Domain section displays the Public API Domain and Private API Domain, both of which support HTTPS access.

• Create an Alibaba Cloud Elasticsearch cluster of version 8.5 or later. For more information, see Create an Alibaba Cloud Elasticsearch cluster. When accessing the cluster over the public network or a VPC, you must add the IP address of your device to the cluster's IP address whitelist. For more information, see Configure a public or private IP address whitelist for an Elasticsearch cluster.

• You have a Python 3.7 or later environment with the aiohttp 3.8.6 and elasticsearch 8.14 packages installed.

Build the RAG development pipeline

Step 1: Select services and download code

Based on your knowledge base and business requirements, select the algorithm services and development framework for your RAG pipeline. This topic uses the Python SDK as an example.

1. Log on to the AI Search Open Platform console.

2. Select the China (Shanghai) region, switch to AI Search Open Platform, and then select your target workspace.

   Note

   • AI Search Open Platform is available only in the China (Shanghai) and Germany (Frankfurt) regions.

   • Users in the China (Hangzhou), China (Shenzhen), China (Beijing), China (Zhangjiakou), and China (Qingdao) regions can use a VPC endpoint to call AI Search Open Platform services across regions.

3. In the left-side navigation pane, click Scene Center. On the RAG Scene-Knowledge Base Online Q & A. card, click Enter.

4. From the drop-down lists, select the services you need based on your business requirements. You can view detailed information for each service on the Service Details tab.

   Note

   • When calling an algorithm service in the RAG pipeline via an API, you must provide the service ID (service_id). For example, the ID for the document content parsing service is ops-document-analyze-001.

   • When you switch services in the list, the service_id in the generated code updates automatically. After you download the code, you can still change the service_id to call a different service.

   Stage	Service description
   Document content parsing	Document content parsing service (ops-document-analyze-001): A general-purpose service that extracts logical structures like titles and paragraphs from unstructured documents (text, tables, and images) and outputs them in a structured format.
   Image content parsing	Image content understanding service (ops-image-analyze-vlm-001): Uses a multimodal large model to parse, understand, and recognize text from images. The extracted text can be used for image retrieval and Q&A scenarios. Image text recognition service (ops-image-analyze-ocr-001): Uses OCR to recognize text in images. The parsed text can be used for image retrieval and Q&A scenarios.
   Document chunking	Document chunking service (ops-document-split-001): A general-purpose text chunking service that splits structured data in HTML, Markdown, and TXT formats based on document paragraphs, text semantics, or specified rules. It also supports extracting code, images, and tables as rich text from documents.
   Text embedding	OpenSearch text embedding service-001 (ops-text-embedding-001): Provides multilingual (40+) text embedding. Maximum input length is 300 tokens, and the output vector dimension is 1536. OpenSearch general text embedding service-002 (ops-text-embedding-002): Provides multilingual (100+) text embedding. Maximum input length is 8,192 tokens, and the output vector dimension is 1024. OpenSearch text embedding service-Chinese-001 (ops-text-embedding-zh-001): Provides Chinese text embedding. Maximum input length is 1,024 tokens, and the output vector dimension is 768. OpenSearch text embedding service-English-001 (ops-text-embedding-en-001): Provides English text embedding. Maximum input length is 512 tokens, and the output vector dimension is 768.
   Sparse text embedding	Converts text into a sparse vector representation. Sparse vectors use less storage and typically represent keywords and term frequencies. Combine them with dense vectors for hybrid search to improve retrieval performance. OpenSearch sparse text embedding service (ops-text-sparse-embedding-001): Provides multilingual (100+) sparse text embedding. Maximum input length is 8,192 tokens.
   Query analysis	Query analysis service 001 (ops-query-analyze-001): Uses a large language model to understand user query intent and expand it with similar questions.
   Search engine	Alibaba Cloud Elasticsearch: A fully managed cloud service built on open-source Elasticsearch, 100% compatible with open-source features, with an out-of-the-box, pay-as-you-go experience. Note If you choose Alibaba Cloud Elasticsearch as the search engine, the sparse text embedding service is unavailable due to compatibility issues. We recommend using text embedding services instead. OpenSearch Vector Search Edition: A large-scale distributed vector search engine developed by Alibaba that supports various vector search algorithms with high precision and cost-effective indexing at scale. Indexes support horizontal scaling, streaming builds, real-time queries, and dynamic data updates. Note If you need to use OpenSearch Vector Search Edition, you can replace the engine configuration and code in the RAG pipeline.
   Reranking service	BGE reranker model (ops-bge-reranker-larger): Scores and sorts documents by relevance to the query, ranking them from highest to lowest, and outputs the scoring results.
   Large language model	OpenSearch-Qwen-Turbo (ops-qwen-turbo): Built on the Qwen-Turbo large language model, this service is fine-tuned with supervised learning to enhance retrieval augmentation and reduce harmful responses. Qwen-Turbo (qwen-turbo): A large-scale language model from the Qwen series that supports various languages, including Chinese and English. For more information, see Introduction to Qwen series LLMs. Qwen-Plus (qwen-plus): An enhanced version of the Qwen-Turbo large language model that supports various languages, including Chinese and English. For more information, see Introduction to Qwen series LLMs. Qwen-Max (qwen-max): A trillion-parameter, ultra-large-scale language model from the Qwen series that supports various languages, including Chinese and English. For more information, see Introduction to Qwen series LLMs.

5. After selecting your services, click After the configuration is completed, enter the code query to view and download the code.

   The code is structured into two parts that reflect the RAG pipeline's runtime flow: offline document processing and online conversational search.

   Process	Function	Description
   Offline document processing	Processes documents, which includes parsing, image extraction, chunking, embedding, and writing the results to an Elasticsearch index.	The document_pipeline_execute main function completes the following workflow. You can input documents via a URL or Base64 encoding. Document parsing. For API details, see Document Parsing API. Call the asynchronous document parsing API to extract content from a document URL or decode content from a Base64-encoded file. Use the create_async_extraction_task function to create a parsing task and the poll_task_result function to poll for the task's completion status. Image extraction. For API details, see Image Content Extraction API. Call the asynchronous image parsing API to extract content from an image URL or decode it from a Base64-encoded file. Use the create_image_analyze_task function to create an image parsing task and the get_image_analyze_task_status function to get its status. Document chunking. For API details, see Document Chunking API. Call the document chunking API to split the parsed document according to a specified strategy. Use the document_split function for both document chunking and rich text content parsing. Text embedding. For API details, see Text Embedding API. Call the text embedding API to create vector representations of the chunked text. Use the text_embedding function to calculate the embedding vector for each chunk. Write to Elasticsearch. For service details, see Use the k-nearest neighbor (kNN) search feature of Elasticsearch. Create an Elasticsearch index configuration that specifies the vector field embedding and the document content field content. Important When you create an Elasticsearch index, any existing index with the same name is deleted. To avoid accidental data loss, change the index name in the code. Use the helpers.async_bulk function to bulk-write the vectorized results to the Elasticsearch index.
   Online conversational search	Processes online user queries, which includes generating query vectors, performing query analysis, retrieving relevant document chunks, reranking search results, and generating a final answer.	The query_pipeline_execute main function completes the following workflow to process a user query and return an answer. Vectorize the query. For API details, see Text Embedding API. Call the text embedding API to convert the user query into a vector. Use the text_embedding function to generate the query vector. Call the query analysis service. For details, see Query Analysis API. This service identifies user intent and generates similar questions by analyzing the conversation history. Search for embedding chunks. For service details, see Use the k-nearest neighbor (kNN) search feature of Elasticsearch. Use Elasticsearch to retrieve document chunks from the index that are similar to the query vector. Use the search API of AsyncElasticsearch combined with a kNN query to perform similarity search. Call the reranking service. For details, see Reranking API. Call the reranking service API to score and sort the retrieved chunks. Use the documents_ranking function to score and sort documents based on the user query. Generate an answer with the large language model. For API details, see Answer Generation API. Call the LLM service, using the llm_call function with the retrieval results and user query to generate the final answer.

   Under Code Query, select Document processing flow and Online Q & A Process, and then click Copy Code or Download File to save the code locally.

Step 2: Configure and test the pipeline

After downloading the code into two local files, such as offline.py and online.py, configure the following key parameters.

# 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"}

After configuring the parameters, run the offline.py script first, followed by the online.py script, in a Python 3.7 or later environment to test the results.

If the knowledge base document is Introduction to AI Search Open Platform, ask the following question: What can AI Search Open Platform do?

You should see the following output:

• Offline document processing result

  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}

• Online conversational search result

  /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

• Source code files

  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

During code execution, you might see an "Unclosed connector" message due to resources not being released in time. You can safely ignore this message.