描述 #

多模态场景下提供大模型的深度思考能力
使用具备深度思考能力的模型，对图片、视频或文本进行分析理解，并返回结构化文本输出。算子会自动构建符合多模态模型规范的 message 结构，用户只需按约定提供图片 / 视频 / 文本数据即可完成推理。

核心功能 #

• 深度思考机制：模型在回答问题前自动进行问题拆解和逻辑推理，生成思维链（reasoning_content）
• 多模态场景支持：同时支持图片 / 视频 / 文本输入，自动完成多模态消息拼装
• 输入简化机制：支持本地文件、HTTP/HTTPS URL、TOS/S3 对象存储等多种数据源，通过简单配置即可实现视觉理解能力
• 灵活思考模式：支持通过 thinking_type 参数控制深度思考模式（enabled / disabled / auto），在回答质量与性能之间灵活权衡

输入输出规范 #

• 输入格式：
  • 图片（images）：支持 string 或 list 形式
    • 支持的数据来源类型：
      • url：网络资源地址，支持 http/https/tos/s3，TOS/S3 会自动生成预签名 URL
      • base64：Base64 编码数据
      • binary：二进制数据（内部会转换为 Base64 编码）
  • 视频（videos）：支持 string 或 list 形式，数据来源类型与图片一致
  • 文本（texts）：支持 string 或 list 形式，作为用户提示词输入

（同一个字段中不允许同时混用字符串和列表两种类型）

• 输出格式：
  • 默认模式：返回 struct 类型，包含以下字段：
    • llm_result：模型最终回答内容
    • reasoning_content：模型的思维链 / 推理过程
  • 当设置环境变量 LAS_LLM_FINISH_REASON_CHECK=true 时，struct 中会额外包含：
    • finish_reason：模型结果结束原因，取值范围包括：
      • stop：正常终止
      • length：超出 token 限制
      • content_filter：命中内容过滤策略
  • 当设置环境变量 LAS_LLM_BOTS_REFERENCES=true 时，struct 中会额外包含：
    • references：模型引用的检索结果或外部知识信息，便于下游透传与解析

模型能力增强 #

• 思维链可视化：通过 reasoning_content 字段输出模型的推理过程，便于观测与复盘
• 结果可靠性控制：通过 finish_reason 字段识别异常终止情况，例如长度截断或内容过滤
• 多模态理解：支持图片 / 视频 / 文本的混合输入解析，适用于复杂多模态场景

版本兼容性说明 #

该算子（ArkLLMThinkingVision）在 Daft 0.6.5（及其以下版本），使用方式与 0.6.14 版本（及其以上版本）不相同。

• 在 Daft 0.6.14 及其以上版本中，输入时可以指定 images、texts、 videos 字段，分别对应模型的图片、文本、视频输入，支持同时输入。下面是一个示例：

  df = df.with_column(
  
          "llm_result",
  
          las_udf(
  
              ArkLLMThinkingVision,
  
              construct_args={
  
                  "model": "doubao-seed-1.6",
  
                  "system_text": "你是一个专业的视频理解助手，能够分析视频中的内容并提供详细的描述。",
  
                  "inference_type": "online",
  
              },
  
          )(videos=col("videos"), texts="请用简短的语言描述视频里的内容，返回格式为json。"),
  
      )
  

• 在 Daft 0.6.5 及其以下版本中，输入时只支持传入视频或图片，不支持同时传入视频和图片，通过multimodal_type指定输入类型，multimodal_type支持传入 "video" 和 "image"、"text"。同时也支持通过指定prompt传入文本信息。下面是一个示例：

df = df.with_column(

        "llm_result",

        las_udf(

            ArkLLMThinkingVision,

            construct_args={

                "model": "doubao-seed-1.6",

                "multimodal_type": "video",

                "system_text": "你是一个专业的视频理解助手，能够分析视频中的内容并提供详细的描述。",

                "prompt": "请用简短的语言描述视频里的内容，返回格式为json。",

                "inference_type": "online",

            },

        )(col("videos")),

    )

接口说明 #

通过「对话接口」调用模型的方式，特点是以一条条请求为单位发起推理，等待模型输出结果。使用Daft方式，可以将一批请求并行处理，提高推理效率。

算子参数 #

输入 #

输出 #

默认情况下，返回字段类型为 struct，包含以下字段：

• llm_result: 模型输出结果
• reasoning_content: 模型输出的思维链内容（推理过程）

当环境变量 LAS_LLM_FINISH_REASON_CHECK=true 时，返回 struct 中会额外包含以下字段：

• finish_reason: 模型输出结束原因，常见取值包括：
• stop：正常终止
• length：生成长度达到上限
• content_filter：命中内容安全/过滤策略

当环境变量 LAS_LLM_BOTS_REFERENCES=true 时，返回 struct 中会额外包含以下字段：

• references: JSON 字符串形式的引用信息，用于承载模型检索到的外部知识或引用结果，方便下游解析与展示。

两个环境变量可以同时开启，此时输出 struct 中会同时包含上述字段。

参数 #

如参数没有默认值，则为必填参数

调用示例 #

下面的代码展示了如何使用 daft 访问火山方舟 多模态深度思考模型（豆包系列）进行批量推理。请注意每次大模型推理结果可能不同。

from __future__ import annotations

import os

import daft
from daft import col
from daft.las.functions.ark_llm.ark_llm_thinking_vision import ArkLLMThinkingVision
from daft.las.functions.udf import las_udf

if os.getenv("DAFT_RUNNER", "native") == "ray":
    import logging

    import ray

    def configure_logging():
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S.%s".format(),
        )
        logging.getLogger("tracing.span").setLevel(logging.WARNING)
        logging.getLogger("daft_io.stats").setLevel(logging.WARNING)
        logging.getLogger("DaftStatisticsManager").setLevel(logging.WARNING)
        logging.getLogger("DaftFlotillaScheduler").setLevel(logging.WARNING)
        logging.getLogger("DaftFlotillaDispatcher").setLevel(logging.WARNING)

    ray.init(dashboard_host="0.0.0.0", runtime_env={"worker_process_setup_hook": configure_logging})
    daft.context.set_runner_ray()

daft.set_execution_config(min_cpu_per_task=0)

if __name__ == "__main__":
    # 需配置环境变量 LAS_API_KEY ： LAS_API_KEY 通过在 LAS 服务页面上创建获取
    tos_dir_url = os.getenv("TOS_DIR_URL", "las-cn-beijing-public-online.tos-cn-beijing.volces.com")
    samples = {
        "videos": [
            f"https://{tos_dir_url}/public/shared_video_dataset/eating_56.mp4",
        ]
    }

    df = daft.from_pydict(samples)
    df = df.with_column(
        "llm_result",
        las_udf(
            ArkLLMThinkingVision,
            construct_args={
                "model": "doubao-1.5-thinking-vision-pro",
                "system_text": "你是一个专业的视频理解助手，你需要根据视频内容回答用户的问题。",
                "inference_type": "online",
            },
        )(videos=col("videos")),
    )

    df = df.with_column("reasoning_content", col("llm_result").struct.get("reasoning_content"))
    df = df.with_column("llm_result", col("llm_result").struct.get("llm_result"))
    df.show()

    #  输出(每次大模型推理结果可能不同)
    # ╭────────────────────────────────┬────────────────────────────────────────────┬────────────────────────────────────────────╮
    # │ videos                         ┆ llm_result                                 ┆ reasoning_content                          │
    # │ ---                            ┆ ---                                        ┆ ---                                        │
    # │ Utf8                           ┆ Utf8                                       ┆ Utf8                                       │
    # ╞════════════════════════════════╪════════════════════════════════════════════╪════════════════════════════════════════════╡
    # │ https://las-public-data-qa.to… ┆ 视频中呈现的是一段动画内容：起初展示的是一        ┆ 用户现在需要描述视频里的内容。首先看画面：        │
    # │                                ┆ 个**多层卡通风…                              ┆ 开头是一个多层蛋…                             │
    # ╰────────────────────────────────┴────────────────────────────────────────────┴────────────────────────────────────────────╯

接口说明 #

该接口用于将数据组装为模型需要的请求格式 JSONL 文件，并将其上传至火山引擎对象存储(TOS)上。采用豆包模型执行批量推理任务，该批量推理任务采用job模式。

算子参数 #

输入 #

输出 #

返回写入TOS上的 JSONL 文件路径列表。该jsonl文件包含了批量推理的输入数据。当JSONL文件生成后，会将文件提交至豆包模型做批量推理，同时在 AI 数据湖服务中任务管理中会有模型推理对应的任务及其进度。

参数 #

如参数没有默认值，则为必填参数

调用示例 #

下面的代码展示了如何使用 submit_ark_batch_inference_job 接口对 Daft DataFrame 进行批量推理。请注意，每次大模型推理结果可能不同。

from __future__ import annotations

import logging

import daft
from daft.io import IOConfig
from daft.las.io.tos import TOSConfig

logging.basicConfig(level=logging.INFO)

# --- 1. 准备：执行任务时，需要根据实际情况 在LAS平台设置以下环境变量---
# LAS_API_KEY
# ACCESS_KEY
# SECRET_KEY
# TOS_ENDPOINT


# --- 2. 定义输入输出路径 ---
# 注意：路径必须是 TOS 或 S3 路径
# 根据实际情况配置对象存储桶名称
bucket_name = "bucket_name"
# 配置原始数据的路径。该路径存储需要用豆包模型进行推理的原始数据（只支持parquet/csv/json的存储格式，建议使用parquet格式）。若该文件不存在，也可以按照后面说的方式创建：
input_data_path = f"s3://{bucket_name}/doubao_test_job_path/original_data.parquet"
# 配置推理结果的输出目录。该目录存储将推理结果合并后的文件。即在input_data_path文件中追加推理结果并保存到新文件中
output_dir = f"tos://{bucket_name}/doubao_test_job_path/to/inference_output/"
# 配置批量推理数据集的输出目录。`batch_inference_dataset_dir` 若不指定，默认为 {output_dir}/batch_inference_dataset。该路径存储豆包模型返回包含推理结果的jsonl文件
batch_inference_dataset_dir = f"tos://{bucket_name}/doubao_test_job_path/to/batch_inference_dataset"

io_config = IOConfig(s3=TOSConfig.from_env().to_s3_config())


# --- 3. 构造 Daft DataFrame ---
# ---- 若input_data_path路径不存在，下面提供简单的创建方法。否则，跳过下面步骤，直接读取该文件。
# 假设我们有一个包含ID和文本提示的 DataFrame
data = {
    "request_id": [f"req_{i}" for i in range(10)],
    "prompt": [f"写一个关于数字 {i} 的短故事" for i in range(10)],
}
df = daft.from_pydict(data)
df.write_parquet(input_data_path.replace("tos://", "s3://"), io_config=io_config)

# 读取input_data_path文件
df = daft.read_parquet(input_data_path, io_config=io_config)

# --- 4. 调用接口提交批量推理任务 ---
generated_files = df.submit_ark_batch_inference_job(
    output_dir=output_dir,
    primary_column_name="request_id",
    model="doubao-seed-1-8",
    version="251228",
    # 定义输入数据列
    text_column_name="prompt",
    # 定义回调所需信息
    input_dir=input_data_path,
    input_format="parquet",
    # 定义LLM推理参数
    max_tokens=256,
    temperature=0.7,
    # (可选) 性能参数：将10条数据分成2个文件进行推理
    # num_batches=2,
    # (可选) 回调后的重分区：将最终结果合并成一个文件
    repartition=1,
    completion_window="2d",
)

print(f"成功生成并提交了 {len(generated_files)} 个输入文件:")
for file_path in generated_files:
    print(f"- {file_path}")