Get LMDB key for an image path.

Get text length (character or token count).

Get shard ID for an image path.

Get shard ID and LMDB key in one MD5 call (avoid double hashing). Args: image_path: Image path string. Returns:

Check if length is out of bounds.

Load images for a batch from LMDB cache or disk. When ``use_lmdb_cache=True``, images are cached to LMDB first and then read back. W

Pre-encode all PIL images to base64 strings in parallel. Encodes images before multi-threaded API calls to avoid PIL thread-safety is

Save checkpoint to file.

Add a keep record (for kept items with reason).

Build all requests for a DataItem. Args: item: DataItem to build requests for. batch_qa: If True, combine al

Create a standard configuration from file paths. Args: file_paths: List of data file paths. sources: Optional mapping

Convenience function to denormalize data. Args: data: List of normalized items. Returns: List of items in their original sch

Encode PIL Image to base64 data URL string. Args: img: PIL Image object. target_size: Maximum dimension for thumbnail (-1 to skip

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

Read image data from cache. Args: image_path: Image path to read. Returns: bytes: Image data, or None if not

Get file_path to source mapping. Returns: dict: Mapping from file_path to source name.

Check if string contains sufficient proportion of alphanumeric characters. Args: text: Text to check. threshold: Minimum rat

Check if question and answer are in different languages. Useful for detecting cases where question is English but answer is Chinese. Ar

Log metrics for tracking (to file and wandb). Args: metrics: Dictionary of metric name -> value step: Option

Mark this item as not rejected (kept).

Convenience function to normalize data. Args: data: List of data items in any supported schema. Returns: List of items in st

Prepare message format for API interface. Args: inputs: Input message list. Returns: tuple: (formatted_messa

Register a format handler (decorator). Args: format_class: Format class to register. Returns: The registered

Register a module class. Can be used as a decorator or called directly: @REGISTRY.register_module() class MyClass: pa

Execute the runner in the configured mode.

Save data to file. Args: data: List of data dictionaries. file_path: Output file path. **kwargs: Format-s

Save YAML config and statistics.

Extract raw dicts from DataItems.

Build kwargs dict to reconstruct OpenAIAPI in worker processes. Returns: Dict of constructor arguments for OpenAIAPI.

Build HTTP headers using instance API key. Overrides BaseAPI._build_openai_headers to use self.key directly. Returns: He

Clear all message buffers to release memory. Args: messages: List of message objects to clear.

Run garbage collection and optionally trim glibc malloc arenas. Args: trim: If True, call malloc_trim(0) to return freed memory t

Extract conversations at specified QA indices.

Run async generation over all messages. Creates a shared :class:`aiohttp.ClientSession` with connection pooling and fires all request

Get path to shard directory.

Load prompt from file or return as-is.

Load statistics from processed file. Args: file_path: Path to processed file. source: Source name. Returns:

Parse QA index from various key formats (int, '0', 'qa_pair_0', etc.).

Re-index per-QA metadata fields based on new indices.

Save original content before rewriting. Args: qa_idx: QA pair index. field: 'answer' or 'question'.

Save YAML configuration file. Args: data_files: List of data file paths. yaml_path: Output YAML file path.

Update dataset-level statistics.

Add a rewrite record.

Build requests for specific QA indices only. Used by SelectiveMLLMRewriter for conditional rewriting. Args: ite

Clear data buffers.

Clean a single data item by removing temporary fields. Args: item: Data item dictionary. Returns: dict: Clea

Close all shards.

Denormalize a list of items back to original schema. Args: data: List of normalized items. Returns: List of

Record operator execution statistics. Args: name: Operator name (uses current if not specified) input_count:

Format all QA pairs for prompt.

Format request content. Args: item: DataItem to format. qa_idx: QA pair index (-1 for all QA). Retu

Generate YAML configuration files. Args: dataset_stats: Dataset statistics {path: {"json_size": N, "source": S}}. cnt

Generate response for given inputs. Must be implemented by subclasses.

Get source name from file path. Args: file_path: Path to data file. Returns: Source name or None if not foun

Group data by source file. Args: data_list: List of data items. Returns: dict: Mapping from source file path

Check if dataset has already been processed.

Initialize logger from config dict. Args: cfg: Logger configuration dict work_dir: Working directory for log

Initialize a single LMDB shard environment. Args: shard_path: Path to shard directory. map_size: Maximum database size in bytes.

Load YAML configuration file. Args: yaml_path: Path to YAML file. Returns: StandardConfig instance.

Load config and return all data file paths. Args: yaml_path: Path to YAML config file. data_root: Optional root direc

Log statistics summary. Args: title: Title for the stats section stats: Dictionary of stat name -> value

Normalize a list of items to standard schema. Args: data: List of data items. Returns: List of normalized it

Print current config in YAML format.

Print formatted statistics summary.

Check if conversation length is within bounds. This is an item-level filter, so we override process() directly.

Process a batch of items. Flow: 1. Build requests using RequestBuilder 2. Execute model 3. Parse responses a

Process a single message with retry logic. Args: message: Single processed message. all_kwargs: Arguments passed to g

Remove think tags from an answer.

Save data to files. Args: data_list: List of data items. is_rejected: Whether this is rejected data. appe

Save current config to YAML file. Args: save_file: Output file path.

Mark operator start.

Get all supported file extensions. Returns: list: List of supported extensions.

Update checkpoint with current progress.

Initialize the MLLM rewriter. Args: model: MLLM model instance. request_builder: RequestBuilder config dict.

Add timestamp to filename before extension. Example: logs/run.log -> logs/run_20251003123456.log

Apply filter decisions. Returns: - (item, None): Keep everything - (None, item): Reject everything

Apply rewrite decisions to modify item content.

Build payload for model.generate().

Build all requests for all items.

Build a sub-pipeline from config. Args: cfg: Sub-pipeline config dict. name: Name of the sub-pipeline.

Calculate exponential backoff delay with full jitter. Delegates to base class with default max_delay of 2.0 seconds. Args:

Clear module-level globals after workers finish.

Count <image> tags in text.

Count actual images in item.

Detect hint patterns and their positions. Returns: List of (start_pos, end_pos) tuples for found patterns, sorted by

Ensure LMDB manager is initialized (lazy initialization).

Execute a single operator and apply results. Returns: List of DataItems (both kept and rejected).

Extract a single turn from multi-turn data.

Find the configs root directory by looking for 'configs' folder. Args: start_path: Starting path to search from. Returns

Fix hint formatting by adding proper newlines. Returns: Tuple of (fixed_text, was_fixed).

Check if text contains instruction patterns.

Check if answer contains image/video/audio tags.

Set module-level globals before forking workers. Must be called in the main process before Pool creation. Args: messages: The full p

Initialize wandb logging.