1. 在开发机中运行以下Python脚本，创建一个新的 Lance 数据集，并将图片数据转为Lance 数据集的 Lance 格式写入新的 Lance 数据集，核心示例代码如下。

   # 请在 LAS 开发机中执行以下代码
   import os
   
   """
   转为 Lance 数据集
   """
   import daft
   from daft.io import CreateLasDatasetOptions, IOConfig, LanceWriteOptions
   from daft.las.io import TOSConfig
   
   # 您需要设置图片数据集名称，并设置新的Lance数据集的TOS路径
   dataset_name = "las_dataset_image"
   lance_tos_dir = f"tos://las-doctest/image2lance/las_dataset_image_lance.lance"
   LANCE_DATASET_FORMAT = "lance"
   
   # 创建相关的配置
   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",
   )
   
   # 读取原始数据
   df = daft.read_las_dataset(name=dataset_name)
   
   # 在这里进行创建，新建的Lance数据集名称规则为：图片数据集名称加上_lance后缀
   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,
   )
   
   # 读取转换后的新数据集
   df = daft.read_las_dataset(name=new_dataset_name)
   print("\n\nnew dataset:")
   df.show()
   

您需要修改以下参数。

• dataset_name：设置为图片数据集名称，本示例为las_dataset_image
• lance_tos_dir：设置上述步骤中准备好的转换后的新Lance数据集的TOS路径。

其他参数可保持示例值即可。

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)

转换后的 Lance 数据集包含“image”、“size”、“num_rows”字段。其中：

• image字段：为当前 Lance 数据集中每个图片数据所在的 TOS 路径（s3格式）。
• size字段：为每个图片的大小。

回显中如果出现ERROR提示，但是正常返回了数据集数据，可忽略ERROR提示。

# 地域设置，本示例使用华北2-北京地域
export REGION="cn-beijing"

# 鉴权AK相关环境变量
# 访问TOS、LAS的认证sk、ak
export LAS_TOS_SECRET_KEY="<your_sk>"
export LAS_TOS_ACCESS_KEY="<your_ak>"

# TOS 访问相关环境变量
# TOS的访问端点
export LAS_TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"
# TOS端点（用于TOSConfig）
export TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"

# 数据集固定参数也可设置为环境变量
export DATASET_NAME="las_dataset_image_lance"

本步骤新增了：

• DATASET_NAME：设置为上述步骤转换后的新 Lance 数据集名称，便于后续任务代码引用。

import os
# 从环境变量引入以下参数取值
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

# 从 tos 路径读取图片数据
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])

        # 仅返回新列
        return new_batch

    return create_assign_raw_image

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

# 您需要修改以下代码中的lance_tos_dir、merge_columns中transform参数
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 数据集")

您需要根据实际情况修改示例代码中的以下参数：

• lance_tos_dir：修改其中Lance 数据集的TOS 路径为您真实的路径。
• merge_columns中transform：调用了函数，入参“image”和“raw_image”可根据实际情况修改：
  • image：为 Lance 数据集中的TOS路径字段
  • raw_image：为自定义的新的字段名，图片处理为二进制数据后，会写入此字段中

如果因为误操作导致数据集中出现无用字段或无用数据，可参考此内容进行删除：参考：常见数据集操作示例。

# 地域设置，本示例使用华北2-北京地域
export REGION="cn-beijing"

# 鉴权AK相关环境变量
# 访问TOS、LAS的认证sk、ak
export LAS_TOS_SECRET_KEY="<your_sk>"
export LAS_TOS_ACCESS_KEY="<your_ak>"
# 调用 LAS 算子的LAS API Key，使用在线方式（API）调用时需要
export LAS_API_KEY="464812df-2dfb-41f6-********"

# TOS 访问相关环境变量
# TOS的访问端点
export LAS_TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"
# TOS端点（用于TOSConfig）
export TOS_ENDPOINT="https://tos-cn-beijing.ivolces.com"

# 数据集固定参数也可设置为环境变量
export DATASET_NAME="las_dataset_image_lance"
export TOS_LANCE_DIR="tos://las-doctest/image2lance/las_dataset_image_lance.lance"

# 测试配置
# 测试用的示例图片URL
export TEST_IMAGE_URL="https://ark-project.tos-cn-beijing.volces.com/images/view.jpeg"

本步骤新增了：

• LAS_API_KEY：使用在线方式调用 LAS 算子时，需使用LAS_API_KEY来鉴权，您可设置为上述“准备工作”节点已获取到的真实取值。
• TOS_LANCE_DIR：将Lance数据集的TOS路径这个固定取值设置为环境变量，便于任务代码引用。
• TEST_IMAGE_URL：用于后续测试算子能力的测试图片，您可直接使用示例中的URL，也可修改为其他测试图片URL。

在开发机中执行以下脚本。

#本段代码需要先执行"source env.sh", env.sh文件在"配置环境变量"部分
# 导入必要的库
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

# 从环境变量获取配置
TOS_LANCE_DIR = os.getenv("TOS_LANCE_DIR")  # Lance数据集路径
DATASET_NAME = os.getenv("DATASET_NAME")    # 数据集名称

# 批处理参数
BATCH_SIZE  = int(os.getenv("BATCH_SIZE",  "10"))  # 单次处理行数上限
MAX_CYCLES  = int(os.getenv("MAX_CYCLES",  "5"))   # 最大批次数

# 列名配置
IMAGE_EMBEDDING_NEW_COLUMN = "image_embedding"    # 新 embedding 列名
DOUBAO_LLM_COLUMN          = "doubao_llm_result"  # 豆包 LLM 结果列名


# 设置 TOS 对象存储相关环境变量
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")

# 配置 IO 设置，用于访问 TOS 对象存储
io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())

def update_image_embedding():
    # 将 TOS 路径转换为 S3 兼容路径
    lance_path = TOS_LANCE_DIR.replace("tos://", "s3://")

    # 检查 embedding 列是否存在，不存在则创建
    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} 列")

    # 读取完整数据集，过滤出待处理行
    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. 构建 Daft DataFrame 并调用算子批量计算 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",   # 按需调整图片格式
                "source_type":  "url",    # image 列存放的是 URL
            },
        )(col("image"), col(DOUBAO_LLM_COLUMN)),  # image列 + 文本列 → 图文联合向量
    )

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

    # 过滤掉算子调用失败（返回 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

    # 更新 Lance 数据集（与原逻辑一致）
    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)} 张图片")

# 程序入口点
if __name__ == "__main__":
    update_image_embedding()

您需按需修改示例代码中的以下参数：

• BATCH_SIZE & MAX_CYCLES：设置每次执行脚本时，批处理图片的行数和批数。
• IMAGE_EMBEDDING_NEW_COLUMN：自定义一个列名，后续会将向量化结果写入此列中。
• DOUBAO_LLM_COLUMN：配置为上述图片内容理解写入的 Lance 数据集列。
• 如果您使用的是您自己的测试图片集，需根据实际情况，修改调用las_udf进行图文向量化时，image_format（图片格式）、source_type（图片来源的类型）的取值。

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

# ✅ 用 UDF 实现字符串包含判断
@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 = "袋鼠"
    # ✅ 使用 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()

本文搜图的示例代码，为向Lance 数据集中搜索“袋鼠”相关图片。
其中main函数中：

• dataset_name：为待查询的Lance数据集名称。
• query_word：为搜索的自然语言描述。

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


# ✅ 用 UDF 实现字符串包含判断
@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 = "袋鼠"
    # ✅ 使用 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
    """
    # 下载查询图片（仅用于显示）
    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)}")


    # 读取LAS数据集
    df = daft.read_las_dataset(name=dataset_name)


    # 获取所有数据
    all_data = df.collect()
    all_rows = all_data.to_pylist()


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


    # 计算所有图片与查询图片的相似度
    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))


    # 按相似度排序，取前3个且相似度大于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


    # 显示结果
    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)

本图搜图的示例代码，为给定一个输入图片，向Lance 数据集中搜索图片，计算图片的余弦相似度，返回相似度TOP 3的图片结果。
其中main函数中：

• dataset_name：为待查询的Lance数据集名称。
• query_pic_url：为搜索的图片URL，本示例以TOS中的图片作为输入，配置为对应图片的TOS路径格式的URL。