Get a registered module by name. Args: name: Name of the registered module. Returns: The registered module c

Log info message. Args: msg: Message (str or dict). Dict auto-logged to wandb. to_wandb: If True, also log s

Update checkpoint for a file. Args: file_path: Path to the data file. processed_idx: Index of last processed item (ex

Get format handler instance based on file extension. Args: file_path: Path to data file. Returns: BaseFormat

Close logger and cleanup.

Log warning message.

Convert Config to plain dict recursively. Returns: dict: Plain dictionary representation.

Check if image exists in cache. Args: image_path: Image path to check. Returns: bool: True if image exists.

Concrete implementation of abstract method.

Add a filter decision.

Concrete implementation of rewrite method.

Get progress info for a file. Args: file_path: Path to the data file. Returns: Progress dict or None if not

Build a module from configuration. Args: cfg: Configuration dict with 'type' key, or string name, or None. registry: Registry con

Add a rewrite decision.

Concrete implementation of check method.

Add decision based on parsed response. Subclasses implement this to add FilterDecision or RewriteDecision. Args:

Main method to generate responses. Args: messages: Input messages (list of message dicts). **kwargs: Additional argum

Convert a single item to standard schema. Stores original schema info for later denormalization. Args: item: Data item i

Get a specific QA pair. Args: qa_idx: Index of the QA pair. Returns: QA object at the specified index.

Get adaptive resize size based on image count. Args: image_count: Number of images in the item. Returns: Res

Clear checkpoint(s). Args: file_path: Specific file to clear, or None to clear all.

Get answer at index.

Get question at index.

Wrap raw dicts into DataItems.

Build dataloaders with checkpoint support. Args: dataset_config: Dataloader configuration dict. logger: Logger instance.

Detect consecutive repetitive patterns at the end of text. Looks for patterns like "ABCABCABC..." where the same substring is repeated m

Detect if text is primarily Chinese or English. Args: text: Input text to analyze. Returns: "chinese", "english", or "u

Check if a file has been fully processed. Args: file_path: Path to the data file. Returns: True if file is c

Parse OpenAI-format API response content. Args: data: Parsed JSON response from OpenAI API. Returns: Extract

Check input content type. Args: msgs: Raw input messages. Returns: str: Message type (str/dict/liststr/listd

Get number of QA pairs in conversation. Args: conversations: List of conversation turns. Returns: Number of complete QA

Extract raw dicts from DataItems.

Add file to configuration list. Args: file_path: Path to data file. is_rejected: Whether this is a rejected data file

Log error message. Errors are logged to wandb by default.

Create config from dict, automatically handling field aliases. Args: data: Raw configuration dictionary. Returns:

Parse MLLM response string to dictionary. Tries in order (fast to slow): 1. Standard json.loads (fastest) 2. json_repair.loads (hand

Parse model response. Args: response: Raw response from model. qa_idx: QA index (used to format key_template

Record operator statistics directly (called from Pipeline). Args: name: Operator name input_count: Number of input it

Log metrics to wandb.

Apply all decisions to a data item. Order: rewrite first, then filter (to preserve rewritten content). Args: it

Get the starting index for processing a file. Args: file_path: Path to the data file. total_items: Total number of it

Record dataset statistics (called from DataSaver). Statistics are merged by filename (basename), so kept and rejected parts

Build request payload for chat completions API. Args: message: Preprocessed message list (OpenAI format). **kwargs: O

Resize PIL image maintaining aspect ratio. Args: img_pil: PIL Image object. resize_image_size: Maximum dimension for the larger s

Convert a single item back to its original schema. Args: item: Data item in standard schema with schema metadata. Return

Initialize global config from file. Args: filename: Path to the config file.

Load data from file. Args: file_path: Path to the data file. Returns: list: List of data dictionaries.

Split this item into kept and rejected parts. Creates two new DataItems: - kept_item: Contains only the kept QA pairs

Wrap a list of raw dicts into DataItems.

Check if this item should be skipped.

Build a single request. Args: item: DataItem to build request for. qa_idx: QA pair index (-1 for batch_qa mo

Calculate QA statistics and optionally adjust batch size. Args: datas: List of data items with "conversations" field. batch_

Detect the schema type of a data item. Args: item: A single data item dict. Returns: Schema type string (Sch

Encode image directly without caching. Args: img_item: PIL Image object or other image data. Returns: Base64

Get or create global statistics collector.

Mark a file as fully processed. Args: file_path: Path to the data file. total_items: Total number of items in the fil

Merge dict b into dict a recursively. Supports special keys: - _cover_: Replace entire dict instead of merging. Args:

Save accumulated data to disk.

Format message with additional key-value pairs.

Lazy-load shard environment with LRU eviction (thread-safe). Args: shard_id: Shard ID to load. Returns: lmdb

Process a single request with concurrency control and retry. Args: session: Shared aiohttp session. semaphore: Concur

Make a non-streaming API request. Args: session: Shared aiohttp session. message: Preprocessed message list.

Check if this QA pair should be rewritten.

Mark pipeline end.

Get the global config instance. Returns: Config: Global configuration object.

Get the global DataStudio logger instance. Args: name: Logger name (ignored if instance exists) Returns: Logger instanc

Calculate shard ID based on key hash. Args: key: Key string to hash. num_shards: Total number of shards. Returns: in

Parse input to determine if it's an image file. Args: s: Input to parse (string path, PIL Image, list, or other). image_name: Nam

Mark pipeline start.

Calculate exponential backoff delay with full jitter. Args: attempt: Current retry attempt (0-indexed). max_delay: Ma

Add a filter record (for rejected items).

Build file path and create parent directories. Creates the directory structure if it doesn't exist and returns the full file path. Args:

Check for text length anomalies in a QA pair.

Return image reference directly (images are pre-encoded to base64).

Detect schema from a list of items (uses first valid item). Args: data: List of data items. Returns: Detecte

Deep-first search conversion of nested dicts to Config objects. Args: cfg_other: Input configuration (dict, list, or value).

Expand multi-turn items into single-turn items. This should be called directly instead of through Pipeline for splitting ope

Export statistics to CSV file. Args: output_path: Output file path (default: work_dir/stats.csv) Returns:

Format a single QA pair for prompt. Display index is always 0 (standard for single-QA prompts).

Get all data file paths. Returns: List of file paths from all datasets.

Get or create singleton logger instance.

Get source name from file path. Args: file_path: Path to data file. Returns: Source name, or 'unknown' if no

Get statistics summary as dictionary.

Incrementally save data if threshold is reached. This enables item-level resume by saving progress periodically without waiting for t

Check if there are no items to process.

Load configuration from a Python file. Args: filename: Path to the config file.

Mark this item as rejected.

Validate and preprocess raw input messages into API-ready format. Args: msg: Raw input message (str, dict, or list). Ret

Convert raw input messages to unified dict list format. Args: inputs: Raw input. Returns: list: Processed in

Create Result for testing.

Set answer at index. Args: qa_idx: Index of the QA pair. value: New answer value. save_original: Whether

Set question at index. Args: qa_idx: Index of the QA pair. value: New question value. save_original: Whet

Shutdown and release resources.

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