Check if image extension is in the exclusion list.

Load checkpoint from file.

Load config file with _base_ inheritance support. Recursively loads and merges base configurations specified by the _base_ key.

Load a Python config file as a dict without resolving _base_ inheritance. Imports the file as a Python module and extracts all non-dunder att

Load source mapping from YAML using standardized config loader.

Log a sample request for debugging.

Log a sample response for debugging.

Make an API request, delegating to the non-streaming handler.

Convert messages schema to standard schema. Input format: { "messages": [ {"role": "user", "c

Apply all normalization patterns.

Parse conversations into QA objects.

Re-index metadata for extracted single turn.

Remove instruction patterns from text.

Resolve base config path. Supports: - "@/path" : relative to configs root directory - "path" : relative to current config f

Resolve dataset file path (supports JSON/JSONL formats). Args: dataset: Dataset path string or config dict. data_root

Cache images only mode: iterate all data and write images to LMDB cache.

Normal pipeline execution mode.

Convert standard schema back to messages schema. Input format: { "conversations": [ {"from":

Add a filter record for the entire item (not per-QA).

Record which model was used by an operator to process this item. Stored as data["model"] = {op_name: model_name, ...}. Args:

Resize CV2 image maintaining aspect ratio. Args: img_cv2: OpenCV image (numpy array), shape is (height, width, channels). resize_

Batch check existence of multiple images, one transaction per shard. Uses cursor.set_key() for fast key-only existence checks (avoids

Batch read multiple images, one transaction per shard. Opens temporary LMDB envs with readahead=True for better sequential read perfo

Check if a QA pair should be rejected. Args: item: DataItem to check. qa_idx: QA pair index. Return

Check if response contains image/video tags.

Check if dataset has already been processed. Args: dataset: Dataset object (must have 'datas' attribute). Returns:

Create a single dataloader for the given dataset path.

Log critical message. Critical errors are logged to wandb by default.

Log debug message.

Return list of supported file extensions. Returns: list: List of file extensions (e.g., ['.json', '.JSON']).

Main method to generate responses. Uses asyncio.run to execute the async generation pipeline. Pre-encodes images before async process

Sync version of generate for single request (BaseAPI compatibility). Args: inputs: Preprocessed message list (OpenAI format).

Get the dataset name.

Check if this dataloader has checkpoint tracking enabled and has progress.

Load config and return file_path to source mapping. Args: yaml_path: Path to YAML config file. Returns: dict

CLI entry point for DataStudio pipeline.

Process a single data item. Args: item: DataItem to process. Returns: Result with filter and/or rew

Check if image meets size requirements (item-level).

Check if all images have acceptable aspect ratios (item-level).

Split is handled specially - we mark items for expansion. The actual splitting is done in expand_items().

Write a single image to cache. Args: image_path: Image path (used as key). img_bytes: Image data as bytes. R

Reset global statistics collector.

Rewrite a QA pair's answer. Args: item: DataItem to rewrite. qa_idx: QA pair index. Returns:

Remove [ANSWER 0] prefix from an answer.

Shutdown and release resources. Session is created and destroyed per generate() call.

Convert to dictionary. Returns: dict: Configuration as dictionary.

Batch write images to sharded LMDB with optimized per-shard grouping. Args: sharded_manager: ShardedLMDBManager instance. image_p

Execute the sub-pipeline. Args: datas: List of raw data dicts to process. Returns: List of all proc

Execute all sub-pipelines in priority order. Args: datas: List of raw data dicts to process. Returns:

Accumulate data for batch saving. Args: datas: List of data dicts. Rejected items have 'rejected=True'.

Check if a module is registered.

Get attribute by name, returning None if not found.

Initialize the Config. Args: *args: Optional single argument - path to config file.

Initialize the sharded LMDB manager. Args: cache_dir: Cache directory path. num_shards: Number of shards to use.

Initialize checkpoint manager. Args: work_dir: Working directory for checkpoint file. checkpoint_name: Name of checkp

Initialize the registry. Args: name: Optional name for the registry (for debugging).

Initialize the MLLM operator. Args: model: MLLM model with generate() method. request_builder: RequestBuilde

Initialize selective rewriter. Args: should_rewrite_fn: Function(question, answer) -> bool.

Initialize request builder. Instance values override class defaults. For key_templates, explicitly passing None disables JSO

Initialize a DataItem. Args: data: Raw data dict (will be modified in place for rewrites). idx: Index in the

Initialize the operator. Args: name: Operator name (defaults to class name). logger: Logger instance.

Initialize the filter. Args: min_size: Minimum dimension (width or height). max_ratio: Maximum aspect ratio

Initialize the filter. Args: check_question: Whether to check questions. check_answer: Whether to check answ

Initialize the filter. Args: max_aspect_ratio: Maximum allowed aspect ratio (width/height or height/width).

Initialize the filter. Args: min_length: Minimum allowed length. max_length: Maximum allowed length.

Initialize the filter. Args: logger: Logger instance.

Initialize the filter. Args: excluded_exts: List of excluded extensions (with dot, e.g., ['.gif', '']).

Initialize the filter. Args: min_length: Minimum number of QA pairs (inclusive). max_length: Maximum number

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the rewriter. Args: logger: Logger instance.

Initialize the pipeline. Args: operations: Dict of operation configs with cfg and priority. logger: Logger i

Initialize the data saver. Args: output_dir: Output directory path. dataset: Original dataset YAML path.

Initialize the saver. Args: output_dir: Output root directory. sources: Mapping from json_path to source name.

Initialize the checker. Args: output_dir: Output directory path. sources: Mapping from file_path to source name.

Initialize the generator. Args: output_dir: Output directory path. yaml_name: YAML file name (without extension).

Initialize multiprocessing OpenAI API wrapper. Args: model: Model name, e.g., gpt-4, Qwen2-VL-72B. key: API key (uses

Initialize the base API. Args: retry: Number of retry attempts on API failure. wait: Wait time between retries (secon

Initialize OpenAI-compatible API wrapper. Args: model: Model name, e.g., gpt-4, Qwen2-VL-72B. key: API key (uses OPEN