1. Run the following Python script on the development machine to create a new Lance dataset and convert the image data into the Lance format of the Lance dataset, writing it into the new Lance dataset. The core sample code is as follows.

   # Please run the following code on the LAS development machine
   import os
   
   """
   Convert to Lance Dataset
   """
   import daft
   from daft.io import CreateLasDatasetOptions, IOConfig, LanceWriteOptions
   from daft.las.io import TOSConfig
   
   # You need to set the image dataset name and configure the TOS path for the new Lance dataset
   dataset_name = "las_dataset_image"
   lance_tos_dir = f"tos://las-doctest/image2lance/las_dataset_image_lance.lance"
   LANCE_DATASET_FORMAT = "lance"
   
   # Create the relevant configuration
   io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())
   write_options = LanceWriteOptions(io_config=io_config, uri=lance_tos_dir)
   create_ds_options = CreateLasDatasetOptions(
       nick_name="daft_test_lance_write",
       privacy="public",
       description="This is my dataset",
   )
   
   # Read the original data
   df = daft.read_las_dataset(name=dataset_name)
   
   Create here: The naming rule for the new Lance dataset is to use the image dataset name with the _lance suffix appended
   new_dataset_name = dataset_name + "_lance"
   df.write_las_dataset(
       name=new_dataset_name,
       format=LANCE_DATASET_FORMAT,
       write_options=write_options,
       create_ds_options=create_ds_options,
   )
   
   # Read the converted new dataset
   df = daft.read_las_dataset(name=new_dataset_name)
   print("\n\nnew dataset:")
   df.show()
   

You need to modify the following parameters.

• dataset_name: Set as the name of the image dataset. In this example, it is las_dataset_image
• lance_tos_dir: Set to the TOS path of the newly converted Lance dataset prepared in the previous steps.

Other parameters can remain as the sample values.

new dataset:
ERROR:daft_local_execution.runtime_stats:RuntimeStatsManager finished with active nodes {0}
╭────────────────────────────────┬────────┬──────────╮
│ image                          ┆ size   ┆ num_rows │
│ ---                            ┆ ---    ┆ ---      │
│ String                         ┆ Int64  ┆ Int64    │
╞════════════════════════════════╪════════╪══════════╡
│ s3://las-doctest/images/anima… ┆ 15390  ┆ None     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┤
│ s3://las-doctest/images/anima… ┆ 240478 ┆ None     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┤
│ s3://las-doctest/images/anima… ┆ 15008  ┆ None     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┤
│ s3://las-doctest/images/anima… ┆ 11645  ┆ None     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┤
│ s3://las-doctest/images/anima… ┆ 9909   ┆ None     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┤
│ s3://las-doctest/images/anima… ┆ 14252  ┆ None     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┤
│ s3://las-doctest/images/anima… ┆ 3226   ┆ None     │
├╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌┼╌╌╌╌╌╌╌╌╌╌┤
│ s3://las-doctest/images/anima… ┆ 878304 ┆ None     │
╰────────────────────────────────┴────────┴──────────╯

(Showing first 8 rows)

The converted Lance dataset contains the fields "image", "size", and "num_rows". Specifically:

• image field: The TOS path (in s3 format) of each image data in the current Lance dataset.
• size field: The size of each image.

If an ERROR prompt appears in the output but the dataset data is returned normally, the ERROR prompt can be ignored.

Region settings
This example uses the North China 2 (Beijing) region.
export REGION="cn-beijing"

# Authentication AK environment variables
Access authentication keys (SK and AK) for TOS and LAS
export LAS_TOS_SECRET_KEY="<your_sk>"
export LAS_TOS_ACCESS_KEY="<your_ak>"

# TOS access-related environment variables
# Access endpoints of TOS
export LAS_TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"
# TOS endpoints (for TOSConfig)
export TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"

# Fixed parameters of the dataset can also be set as environment variables
export DATASET_NAME="las_dataset_image_lance"

This step adds:

• DATASET_NAME: Set as the name of the newly converted Lance dataset from the previous step for easy reference in subsequent task code.

import os
# Import the values of the following parameters from environment variables
LAS_TOS_ACCESS_KEY = os.environ["LAS_TOS_ACCESS_KEY"] 
LAS_TOS_SECRET_KEY = os.environ["LAS_TOS_SECRET_KEY"] 
TOS_ENDPOINT = os.environ["TOS_ENDPOINT"] 
REGION = os.environ["REGION"]
dataset_name = os.environ["DATASET_NAME"]

import pyarrow as pa
import tos
from daft.io.lance import merge_columns
from daft.io import IOConfig
from daft.las.io import TOSConfig

# Read image data from tos path
def read_tos_path(path: str, tos_client: tos.TosClientV2):
    if (not path) or path.strip() == "":
        return None

    if path.startswith("tos://"):
        path = path[len("tos://") :]

    if path.startswith("s3://"):
        path = path[len("s3://") :]

    bucket, path = path.split("/", 1)
    obj = tos_client.get_object(bucket=bucket, key=path)
    return obj.read()

def get_create_assign_raw_image_func(input_col: str, output_col: str):
    def create_assign_raw_image(batch: pa.RecordBatch) -> pa.RecordBatch:
        paths = batch.column(input_col).to_pylist()
        tos_client = tos.TosClientV2(LAS_TOS_ACCESS_KEY, LAS_TOS_SECRET_KEY, TOS_ENDPOINT, REGION)

        image_data_list = []
        for path in paths:
            data = read_tos_path(path, tos_client)
            image_data_list.append(data)

        image_data_array = pa.array(image_data_list, type=pa.binary())
        new_batch = pa.RecordBatch.from_arrays([image_data_array], names=[output_col])

        Only return new columns
        return new_batch

    return create_assign_raw_image

print("正在将图片原始数据存入 lance 数据集中 ...")

Modify lance_tos_dir and the transform parameter in merge_columns in the following code
io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())
lance_tos_dir = f"tos://las-doctest/image2lance/{dataset_name}.lance"
merge_columns(
    uri=lance_tos_dir.replace("tos://", "s3://"),
    io_config=io_config,
    transform=get_create_assign_raw_image_func("image", "raw_image"),
)

print("图片原始数据已存入 lance 数据集")

You need to modify the following parameters in the sample code according to your actual situation:

• lance_tos_dir: Modify the TOS path of the Lance dataset to your actual path.
• transform in merge_columns: A function is called, and the input parameters "image" and "raw_image" can be modified as needed:
  • image: The TOS path field in the Lance dataset
  • raw_image: A custom new field name. After the image is processed into binary data, it will be written into this field.

To delete unnecessary fields or data from a dataset caused by accidental operations, refer to the following content: Reference: Common dataset operation examples.

# Region settings
This example uses the North China 2 (Beijing) region.
export REGION="cn-beijing"

# AK authentication-related environment variables
# Access the authentication SK and AK of TOS and LAS
export LAS_TOS_SECRET_KEY="<your_sk>"
export LAS_TOS_ACCESS_KEY="<your_ak>"
# LAS API Key required for invoking LAS operators using the online method (API)
export LAS_API_KEY="464812df-2dfb-41f6-********"

# TOS access-related environment variables
TOS's access endpoints
export LAS_TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"
TOS endpoint (for TOSConfig)
export TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"

# Fixed parameters of the dataset can also be set as environment variables
export DATASET_NAME="las_dataset_image_lance"
export TOS_LANCE_DIR="tos://las-doctest/image2lance/las_dataset_image_lance.lance"

Test configuration
# Sample image URL for testing
export TEST_IMAGE_URL="https://ark-project.tos-cn-beijing.volces.com/images/view.jpeg"

This step adds:

• LAS_API_KEY: When invoking LAS operators online, LAS_API_KEY is required for authentication. You can set it to the actual value obtained in the above "Preparation" node.
• TOS_LANCE_DIR: Set the fixed value of the TOS path for the Lance dataset as an environment variable to facilitate referencing in task code.
• TEST_IMAGE_URL: A test image for subsequent operator capability testing. You can use the URL in the example directly or modify it to another test image URL.

Run the following script on the development machine.

#This code segment requires you to execute "source env.sh" first. The env.sh file is located in the "Configure environment variables" section.
# Import necessary libraries
import os

import daft
import lance
import pandas as pd
import pyarrow as pa
from daft import col
from daft.io import IOConfig
from daft.io.object_store_options import io_config_to_storage_options

from daft.las.functions.ark_llm.doubao_embedding_vision import DoubaoEmbeddingVision
from daft.las.functions.udf import las_udf
from daft.las.io import TOSConfig
from daft.las.infra.las_dataset import LasDatasetClient, LasDatasetConfig

# Get configuration from environment variables
TOS_LANCE_DIR = os.getenv("TOS_LANCE_DIR")  # Lance dataset path
DATASET_NAME = os.getenv("DATASET_NAME")    # Dataset namename

# Batch processing parameters
BATCH_SIZE  = int(os.getenv("BATCH_SIZE",  "10"))  # Maximum number of rows per operation
MAX_CYCLES  = int(os.getenv("MAX_CYCLES",  "5"))   # Maximum batch count

# Column name configuration
IMAGE_EMBEDDING_NEW_COLUMN = "image_embedding"    # New embedding column name
DOUBAO_LLM_COLUMN          = "doubao_llm_result"  # Doubao LLM result column name


# Set environment variables related to TOS object storage
os.environ["LAS_TOS_ACCESS_KEY"] = os.getenv("LAS_TOS_ACCESS_KEY")
os.environ["LAS_TOS_SECRET_KEY"] = os.getenv("LAS_TOS_SECRET_KEY")
os.environ["TOS_ENDPOINT"]       = os.getenv("LAS_TOS_ENDPOINT")

Configure IO settings for accessing TOS object storage
io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())

def update_image_embedding():
    # Convert TOS path to S3-compatible path
    lance_path = TOS_LANCE_DIR.replace("tos://", "s3://")

    Check whether the embedding column exists; if not, create it
    storage_options = io_config_to_storage_options(io_config, lance_path)
    lance_ds = lance.dataset(uri=lance_path, storage_options=storage_options)
    existing_columns = [field.name for field in lance_ds.schema]

    column_exists = IMAGE_EMBEDDING_NEW_COLUMN in existing_columns
    print(f"{IMAGE_EMBEDDING_NEW_COLUMN} 列存在: {column_exists}")

    if not column_exists:
        print(f"创建 {IMAGE_EMBEDDING_NEW_COLUMN} 列...")
        vec_type = pa.list_(pa.float64())

        @lance.batch_udf()
        def add_empty_column(batch):
            null_array = pa.array([None] * batch.num_rows, type=vec_type)
            return pa.RecordBatch.from_arrays([null_array], names=[IMAGE_EMBEDDING_NEW_COLUMN])

        lance_ds.add_columns(add_empty_column)
        print(f"成功创建 {IMAGE_EMBEDDING_NEW_COLUMN} 列")

    # Read the complete dataset and filter for rows to be processed
    df = daft.read_lance(lance_path, io_config=io_config)
    all_data = df.collect().to_pandas()

    if IMAGE_EMBEDDING_NEW_COLUMN in all_data.columns:
        pending_data = all_data[all_data[IMAGE_EMBEDDING_NEW_COLUMN].isna()].copy()
        pending_data = pending_data.reset_index(drop=True)
        print(f"跳过已有 embedding 的 {len(all_data) - len(pending_data)} 行")
    else:
        pending_data = all_data.copy()

    print(f"待处理图片总数: {len(pending_data)}")

    process_limit = MAX_CYCLES * BATCH_SIZE
    pending_data = pending_data.iloc[:process_limit].copy()
    print(f"本次实际处理: {len(pending_data)} 张图片")

    if pending_data.empty:
        print("所有图片已处理完成，无需继续")
        return

    pending_data[DOUBAO_LLM_COLUMN] = (
        pending_data[DOUBAO_LLM_COLUMN]
        .fillna("图片内容")
        .replace("", "图片内容")
    )

    # 2. Build Daft DataFrame and invoke operators to batch compute embedding
    pending_daft_df = daft.from_pandas(pending_data)

    pending_daft_df = pending_daft_df.with_column(
        IMAGE_EMBEDDING_NEW_COLUMN,
        las_udf(
            DoubaoEmbeddingVision,
            construct_args={
                "image_format": "jpeg",   # Adjust image format as needed
                "source_type":  "url",    The # image column stores URL
            },
        )(col("image"), col(DOUBAO_LLM_COLUMN)),  # image column + text column → multimodal joint vector
    )

    result_df = pending_daft_df.collect().to_pandas()
    # ------------------------------------------------------------------

    # Filter out rows where the operator call failed (returned None)
    valid_mask = result_df[IMAGE_EMBEDDING_NEW_COLUMN].notna()
    valid_df   = result_df[valid_mask].copy()
    print(f"成功生成 embedding: {len(valid_df)} 行，失败: {(~valid_mask).sum()} 行")

    if valid_df.empty:
        print("没有成功生成的 embedding，跳过更新")
        return

    # Update the Lance dataset (consistent with the original logic)
    try:
        lance_ds = lance.dataset(uri=lance_path, storage_options=storage_options)

        update_table = pa.Table.from_pandas(valid_df, preserve_index=False)
        update_table = update_table.cast(lance_ds.schema)

        lance_ds.merge_insert("image").when_matched_update_all().execute(update_table)
        print("数据集更新成功")

    except Exception as e:
        print(f"更新数据集失败: {e}")
        print("尝试备用方案（内存更新，不写回）...")
        try:
            current_df = daft.read_lance(lance_path, io_config=io_config).collect().to_pandas()
            for _, row in valid_df.iterrows():
                mask = current_df["image"] == row["image"]
                idx  = current_df[mask].index
                if len(idx) > 0:
                    current_df.at[idx[0], IMAGE_EMBEDDING_NEW_COLUMN] = row[IMAGE_EMBEDDING_NEW_COLUMN]
            print("备用方案：内存已更新，但未写回数据集")
        except Exception as e2:
            print(f"备用方案也失败: {e2}")

    print(f"\n处理完成！共处理了 {len(valid_df)} 张图片")

# Program entry point
if __name__ == "__main__":
    update_image_embedding()

You need to modify the following parameters in the sample code as needed:

• BATCH_SIZE & MAX_CYCLES: Set the number of rows and batches of images processed each time the script is executed.
• IMAGE_EMBEDDING_NEW_COLUMN: Customize a column name; the vectorization results will be written to this column later.
• DOUBAO_LLM_COLUMN: Configure this as the Lance dataset column where the above image content understanding results are written.
• If you are using your own test image set, you need to modify the values of las_udf when performing image-text vectorization according to your actual situation, including image_format (image format) and source_type (type of image source).

import daft
import os
from PIL import Image
import io
import matplotlib.pyplot as plt

# ✅ Use UDF to implement string containment check
@daft.udf(return_dtype=daft.DataType.bool())
def contains_keyword(col, keyword):
    return [keyword in (s or "") for s in col.to_pylist()]

def query_pic_with_word(dataset_name, query_word):
    df = daft.read_las_dataset(name=dataset_name)

    keyword = "袋鼠"
    # ✅ Filter using UDF
    result = df.where(contains_keyword(df["doubao_llm_result"], keyword))

    result_data = result.collect()

    for i, row in enumerate(result_data.to_pylist()):
        print(f"URL: {row['image']}")

        if 'raw_image' in row and row['raw_image'] is not None:
            image_data = row['raw_image']
            raw_image = Image.open(io.BytesIO(image_data))
            plt.figure(figsize=(8, 6))
            plt.imshow(raw_image)
            plt.axis('off')
            plt.show()

        if 'doubao_llm_result' in row and row['doubao_llm_result']:
            print(f"描述: {row['doubao_llm_result']}")

        print("-" * 50)

def main():
    dataset_name = "las_dataset_image_lance"
    query_word = "帮我查询袋鼠图片"
    query_pic_with_word(dataset_name, query_word)

if __name__ == "__main__":
    main()

The sample code for image search in this document is used to search for images related to "kangaroo" in the Lance dataset.
In the main function:

• dataset_name: The name of the Lance dataset to be queried.
• query_word: The natural language description to search for.

import numpy as np
from daft.las.functions.ark_llm.doubao_embedding_vision import DoubaoEmbeddingVision
from daft.las.functions.udf import las_udf
from daft import col
import daft
import os
from PIL import Image
import io
import matplotlib.pyplot as plt


# ✅ Use UDF to implement string containment check
@daft.udf(return_dtype=daft.DataType.bool())
def contains_keyword(col, keyword):
    return [keyword in (s or "") for s in col.to_pylist()]


def query_pic_with_word(dataset_name, query_word):
    df = daft.read_las_dataset(name=dataset_name)


    keyword = "袋鼠"
    # ✅ Filter using UDF
    result = df.where(contains_keyword(df["llm_result03"], keyword))


    result_data = result.collect()


    for i, row in enumerate(result_data.to_pylist()):
        print(f"URL: {row['image']}")


        if 'raw_image' in row and row['raw_image'] is not None:
            image_data = row['raw_image']
            raw_image = Image.open(io.BytesIO(image_data))
            plt.figure(figsize=(8, 6))
            plt.imshow(raw_image)
            plt.axis('off')
            plt.show()


        if 'llm_result03' in row and row['llm_result03']:
            print(f"描述: {row['llm_result03']}")


        print("-" * 50)


def main():
    dataset_name = "las_dataset_image_lance"
    query_word = "帮我查询袋鼠图片"
    query_pic_with_word(dataset_name, query_word)


def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))


def generate_image_embedding(image_url):
    """使用算子生成图片embedding（直接传URL，无需下载转base64）"""
    df = daft.from_pydict({"image_path": [image_url]})
    df = df.with_column(
        "embedding",
        las_udf(
            DoubaoEmbeddingVision,
            construct_args={
                "image_format": "jpeg",
                "source_type": "url",
            },
        )(col("image_path")),
    )
    result = df.collect().to_pydict()
    return result["embedding"][0]


def query_pic_with_pic(dataset_name, query_pic_url):
    """
    以图片向量的方式查询相似图片并显示


    参数:
        dataset_name: LAS数据集名称
        query_pic_url: 查询的图片URL
    """
    # Download the query image (for display only)
    from urllib.parse import urlparse
    import tos


    parsed = urlparse(query_pic_url)
    bucket = parsed.netloc
    key = parsed.path.lstrip('/')
    client = tos.TosClientV2(os.environ["LAS_TOS_ACCESS_KEY"], os.environ["LAS_TOS_SECRET_KEY"], "tos-cn-beijing.volces.com", "cn-beijing")
    resp = client.get_object(bucket=bucket, key=key)
    query_image_data = resp.read()
    query_image = Image.open(io.BytesIO(query_image_data))


    print("正在生成查询图片的embedding...")
    query_embedding = generate_image_embedding(query_pic_url)


    if query_embedding is None:
        print("无法生成查询图片的embedding")
        return


    print(f"成功生成embedding，维度: {len(query_embedding)}")


    # Read the LAS dataset
    df = daft.read_las_dataset(name=dataset_name)


    # Retrieve all data
    all_data = df.collect()
    all_rows = all_data.to_pylist()


    # Display the query image
    print("查询图片:")
    plt.figure(figsize=(8, 6))
    plt.imshow(query_image)
    plt.title("查询图片")
    plt.axis('off')
    plt.show()
    print("-" * 50)


    # Calculate the similarity between all images and the query image
    similarities = []
    for i, row in enumerate(all_rows):
        if row['image_embedding'] is not None:
            similarity = cosine_similarity(query_embedding, row['image_embedding'])
            similarities.append((similarity, i, row))


    # Sort by similarity, select the top 3 with similarity greater than 40%
    similarities.sort(key=lambda x: x[0], reverse=True)
    top_3 = [s for s in similarities[:3] if s[0] > 0.40]


    if not top_3:
        print("未找到相似度大于40%的图片")
        return


    # Display the results
    for similarity, idx, row in top_3:
        print(f"URL: {row['image']}")
        print(f"相似度: {similarity:.4f}")


        if 'raw_image' in row and row['raw_image'] is not None:
            image_data = row['raw_image']
            raw_image = Image.open(io.BytesIO(image_data))
            plt.figure(figsize=(8, 6))
            plt.imshow(raw_image)
            plt.axis('off')
            plt.show()


        if 'llm_result03' in row and row['llm_result03']:
            print(f"描述: {row['llm_result03']}")


        print("-" * 50)


if __name__ == "__main__":
    dataset_name = "las_dataset_image_lance"
    query_pic_url = "tos://las-doctest/images/animals/pic_search_1000_images/0009fc27d9.jpg"


    query_pic_with_pic(dataset_name, query_pic_url)

The sample code for image search in this document is used to search for images in the Lance dataset based on a given input image, calculate the cosine similarity, and return the top 3 images with the highest similarity.
In the main function:

• dataset_name: The name of the Lance dataset to be queried.
• query_pic_url: The URL of the image to search for. In this example, an image from TOS is used as input, configured as the URL in the TOS path format for the corresponding image.