OpenSearch:Parse and embed multimodal data

Document parsing

Image parsing

Choose between VLM-based and OCR-based parsing depending on your accuracy and cost requirements.

Document chunking

Text embedding

Choose an embedding model based on your language support needs, maximum input length, and desired vector dimensions.

Text sparse embedding

Text sparse embedding converts text into sparse vectors that occupy less storage space. Sparse vectors express keywords and frequently used term information. Combine sparse and dense vectors in a hybrid search to improve retrieval performance.

Pipeline execution flow

The pipeline code executes the following stages in sequence. Call the main function document_pipeline_execute to run the entire pipeline. You can specify the document to process by using a document URL or Base64-encoded file.

1. Parse the document or image. For more information, see Document content parsing and Image content extraction.

   • Call the asynchronous operation for document parsing to extract content from a document URL or Base64-encoded file.

   • Call the asynchronous operation for image parsing to extract content from an image URL or Base64-encoded file.

2. Chunk the document. For more information, see Document chunking.

   • Call the document chunking operation to segment the parsed document based on a specific policy.

   • Call the document_split function to segment the document. This process includes text segmentation and rich text parsing.

3. Generate text embeddings. For more information, see Text embedding and Sparse text embedding.

   • Call the text embedding operation to convert the chunked data into dense vectors.

   • Call the text sparse embedding operation to convert the chunked data into sparse vectors. If you need to perform content retrieval later, you can write the vectors to a search engine.

On the Code Query tab, click Text Parsing and Vectorization. In the code editor, click Copy Code or Download File to download the code to your device.

Sample code: Document parsing and embedding

# Multimodal data processing solution

# Requirements:
# Python 3.8.1 or later
# 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"
}

# Specify the document URL.
# In this example, the OpenSearch product description document is used.
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):
    """Poll until the document analysis task completes or fails."""
    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):
    """Check whether the URL points to a supported image format."""
    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):
    """Parse an image URL and return the extracted text content."""
    try:
        print("image analyze: " + url)
        if url.startswith("//"):
            url = "https:" + url
        if not is_analyzable_url(url):
            print(url + " is unanalyzable.")
            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):
    """Split a list into batches of the specified 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
):
    # Initialize the AI Search Open Platform client.
    config = Config(
        bearer_token=api_key,
        endpoint=aisearch_endpoint,
        protocol="http"
    )
    ops_client = Client(config=config)

    # Stage 1: Parse the document or image.
    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

    # Stage 2: Chunk the document.
    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))
    )

    # Stage 3: Generate text embeddings.
    # Extract the chunking results. For image chunks, use the image parsing
    # service to extract text before embedding.
    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 vectors can be generated 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 pipeline.
    #    import nest_asyncio  # Uncomment if running in Jupyter Notebook.
    #    nest_asyncio.apply()  # Uncomment if running in Jupyter Notebook.
    asyncio.run(document_pipeline_execute(document_url))
    # You can also use a Base64-encoded file to specify the document:
    # asyncio.run(document_pipeline_execute(
    #     document_base64="eHh4eHh4eHg...", file_name="attention.pdf"
    # ))