Multimodal data preprocessing pipeline - AI Search Open Platform - OpenSearch

Build an end-to-end multimodal data preprocessing pipeline on AI Search Open Platform to parse documents and images, chunk content, and generate text embeddings.

Use cases

The multimodal data preprocessing pipeline processes unstructured documents and images through five services: document parsing, image parsing, document chunking, text embedding, and sparse text embedding. All services are called through the AI Search Open Platform API. You pay based on actual usage.

Prerequisites

AI Search Open Platform is activated. Activate Service.
Your API endpoint and API key are ready. Query service endpoint, Manage API key.

You can call services over the Internet or through a VPC. Cross-region VPC calls are supported in the China (Shanghai), China (Hangzhou), China (Shenzhen), China (Beijing), China (Zhangjiakou), and China (Qingdao) regions.

Build a multimodal data preprocessing pipeline

Note

To simplify development, AI Search Open Platform provides four development frameworks:

Java SDK.
Python SDK.
If your application already uses the LangChain development framework, select the LangChain integration.
If your application already uses the LlamaIndex development framework, select the LlamaIndex integration.

Step 1: Select services and download the code

This procedure uses the Python SDK to build a multimodal data preprocessing pipeline.

Log on to the AI Search Open Platform console.
Select the China (Shanghai) region, switch to AI Search Open Platform, and then switch to the 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.
In the left-side navigation pane, select Scene Center. Find Multimodal Data Preprocessing - Data Parsing and Vectorization and click Enter.

From the drop-down list, select the services you need. You can view detailed information about each service on the Service Details page.

Note

Each service requires a service ID (service_id). For example, the document parsing service ID is ops-document-analyze-001.
The service_id in the generated code updates automatically when you switch services. You can also change the service_id manually after downloading the code.

Stage	Description
Document content parsing	Document Content Parsing Service (`ops-document-analyze-001`): Extracts logical structures such as titles and paragraphs from unstructured documents (text, tables, and images) and returns structured content.
Image content parsing	Image Content Understanding Service (`ops-image-analyze-vlm-001`): Uses multimodal LLMs to parse image content and recognize text for image retrieval and question answering. Image Text Recognition Service (`ops-image-analyze-ocr-001`): Uses OCR to recognize text in images for retrieval and question answering.
Document chunking	Document Chunking Service (`ops-document-split-001`): Splits structured data in HTML, Markdown, and TXT formats by paragraph, semantics, or custom rules. Also extracts code, images, and tables from rich text.
Text embedding	OpenSearch text embedding service-001 (`ops-text-embedding-001`): Supports 40+ languages. Max input: 300 tokens. Output: 1,536-dimensional vectors. OpenSearch general text embedding service-002 (`ops-text-embedding-002`): Supports 100+ languages. Max input: 8,192 tokens. Output: 1,024-dimensional vectors. OpenSearch text embedding service-Chinese-001 (`ops-text-embedding-zh-001`): Chinese-specific. Max input: 1,024 tokens. Output: 768-dimensional vectors. OpenSearch text embedding service-English-001 (`ops-text-embedding-en-001`): English-specific. Max input: 512 tokens. Output: 768-dimensional vectors.
Sparse text embedding	Converts text into sparse vector representations. Sparse vectors require less storage and capture keywords and term frequencies. Combine them with dense vectors for hybrid search to improve retrieval. OpenSearch sparse text embedding service (`ops-text-sparse-embedding-001`): Supports 100+ languages. Max input: 8,192 tokens.

After selecting the services, click After the configuration is completed, enter the code query to view and download the code. The following table describes the runtime flow of the data preprocessing pipeline.

Function

Description

Handles document processing, including document/image parsing, document chunking, and text embedding.

The document_pipeline_execute main function runs the entire pipeline. It accepts a document URL or Base64-encoded string.

Parse documents and images using the Document Parsing API and Image Content Extraction API.
- Call the asynchronous document parsing API to extract content from a document URL or Base64-encoded file.
- Call the asynchronous image parsing API to extract content from an image URL or Base64-encoded file.
Chunk documents using the Document Chunking API.
- Call the document chunking API to split the parsed document by the specified strategy.
- The document_split function splits the document and parses rich text content.
Generate text embeddings using the Text Embedding API and Sparse Text Embedding API.
- Call the text embedding API to convert the chunked data into a dense vector.
- Call the sparse text embedding API to convert the chunked data into a sparse vector, then write the results to a search engine for retrieval.

Under Code Query, select Document Parsing and Vectorization. Click Copy Code or Download File to save the code locally.

Step 2: Configure environment and test pipeline

Configure the following parameters after downloading the code.

Category	Parameter	Description
AI Search Open Platform	api_key	The API key for authentication. Manage API keys.
	aisearch_endpoint	The API endpoint. Query service endpoint. Note Remove the `http://` prefix. You can call the API over the Internet or through a VPC.
	workspace_name	The name of your workspace in AI Search Open Platform.
	service_id	The service ID. Use `service_id_config` to configure service IDs. `# 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"}`

Run the code in Python 3.8.1 or later to verify the results.

Sample output from preprocessing the Introduction to AI Search Open Platform document:

C:\Users\issuser\PycharmProjects\pythonProject630Demo\.venv\Scripts\python.exe C:\Users\issuser\PycharmProjects\pythonProject3\.venv\Lib\site-packages\alibabacloud_searchplat2...
document_analyze task_id:f6dd92de-01e4-46a2-96c4-b80e7b120239
document_analyze done
document-split done, chunks count: 6 rich text count:9
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  unanalysable.
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/0517992271/p821030.png
image analyze ://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif
https://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif is  unanalysable.
image analyze ://img.alicdn.com/tfs/TB1..50QpXXXXX7XpXXXXXXXXXX-40-40.png
image analyze :https://img.alicdn.com/tfs/TB1UdKEM6TpK1RjSZKPXXa3UpXa-256-256.png
image analyze :https://img.alicdn.com/tfs/TB1A0dINW6qK1RjSZFmXXX0PFXa-258-258.jpg
image analyze ://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif
https://gw.alicdn.com/tfs/TB16xwdSXXXXXa.aXXXXXXXXXXX-65-70.gif is  unanalysable.
image analyze ://img.alicdn.com/tfs/TB1..50QpXXXXX7XpXXXXXXXXXX-40-40.png
text-embedding done.
Process finished with exit code 0

Complete pipeline code:

# Multimodal data processing pipeline
# Requirements:
# Python version: 3.8.1 or later
# Package requirements:
# pip install alibabacloud_searchplat20240529
# AI Search Open Platform configuration
aisearch_endpoint = "xxx.platform-cn-shanghai.opensearch.aliyuncs.com"
api_key = "OS-xxx"
workspace_name = "default"
service_id_config = {"document_analyze": "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"}
# Input document URL. The example document is the introduction to 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 operator import attrgetter
from typing import List
from Tea.exceptions import TeaException, RetryError
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, \
    GetImageAnalyzeTaskStatusResponse, CreateImageAnalyzeTaskRequest, GetImageAnalyzeTaskStatusRequest, \
    CreateImageAnalyzeTaskRequestDocument, CreateImageAnalyzeTaskResponse
async def poll_doc_analyze_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:
            print("error: " + response.body.result.error)
            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  unanalysable.")
            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 document_pipeline_execute(document_url: str = None, document_base64: str = None, file_name: str = None):
    # Create an AI Search Open Platform client.
    config = Config(bearer_token=api_key,endpoint=aisearch_endpoint,protocol="http")
    ops_client = Client(config=config)
    # Step 1: Document/Image 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["document_analyze"],request=document_analyze_request)
    print("document_analyze task_id:" + document_analyze_response.body.result.task_id)
    extraction_result = await poll_doc_analyze_task_result(ops_client, document_analyze_response.body.result.task_id, service_id_config["document_analyze"])
    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 image parsing service extracts text content.
    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"]
                )
    # A maximum of 32 embeddings can be calculated per request.
    chunk_size = 32  
    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.")
if __name__ == "__main__":
    # Run the asynchronous task.
    #    import nest_asyncio # Uncomment the following two lines if you run this in a Jupyter notebook.
    #    nest_asyncio.apply() # Uncomment the following two lines if you run this in a Jupyter notebook.
    asyncio.run(document_pipeline_execute(document_url))
    # asyncio.run(document_pipeline_execute(document_base64="eHh4eHh4eHg...", file_name="attention.pdf")) # Alternative method of calling