Description #

Provides deep reasoning capabilities of large models in multimodal scenarios
Uses models with deep reasoning capabilities to analyze and interpret images, videos, or text, and returns structured text output. The operator automatically constructs a message structure compliant with multimodal model specifications. Users only need to provide image, video, or text data as specified to complete inference.

Key features #

• Deep reasoning mechanism: The model automatically decomposes questions and performs logical reasoning before answering, generating a reasoning chain (reasoning_content)
• Multimodal scenario support: Supports simultaneous input of images, videos, and text, and automatically assembles multimodal messages
• Input simplification mechanism: Supports multiple data sources including local files, HTTP/HTTPS URLs, TOS/S3 object storage, and more. You can achieve visual understanding capabilities through simple configuration.
• Flexible reasoning mode: Supports controlling the deep reasoning mode (enabled / disabled / auto) via the thinking_type parameter, allowing flexible trade-offs between answer quality and performance

Input and output specifications #

• Input format:
  • Images (images): Supports string or list format
    • Supported data source types:
      • url: Network resource address, supports http/https/tos/s3. TOS/S3 will automatically generate a pre-signed URL
      • base64: Base64-encoded data
      • binary: Binary data (internally converted to Base64 encoding)
  • Videos (videos): Supports string or list format. Data source types are the same as for images
  • Text (texts): Supports string or list format, used as user prompt input

(String and list types cannot be mixed within the same field)

• Output format:
  • Default mode: Returns struct type, containing the following fields:
    • llm_result: The model's final answer content
    • reasoning_content: The model's reasoning chain / reasoning process
  • When the environment variable LAS_LLM_FINISH_REASON_CHECK=true is set, the struct will additionally include:
    • finish_reason: The reason for the model result ending. Possible values include:
      • stop: Normal termination
      • length: Exceeds token limit
      • content_filter: Content filtering strategy triggered
  • When the environment variable LAS_LLM_BOTS_REFERENCES=true is set, the struct will additionally include:
    • references: Retrieval results or external knowledge information referenced by the model, facilitating downstream transmission and parsing

Model capability enhancement #

• Chain-of-thought visualization: The reasoning process of the model is output through the reasoning_content field, making it easier to observe and review.
• Result reliability control: The finish_reason field is used to identify abnormal termination cases, such as length truncation or content filtering.
• Multimodal understanding: Supports mixed input parsing of images, videos, and text, suitable for complex multimodal scenarios.

Version compatibility notes #

This operator (ArkLLMThinkingVision) is used differently in Daft 0.6.5 (and earlier versions) compared to Daft 0.6.14 (and later versions).

• In Daft 0.6.14 and later versions, you can specify the images, texts, and videos fields when providing input, which correspond to the model's image, text, and video inputs. You can provide all of them at the same time. Here is an example:

  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=""),
  
      )
  

• In Daft 0.6.5 and earlier versions, only video or image can be provided as input, and simultaneous input of video and image is not supported. The input type is specified using multimodal_type, which accepts "video", "image", and "text". It also supports providing text information by specifying the prompt. Here is an example:

df = df.with_column(

        "llm_result",

        las_udf(

            ArkLLMThinkingVision,

            construct_args={

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

                "multimodal_type": "video",

                "system_text": "",

                "prompt": "",

                "inference_type": "online",

            },

        )(col("videos")),

    )

Interface description #

The model is invoked via the "dialog interface", where inference is initiated for each request, and the system waits for the model to output results. Using Daft, a batch of requests can be processed in parallel to improve inference efficiency.

Operator parameters #

Input #

Output #

By default, the returned field type is struct and includes the following fields:

• llm_result: Model output result
• reasoning_content: Chain-of-thought content output by the model (reasoning process)

When the environment variable LAS_LLM_FINISH_REASON_CHECK=true, the struct will additionally include the following field:

• finish_reason: The reason for model output termination. Common values include:
• stop: Normal termination
• length: The generated length has reached the upper limit
• content_filter: Content safety or filtering policy triggered

When the environment variable LAS_LLM_BOTS_REFERENCES=true, the struct will additionally include the following field:

• references: Reference information in JSON string format, used to carry external knowledge or reference results retrieved by the model, facilitating downstream parsing and display.

Both environment variables can be enabled simultaneously, in which case the struct will include all of the above fields.

Parameters #

If a parameter does not have a default value, it is required.

Examples #

The following code demonstrates how to use Daft to access the Volcano ModelArk multimodal deep thinking model (Doubao series) for batch inference. Please note: The results of each large model inference may vary.

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.set_runner_ray()

daft.set_execution_config(min_cpu_per_task=0)

if __name__ == "__main__":
    # The environment variable LAS_API_KEY must be configured: LAS_API_KEY is obtained by creating it on the LAS service page
    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")["reasoning_content"])
    df = df.with_column("llm_result", col("llm_result")["llm_result"])
    df.show()

    #  Output (the result of each large model inference may vary)
    # ╭────────────────────────────────┬────────────────────────────────────────────┬────────────────────────────────────────────╮
    # │ videos                         ┆ llm_result                                 ┆ reasoning_content                          │
    # │ ---                            ┆ ---                                        ┆ ---                                        │
    # │ Utf8                           ┆ Utf8                                       ┆ Utf8                                       │
    # ╞════════════════════════════════╪════════════════════════════════════════════╪════════════════════════════════════════════╡
    # │ https://las-public-data-qa.to… ┆ The video presents an animated segment: initially, it shows a        ┆ The user now needs to describe the content in the video. First, observe the visuals:        │
    # │                                ┆ A **multi-layer cartoon wind…                              ┆ The beginning features a multi-layer egg…                             │
    # ╰────────────────────────────────┴────────────────────────────────────────────┴────────────────────────────────────────────╯

Interface description #

This interface is used to assemble data into the request format required by the model as a JSONL file and upload it to Volcano Engine Object Storage (TOS). Batch inference tasks are executed using the Doubao model.

Operator parameters #

Input #

Output #

Returns a list of JSONL file paths written to TOS. This JSONL file contains the input data for batch inference. After the JSONL file is generated, it will be submitted to the Doubao model for batch inference. At the same time, a corresponding inference task and its progress will be available in the task management section of the AI Data Lake Service.

Parameters #

If a parameter does not have a default value, it is required.

Examples #

The following code demonstrates how to use the submit_ark_batch_inference_job interface to perform batch inference on a Daft DataFrame. Please note: The results of each large model inference may vary.

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. Preparation: When executing the task, set the following environment variables on the LAS platform as needed ---
# LAS_API_KEY
# ACCESS_KEY
# SECRET_KEY
# TOS_ENDPOINT


# --- 2. Define input and output paths ---
# Caution: The path must be a TOS or S3 path
# Configure the object storage bucket name as needed
bucket_name = "bucket_name"
# Configure the path for the raw data. This path stores the raw data that needs to be inferred using the Doubao model (only parquet/csv/json storage formats are supported; parquet format is recommended). If the file does not exist, you can also create it using the method described below:
input_data_path = f"s3://{bucket_name}/doubao_test_job_path/original_data.parquet"
# Configure the output directory for inference results. This directory stores the files with merged inference results. That is, append the inference results to the input_data_path file and save them to a new file.
output_dir = f"tos://{bucket_name}/doubao_test_job_path/to/inference_output/"
# Configure the output directory for the batch inference dataset. If batch_inference_dataset_dir is not specified, the default is {output_dir}/batch_inference_dataset. This path stores the jsonl files containing inference results returned by the Doubao model.
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. Construct Daft DataFrame ---
# ---- If the input_data_path does not exist, a simple creation method is provided below. Otherwise, skip the following steps and read the file directly.
# Suppose we have a DataFrame containing IDs and text prompts
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)

# Read the input_data_path file
df = daft.read_parquet(input_data_path, io_config=io_config)

# --- 4. Call the interface to submit the batch inference task ---
generated_files = df.submit_ark_batch_inference_job(
    output_dir=output_dir,
    primary_column_name="request_id",
    model="doubao-seed-1-8",
    version="251228",
    # Define the input data columns
    text_column_name="prompt",
    # Define the information required for the callback
    input_dir=input_data_path,
    input_format="parquet",
    # Define the LLM inference parameters
    max_tokens=256,
    temperature=0.7,
    # (Optional) Performance parameters: split 10 data entries into 2 files for inference
    # num_batches=2,
    # (Optional) Repartition after callback: merge the final results into a single file
    repartition=1,
    completion_window="2d",
)

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