This practical guide demonstrates how to use LAS to build an end-to-end RAG (Retrieval-Augmented Generation) knowledge base based on private PDF documents. The entire process covers uploading original PDF documents, intelligent parsing (separating images and text), multimodal vectorization, and finally ingesting data into the ByteHouse vector database. It also provides a complete implementation for efficient vector retrieval via code.
This article breaks down the full process, including environment preparation, pipeline development for data ingestion, batch task execution, and vector retrieval validation, offering users a comprehensive and reproducible workflow for building an enterprise-level knowledge base.

Background information #

In enterprise operations and industry research, vast amounts of knowledge are often accumulated in unstructured documents such as PDF and Word files. How to efficiently extract value from these documents and combine it with large language models to provide intelligent services such as question answering, summarization, and analysis is a major focus in the current technology field. RAG technology enhances the generation capabilities of large models by retrieving external knowledge bases, which is key to solving this problem.
Building an efficient RAG system requires overcoming the following challenges:

• Complex document parsing: Documents such as PDFs often contain text, images, tables, and other elements, requiring precise separation of images and text and content extraction.
• Effective data chunking and vectorization: Long texts need to be split into semantically complete chunks and, together with images, converted into high-quality vectors (embeddings).
• Reliable storage and retrieval: A stable and efficient vector database is needed to store and manage billions of vectors and provide low-latency similarity retrieval capabilities.
• Cumbersome infrastructure setup: The entire process involves setting up and connecting multiple components, including object storage, computing resources, development environments, and databases.

The LAS Data Lake Service greatly simplifies the above process by providing one-stop multimodal data processing capabilities and an integrated development environment.

How it works #

This solution consists of two core stages: data ingestion and vector retrieval.

1. Data ingestion
   1. The goal of this stage is to process raw PDF files into searchable vector data and store them in ByteHouse.
      • PDF parsing: Use the PDFParse operator to deeply parse the PDF, separating plain text content and image files.
      • Text chunking: Use the ChunkTextSentenceSplitter operator to split the parsed long text into fixed-size, overlapping semantic chunks.
      • Multimodal embedding: Use the DoubaoEmbeddingVision operator to vectorize both text chunks and images, generating embedding vectors that represent their semantics.
      • Write to database: Use a custom ClickhouseWriter UDF to write the processed text chunks, image paths, and their corresponding vector data into the specified table in ByteHouse.
2. Vector retrieval
   1. The goal of this stage is to retrieve the most relevant document fragments from ByteHouse based on the user's query.
      • Query embedding: Use the DoubaoEmbeddingVision operator to convert the user's text query into a vector.
      • Similarity search: Perform a vector similarity search (cosineDistance) in ByteHouse to find the Top-K text or image vectors most similar to the query vector.
      • Return results: Return the original data fragments (text content or image paths) corresponding to the Top-K vectors. These can be used as context for subsequent input to large language models.

Prerequisites #

Volcano Engine account preparation #

1. To use the LAS console, you must first register a Volcano Engine account. For details, see Account Registration.
2. After successfully registering a Volcano Engine account, you must complete the relevant authentication steps. For details, see Personal Authentication and Enterprise Authentication.

Environment and resource preparation #

The following services have been enabled: AI Data Lake Service, Object Storage TOS, and ByteHouse Enterprise Edition.

Step 1: Prepare object storage (TOS)

1. Log in to the Object Storage TOS Console.
2. Go to the Bucket List and create a storage bucket (Bucket), for example, las-rag-bj.
3. Create a folder path in the bucket to store data files, for example, test.
4. Upload the prepared PDF documents to this path.

Step 2: Create and configure the ByteHouse vector database

1. Log in to the ByteHouse Enterprise Edition Console.
2. On the Cluster Management page, select Cluster List and click Create Cluster. Wait for the cluster creation to complete.
   1. Configure compute resource specifications: General-purpose (1:4) - 16vCPU 64GiB, single replica; number of nodes: 2.
   2. Configure storage resources: Single-node local storage: 100GiB, do not enable hot and cold tiering.
   3. Configure network resources: Select VPC, subnet, and security group.
   4. Key configuration: In "Advanced Features", be sure to enable the Vector Search feature.

3. Go to the Data Management page, click +Create on the left, then select Create Database to create a new database, for example, pdf_rag_db.
4. Under this database, select Create Table on the right to create a table for storing vector data. For detailed parameter configuration, see Create Database/Table.
   • Example table structure: Sort Key/Primary Key settings can be selected as needed, for example, pdf_tos_path.

5. Record the connection information of the ByteHouse cluster, which will be used in subsequent code.
   1. bh_host is obtained from ByteHouse Enterprise Edition > Cluster Management > Basic Information's Gateway.
   2. bh_port defaults to 8123.
   3. bh_user is obtained from ByteHouse Enterprise Edition > Cluster Management > Connect to Cluster.
   4. bh_password is obtained from Account Management.

Step 3: Create LAS Data Lake resources

1. Log in to the AI Data Lake Service Console.
2. In Resource Management > Queue Management, create a general-purpose queue for running the development machine.
3. Return to the AI Data Lake Service console, and in Data Processing > Development Machine, click Create Development Machine. For details on the creation operation, see Create Development Machine and Development Machine Quick Start.
   • Associate the general-purpose queue created in the previous step.
   • Paste the locally generated SSH public key (the contents of the .pub file) into the configuration.
4. Wait for the development machine to be created, and record the load balancer's public IP and the development machine's access port.

Operation process #

The basic operation process for this practice is as follows:
Step 1: Data lake pipeline development and execution
Step 2: Vector data retrieval

Step 1: Data lake pipeline development and execution #

1. Develop data lake ingestion scripts.
   Log in to the created development machine via SSH. For instructions on connecting to the development machine, refer to Remote connection to the development machine. Create a Python script (for example, pdf_to_bh.py) and write the data processing pipeline.

from __future__ import annotations

from clickhouse_connect.driver import create_client

import daft
from daft import col, lit
from daft.las.functions.ark_llm import DoubaoEmbeddingVision
from daft.las.functions.doc import PDFParse
from daft.las.functions.text import PreSignUrlForTos
from daft.las.functions.text.chunk_text_sentence_splitter import ChunkTextSentenceSplitter
from daft.las.functions.udf import las_udf
from daft.las.io.tos import TOSConfig
import os
import ray
# Add when building the image for task management; these two lines are not needed for debugging on the development machine
ray.init()
daft.context.set_runner_ray("ray://127.0.0.1:10001")
@daft.udf(return_dtype=str)
class ClickhouseWriter:
    def __init__(self):
        self.bh_client = create_client(
            host=os.getenv("bh_host", "host"),
            port=os.getenv("bh_port", "8123"),
            user=os.getenv("bh_user", "user"),
            password=os.getenv("bh_pass", "pass"),
            database=os.getenv("bh_database", "database"),
            compression="gzip",
        )
        self.table_name = os.getenv("bh_table", "bh_table")

    def __call__(self, *columns):
        try:
            data_to_insert = [col.to_pylist() for col in columns]

            rows = list(zip(*data_to_insert))
            self.bh_client.insert(
                table=self.table_name,
                data=rows,
            )
            return ["success"] * len(rows)
        except Exception as e:
            print(f"Error: {e}")
            return ["error"]


def run_pipeline(input_tos_dir, output_tos_path):
    input_s3_dir = input_tos_dir.replace("tos://", "s3://", 1)
    tos_config = TOSConfig.from_env()
    IO_CONFIG = daft.io.IOConfig(s3=tos_config.to_s3_config())

    df = daft.from_glob_path(
        f"{input_s3_dir}/*.pdf",
        io_config=IO_CONFIG,
    )

    df = df.with_column(
        "pdf_tos_path",
        col("path").apply(
            lambda p: p.replace("s3://", "tos://", 1),
            return_dtype=daft.DataType.string(),
        ),
    )

    # pre sign url for tos
    df = df.with_column(
        "input_pdf_url",
        las_udf(
            PreSignUrlForTos,
            construct_args={
                "expires": 3600,
            },
        )(col("pdf_tos_path")),
    )

    # get filename for PDFParse
    df = df.with_column(
        "filename", col("pdf_tos_path").apply(lambda p: p.split("/")[-1], return_dtype=daft.DataType.string())
    )

    # pdf parse
    df = df.with_column(
        "result",
        las_udf(
            PDFParse,
            construct_args={
                "input_type": "url",
                "output_tos_path": output_tos_path,
                "qps": 1,
            },
            concurrency=1,
            batch_size=1,
        )(col("input_pdf_url"), col("filename")),
    )

    df = df.with_column("plain_text", col("result").struct.get("parsed_plain_text"))
    df = df.with_column("image_filenames", col("result").struct.get("parsed_image_filenames"))
    df = df.with_column("parsed_markdown_tos_path", col("result").struct.get("parsed_file_path"))
    df.collect()

    # image embedding
    df_image = df.select("pdf_tos_path", "image_filenames", "parsed_markdown_tos_path").explode("image_filenames")
    df_image = df_image.with_column("image_tos_path", output_tos_path + "/images/" + col("image_filenames"))

    df_image = df_image.with_column(
        "image_http_path",
        las_udf(
            PreSignUrlForTos,
            construct_args={
                "expires": 3600,
            },
        )(col("image_tos_path")),
    )

    df_image = df_image.with_column(
        "embedding",
        las_udf(
            DoubaoEmbeddingVision,
            construct_args={
                "version": "250328",
                "image_format": "jpeg",
            },
            concurrency=1,
        )(col("image_http_path")),
    )

    df_image = (
        df_image.with_column("modality", lit("image"))
        .with_column("content", lit(""))
        .select("pdf_tos_path", "modality", "content", "image_tos_path", "parsed_markdown_tos_path", "embedding")
    )

    # text chunking and embedding
    df_text = df.with_column(
        "chunks",
        las_udf(
            ChunkTextSentenceSplitter,
            construct_args={
                "content_type": "md",
                "chunk_size": 512,
                "chunk_overlap": 50,
            },
            concurrency=1,
        )(col("plain_text")),
    )

    df_text = df_text.select("pdf_tos_path", "chunks", "parsed_markdown_tos_path").explode("chunks")

    df_text = df_text.with_column(
        "embedding",
        las_udf(
            DoubaoEmbeddingVision,
            construct_args={
                "version": "250328",
                "multimodal_type": "text",
            },
            concurrency=1,
        )(col("chunks")),
    )

    df_text = (
        df_text.select("pdf_tos_path", "chunks", "embedding", "parsed_markdown_tos_path")
        .with_column("modality", lit("text"))
        .with_columns_renamed({"chunks": "content"})
        .with_column("image_tos_path", lit(""))
        .select("pdf_tos_path", "modality", "content", "image_tos_path", "parsed_markdown_tos_path", "embedding")
    )

    df_merged = df_text.concat(df_image)
    df_merged = df_merged.filter(col("embedding").list.length() > 0)
    # write to bytehouse
    df_merged = df_merged.with_column(
        "write_to_bytehouse",
        ClickhouseWriter(*df_merged.columns),
    )

    print(df_merged.filter(col("write_to_bytehouse") == "error").count_rows())


if __name__ == "__main__":
    input_tos_dir = os.getenv("input_tos_dir", "input_tos_dir")
    output_tos_path = os.getenv("output_tos_path", "output_tos_path")
    run_pipeline(input_tos_dir, output_tos_path)

2. Debug the script on the development machine.
   In the terminal of the development machine, use the export command to set all required environment variables, then run the script for debugging. Example:

   

   # PDF parsing parameters
    export TOS_ENDPOINT="https://tos-cn-beijing.volces.com"
    export ACCESS_KEY="xxx"
    export SECRET_KEY="xx="
    export LAS_ACCESS_KEY="xxx"
    export LAS_ACCOUND_ID="xx"
    export DAFT_RUNNER="xxx"
    # bh write parameters
    export bh_host="xx"
    export bh_port="8123"
    export bh_user="xx"
    export bh_pass="xx"
    export bh_database="xx"
    export bh_table="xx"
    # file entry parameters
    export input_tos_dir="xx"
    export output_tos_path="xx"
    # startup command
    unset  LAS_BASE_URL
    python pdf_2_bh_new.py
   

3. Save as a custom image.
   After successful debugging, go to the LAS Console > Development Machine page, find the corresponding development machine, and click Save Image. Name and create the image. Wait for the image build to complete. You can go to Image Management > Custom Images to view the image build progress.
   

4. Create and execute a batch data ingestion task.

   1. Go to the Data Processing > Task Management page and click Create Task.

   2. In the Image option, select "Custom Image" and find the image saved in the previous step.

      

   3. In Parameter Settings, enter all required environment variables as key-value pairs (the same variables as set with export during debugging).

      

   4. After creating the task, click Execute in the task list, select a compute queue, configure resources, and start the task.

      1. You can choose either the Public Queue (postpaid) or a Dedicated Queue (prepaid) created under Resource Management > Queue Management > Compute Queue.
      2. Set the minimum compute parameters at startup as follows:
         • WorkCPU：3
         • WorkMemory：24
      

   5. The task will automatically pull the image and batch process all PDF files in input_tos_dir on the specified compute queue, writing the results to ByteHouse.

Step 2: Vector data retrieval #

1. Develop a vector retrieval script

Log in to the development machine again and create a new Python script (for example, bh_query.py) for performing vector retrieval.

from __future__ import annotations

from clickhouse_connect import get_client

from daft import DataType, col, from_pydict
from daft.las.functions.ark_llm import DoubaoEmbeddingVision
from daft.las.functions.udf import las_udf
import os

def make_query_func(k=3, modality=None):
    def query_fn(embedding):
        if not embedding:
            return []

        bh_client = get_client(
            host=os.getenv("bh_host", "host"),
            port=os.getenv("bh_port", "8123"),
            user=os.getenv("bh_user", "user"),
            password=os.getenv("bh_pass", "pass"),
            connect_timeout=3000000,
            send_receive_timeout=300000,
        )
        database =os.getenv("bh_database", "database")
        collection =os.getenv("bh_table", "bh_table")
        metric = "cosine"
        modality_filter = f"WHERE modality = '{modality}'" if modality else ""

        query = f"""
            SELECT pdf_tos_path, modality, content, image_tos_path
            FROM {database}.{collection}
            {modality_filter}
            ORDER BY {metric}Distance(embedding, {embedding!s})
            LIMIT {k}
            settings enable_new_ann=1, hnsw_ef_s=200
        """

        try:
            return bh_client.query(query).result_rows
        except Exception as e:
            print(f"ClickHouse query failed: {e}")
            return []

    return query_fn


if __name__ == "__main__":
    query_fn_text = make_query_func(k=2, modality="text")
    query_fn_image = make_query_func(k=2, modality="image")

    # text example
    text_list = ["春天来了", "科技在各个领域都展现出了强大的影响力"]
    df = from_pydict({"text": text_list})
    df = df.with_column(
        "embedding",
        las_udf(
            DoubaoEmbeddingVision,
            construct_args={
                "version": "250328",
                "multimodal_type": "text",
            },
        )(col("text")),
    )

    df = df.with_column(
        "topk_text_results", df["embedding"].apply(query_fn_text, return_dtype=DataType.python())
    ).with_column("topk_image_results", df["embedding"].apply(query_fn_image, return_dtype=DataType.python()))

    df.show()

2. Run the retrieval on the development machine

Similarly, first export the ByteHouse-related environment variables, then run the retrieval script. Example:

# bh read parameters
 export bh_host="xxx"
 export bh_port="xx"
 export bh_user="xxx"
 export bh_pass="xxx"
 export bh_database="xx"
 export bh_table="xx"
 
 # startup command
 python bh_query.py

You will see a table in the terminal showing the most relevant Top-K text fragments and image paths retrieved for each query text. For more technical details or customization requirements, contact the technical support team.