docs : 三轮深度验证修复文档与代码不一致

- architecture.md: 修正 unwrap_model 返回类型、Config Optional 标注、方法签名错误、类名错误
- training.md: 补充 on_error 回调、修正训练循环顺序、补全策略参数、model.safetensors
- inference.md: 修正 GenerationRequest 参数顺序、async 语法、KVCache 描述、temperature 约束
- dataflow.md: 补充 Store.load/fetch 流程、修正可选参数默认值
- README/params: 多 GPU 示例补全 --parallel_mode、文档表补充 preprocessing.md
- preprocessing.md: Chat 模式算法补全 BOS token 步骤

README.md

@@ -82,6 +82,7 @@ export CUDA_VISIBLE_DEVICES=0,1,2,3
 
 nohup python scripts/tools/train.py \
     --nprocs=4 \
+    --parallel_mode=ddp \
     --train_type=seq \
     --data_root_path=/path/to/dataset \
     --param_path=/path/to/model \
@@ -108,8 +109,8 @@ Full reference at [Parameter Guide](assets/docs/params.md).
 ```bash
 python scripts/tools/generate.py \
     --param_path /path/to/model \
-    --input_json_file /path/to/input.json \
-    --output_json_file /path/to/output.json
+    --input_json_file /path/to/input.jsonl \
+    --output_json_file /path/to/output.jsonl
 ```
 
 #### Docker
@@ -224,6 +225,7 @@ Watch a video walkthrough on [bilibili](https://www.bilibili.com/video/BV1fuLB6y
 | [Training](./assets/docs/training.md) | Training loop, strategies & formulas |
 | [Inference](./assets/docs/inference.md) | KVCache, continuous batching, sampling & HTTP API |
 | [Data Flow](./assets/docs/dataflow.md) | Data pipeline, storage backends & dataset architecture |
+| [Preprocessing](./assets/docs/preprocessing.md) | Declarative JSON-driven data preprocessing |
 
 ### Contributing
 

assets/docs/README-zh-CN.md

@@ -88,6 +88,7 @@ export CUDA_VISIBLE_DEVICES=0,1,2,3
 
 nohup python scripts/tools/train.py \
     --nprocs=4 \
+    --parallel_mode=ddp \
     --train_type=seq \
     --data_root_path=/path/to/dataset \
     --param_path=/path/to/model \
@@ -114,8 +115,8 @@ nohup python scripts/tools/train.py \
 ```bash
 python scripts/tools/generate.py \
     --param_path /path/to/model \
-    --input_json_file /path/to/input.json \
-    --output_json_file /path/to/output.json
+    --input_json_file /path/to/input.jsonl \
+    --output_json_file /path/to/output.jsonl
 ```
 
 #### Docker
@@ -230,6 +231,7 @@ python scripts/demo/generate_ar.py
 | [训练文档](./training.md) | 训练循环、策略与公式 |
 | [推理文档](./inference.md) | KVCache、连续批处理、采样与 HTTP API |
 | [数据流程](./dataflow.md) | 数据管道、存储后端与数据集架构 |
+| [数据预处理](./preprocessing.md) | 声明式 JSON 驱动数据预处理 |
 
 ### 贡献
 

assets/docs/architecture.md

@@ -8,6 +8,8 @@ classDiagram
         class BaseConfig {
             +to_dict() Dict
             +from_dict(d) Self
+            +from_json(path) Self
+            +to_json(path)
         }
 
         class BaseModelConfig {
@@ -17,42 +19,42 @@ classDiagram
         }
 
         class AutoRegressiveLMConfig {
-            +int vocab_size
-            +int dim
-            +int n_layers
-            +float norm_eps
-            +int dim_ffn
+            +Optional[int] vocab_size
+            +Optional[int] dim
+            +Optional[int] n_layers
+            +Optional[float] norm_eps
+            +Optional[int] dim_ffn
             +Optional[bool] tie_weight
             +Optional[dict] rope_scaling
-            +int max_len
-            +float rope_theta
+            +Optional[int] max_len
+            +Optional[float] rope_theta
             +str attn_type
-            +int n_heads
-            +int n_kv_heads
-            +bool use_qk_norm
-            +bool use_gated_attention
+            +Optional[int] n_heads
+            +Optional[int] n_kv_heads
+            +Optional[bool] use_qk_norm
+            +Optional[bool] use_gated_attention
             +Optional[int] kv_lora_rank
             +Optional[int] qk_nope_head_dim
             +Optional[int] qk_rope_head_dim
             +str ffn_type
-            +int n_routed_experts
-            +int n_shared_experts
-            +int n_activated_experts
+            +Optional[int] n_routed_experts
+            +Optional[int] n_shared_experts
+            +Optional[int] n_activated_experts
             +Optional[str] topk_method
         }
 
         class EncoderConfig {
-            +int vocab_size
-            +int dim
-            +int n_layers
-            +float norm_eps
-            +int dim_ffn
-            +int max_len
-            +float rope_theta
-            +int n_heads
-            +int n_kv_heads
-            +bool use_qk_norm
-            +bool use_gated_attention
+            +Optional[int] vocab_size
+            +Optional[int] dim
+            +Optional[int] n_layers
+            +Optional[float] norm_eps
+            +Optional[int] dim_ffn
+            +Optional[int] max_len
+            +Optional[float] rope_theta
+            +Optional[int] n_heads
+            +Optional[int] n_kv_heads
+            +Optional[bool] use_qk_norm
+            +Optional[bool] use_gated_attention
             +Optional[dict] rope_scaling
             +Optional[str] pooling_type
             +Optional[bool] normalize_embeddings
@@ -64,6 +66,38 @@ classDiagram
             +load(raw) BaseConfig
         }
 
+        class InputConfig {
+            +str type
+            +str messages_key
+            +str prompt_key
+            +str response_key
+            +str text_key
+        }
+
+        class ProcessingConfig {
+            +int max_seq_len
+            +int min_chars
+            +int max_chars
+            +bool deduplicate
+            +Optional[int] max_items
+        }
+
+        class OutputConfig {
+            +Optional[str] domain_key
+            +str storage_format
+            +int max_tokens_per_shard
+        }
+
+        class PipelineConfig {
+            +int version
+            +InputConfig input
+            +dict mask
+            +str mask_default
+            +ProcessingConfig preprocessing
+            +OutputConfig output
+            +from_dict(d) Self
+        }
+
         class TrainConfig {
             +Callable[[], nn.Module] model_fn
             +str strategy
@@ -312,10 +346,29 @@ classDiagram
         }
     }
 
+    namespace preprocessing {
+        class BaseMaskBuilder {
+            <<abstract>>
+            +build(item, config, tokenizer) Optional[dict]
+        }
+
+        class ChatMaskBuilder {
+            +build(item, config, tokenizer) Optional[dict]
+        }
+
+        class InstructionMaskBuilder {
+            +build(item, config, tokenizer) Optional[dict]
+        }
+
+        class TextMaskBuilder {
+            +build(item, config, tokenizer) Optional[dict]
+        }
+    }
+
     namespace tokenize {
         class AutoTokenizer {
             +vocab_size int
-            +encode(tokens, out_ids, is_pretokenized, add_special_tokens) List[int]
+            +encode(tokens, out_ids, is_pretokenized, add_special_tokens) List
             +decode(tokens, skip_special_tokens) str
             +__getattr__(name) Any (bos_id, eos_id, pad_id, stop_ids)
             +apply_chat_template(messages, system_prompt, tokenize, add_generation_prompt) Union[str, List[int]]
@@ -346,14 +399,20 @@ classDiagram
             +create(name, *args, **kwargs) T
             +list_registered() list
         }
+
+        class MaskBuilderFactory {
+            +Registry _registry
+            +register(name) decorator
+            +create(input_type, config, tokenizer) BaseMaskBuilder
+        }
     }
 
     namespace trainer {
         class Trainer {
             +TrainConfig train_config
             +List[TrainCallback] callbacks
-            +train(checkpoint)
-            +_get_default_callbacks() List[TrainCallback]
+            +train(resume_dir)
+            -_get_default_callbacks() List[TrainCallback]
         }
 
         class TrainContext {
@@ -383,8 +442,12 @@ classDiagram
         }
 
         class BaseStrategy {
-            +Union[Callable, nn.Module] model
+            +Callable model
+            +Optional[BaseExecutor] executor
+            +Optional[Callable] model_fn
+            +dict extra_kwargs
             +str device
+            +__call__(batch) Tensor
             +compute_loss(batch) Tensor
         }
 
@@ -425,6 +488,8 @@ classDiagram
         class BaseScheduler {
             +get_lr() List[float]
             +step()
+            +state_dict() dict
+            +load_state_dict(d)
         }
 
         class SchedulerFactory {
@@ -436,6 +501,7 @@ classDiagram
         class CosineScheduler {
             +int warmup_steps
             +int lr_decay_steps
+            +int total_steps
             +float min_rate
         }
 
@@ -474,11 +540,11 @@ classDiagram
             +int interval
             +bool weight_only
             +Callable save_extra_fn
-            +_save_checkpoint(context)
+            -_save_checkpoint(context)
             +on_batch_end(context)
             +on_train_end(context)
             +on_error(context)
-            +save_extra(context)$
+            +save_extra(context) dict$
         }
 
         class ProgressBarCallback {
@@ -491,7 +557,7 @@ classDiagram
         }
 
         class MetricLoggerCallback {
-            +str log_dir
+            +Path log_dir
             +int save_interval
             +int log_interval
             +List[str] metrics
@@ -501,7 +567,7 @@ classDiagram
         }
 
         class ValidationCallback {
-            +_run_validation(context)
+            -_run_validation(context)
             +on_optimizer_step(context)
         }
 
@@ -517,7 +583,7 @@ classDiagram
             +float weight_decay
             +bool nesterov
             +int ns_steps
-            +float adamw_lr
+            +Optional[float] adamw_lr
             +tuple adamw_betas
             +float adamw_eps
             +float adamw_wd
@@ -634,7 +700,7 @@ classDiagram
         class Task {
             +str task_id
             +List prompt_ids
-            +int max_tokens
+            +Optional[int] max_tokens
             +float temperature
             +float top_p
             +int top_k
@@ -643,8 +709,8 @@ classDiagram
             +int input_tokens
             +int output_tokens
             +float arrival_time
-            +float finish_time
-            +Callable stream_callback
+            +Optional[float] finish_time
+            +Optional[Callable] stream_callback
             +int next_pos
             +is_finished(stop_ids) bool
         }
@@ -671,6 +737,11 @@ classDiagram
             +activate(task)
             +return_to_waiting(tasks)
             +get_active_tasks() List[Task]
+            +has_work() bool
+            +wait_for_tasks(timeout)
+            +get_waiting_tasks() List[Task]
+            +clear_queues()
+            +wake()
             +get_stats() Dict
         }
 
@@ -760,7 +831,7 @@ classDiagram
 
         class ResponseBuilder {
             <<abstract>>
-            +prepare(request, engine) Tuple[str, GenContext, List[str]]
+            +prepare(request, tokenizer) Tuple[str, GenContext, List[str]]
             +format_stream_start(ctx) List[str]
             +format_chunk(token) str
             +format_stream_end(ctx, stop) List[str]
@@ -768,7 +839,7 @@ classDiagram
         }
 
         class OpenAIResponseBuilder {
-            +prepare(request, engine) Tuple
+            +prepare(request, tokenizer) Tuple
             +format_stream_start(ctx) List[str]
             +format_chunk(token) str
             +format_stream_end(ctx, stop) List[str]
@@ -776,7 +847,7 @@ classDiagram
         }
 
         class AnthropicResponseBuilder {
-            +prepare(request, engine) Tuple
+            +prepare(request, tokenizer) Tuple
             +format_stream_start(ctx) List[str]
             +format_chunk(token) str
             +format_stream_end(ctx, stop) List[str]
@@ -787,12 +858,13 @@ classDiagram
             +request
             +engine
             +builder: ResponseBuilder
-            +handle() Union[StreamingResponse, Dict]
-            -_handle_stream(agen, ctx, stops) StreamingResponse
-            -_handle_non_stream(agen, ctx, stops) Dict
+            +async handle() Union[StreamingResponse, Dict]
+            -_handle_stream(agen, ctx, stop_sequences) StreamingResponse
+            -async _handle_non_stream(agen, ctx, stop_sequences) Dict
         }
 
         class StopChecker {
+            +__init__(sequences)
             +check(text) Optional[str]
         }
 
@@ -804,6 +876,12 @@ classDiagram
             +int completion_tokens
         }
 
+        class StopInfo {
+            +Optional[str] matched
+            +str body
+            +str yielded
+        }
+
         class app {
             <<singleton>>
             +FastAPI app
@@ -829,14 +907,14 @@ classDiagram
     }
 
     namespace parallel {
-        class Functions {
+        class setup {
             <<module>>
             +spawn_parallel_fn(func, world_size, backend, master_addr, master_port, device_type, start_method, **kwargs)
-            +setup_parallel(rank, world_size, backend, master_addr, master_port, device_type)
+            +setup_parallel(rank, world_size, backend, master_addr, master_port, device_type) contextmanager
             +get_current_device() str
             +get_world_size() int
             +get_rank() int
-            +only_on_rank(rank, sync) decorator
+            +only_on_rank(rank, sync=False) decorator
         }
 
         class GradientState {
@@ -847,6 +925,7 @@ classDiagram
         class AccumOptimizer {
             +Optimizer optimizer
             +GradientState gradient_state
+            +param_groups (property)
             +step(closure)
             +zero_grad()
             +state_dict() dict
@@ -867,7 +946,7 @@ classDiagram
             +prepare(model, optimizer, dataloader, scheduler) tuple
             +accumulate(model) context manager
             +backward(loss)
-            +unwrap_model(model) nn.Module
+            +unwrap_model(model) dict
             +sync_gradients (property) bool
             +grad_accum_steps (property) int
         }
@@ -876,14 +955,14 @@ classDiagram
         }
 
         class DDPExecutor {
-            +_prepare_model(model) nn.Module
-            +_no_sync(model) context manager
-            +unwrap_model(model) nn.Module
+            -_prepare_model(model) nn.Module
+            -_no_sync(model) context manager
+            +unwrap_model(model) dict
         }
 
         class FSDPExecutor {
-            +_prepare_model(model) nn.Module
-            +unwrap_model(model) nn.Module
+            -_prepare_model(model) nn.Module
+            +unwrap_model(model) dict
         }
 
         class ExecutorFactory {
@@ -899,11 +978,25 @@ classDiagram
         }
 
         class ColumnParallelLinear {
+            +int in_features
+            +int out_features
+            +int out_features_per_rank
+            +bool gather_results
+            +Parameter weight
+            +Optional[Parameter] bias
             +forward(x) Tensor
+            +load_state_dict(state_dict)
         }
 
         class RowParallelLinear {
+            +int in_features
+            +int out_features
+            +int in_features_per_rank
+            +bool reduce_results
+            +Parameter weight
+            +Optional[Parameter] bias
             +forward(x) Tensor
+            +load_state_dict(state_dict)
         }
     }
 
@@ -938,6 +1031,10 @@ classDiagram
     AutoModel <|-- EmbeddingEncoder
     BaseConfig <|-- BaseModelConfig
     BaseConfig <|-- TrainConfig
+    BaseConfig <|-- InputConfig
+    BaseConfig <|-- ProcessingConfig
+    BaseConfig <|-- OutputConfig
+    BaseConfig <|-- PipelineConfig
     BaseModelConfig <|-- AutoRegressiveLMConfig
     BaseModelConfig <|-- EncoderConfig
     BaseFactory <|-- AutoModel
@@ -950,11 +1047,15 @@ classDiagram
     BaseFactory <|-- StoreFactory
     BaseFactory <|-- ExecutorFactory
     BaseFactory <|-- ConfigFactory
+    BaseFactory <|-- MaskBuilderFactory
     BaseExecutor <|-- NoneExecutor
     BaseExecutor <|-- DDPExecutor
     BaseExecutor <|-- FSDPExecutor
     ResponseBuilder <|-- OpenAIResponseBuilder
     ResponseBuilder <|-- AnthropicResponseBuilder
+    BaseMaskBuilder <|-- ChatMaskBuilder
+    BaseMaskBuilder <|-- InstructionMaskBuilder
+    BaseMaskBuilder <|-- TextMaskBuilder
 
     %% --- Composition (strong ownership, part destroyed with whole) ---
     KVCache *-- PagePool
@@ -994,6 +1095,8 @@ classDiagram
 
     %% --- Dependency (uses temporarily) ---
     TrainConfig ..> BaseStrategy : selects
+    PipelineConfig ..> MaskBuilderFactory : selects
+    MaskBuilderFactory ..> BaseMaskBuilder : creates
     StrategyFactory ..> BaseStrategy : creates
     SchedulerFactory ..> BaseScheduler : creates
     DatasetFactory ..> BaseDataset : creates
@@ -1046,7 +1149,8 @@ classDiagram
 
 | Module | Components | Description |
 |--------|------------|-------------|
-| **astrai.config** | BaseConfig, BaseModelConfig, AutoRegressiveLMConfig, EncoderConfig, ConfigFactory, TrainConfig | Configuration management (to_dict/from_dict, to_file/from_file) |
+| **astrai.config** | BaseConfig, BaseModelConfig, AutoRegressiveLMConfig, EncoderConfig, ConfigFactory, TrainConfig, PipelineConfig, InputConfig, ProcessingConfig, OutputConfig | Configuration management (to_dict/from_dict, to_file/from_file, from_json/to_json) |
+| **astrai.preprocessing** | BaseMaskBuilder, MaskBuilderFactory, ChatMaskBuilder, InstructionMaskBuilder, TextMaskBuilder, Pipeline, filter_by_length, dedup_signature | Declarative JSON-driven data preprocessing |
 | **astrai.dataset** | BaseDataset–GRPODataset, Store–MmapStore, StoreFactory, ResumableDistributedSampler, DatasetFactory | Dataset loading and management |
 | **astrai.serialization** | Checkpoint | Model serialization |
 | **astrai.model** | AutoModel, AutoRegressiveLM, EmbeddingEncoder, DecoderBlock, GQA, MLA, MLP, DeepSeekMoE, AttnFactory, FFNFactory, RMSNorm, Linear, RotaryEmbedding, Embedding | Neural network model |
@@ -1070,14 +1174,14 @@ classDiagram
 | **Observer** | `TrainCallback`, callback implementations | Training process monitoring |
 | **Context** | `TrainContext` | Unified training state bag |
 | **Object Pool** | `Allocator`, `PagePool` | Page-based KV cache with LRU eviction |
-| **Executor** | `BaseExecutor`, `NoneExecutor`, `DDPExecutor` | Gradient accumulation & model distribution |
+| **Executor** | `BaseExecutor`, `NoneExecutor`, `DDPExecutor`, `FSDPExecutor` | Gradient accumulation & model distribution |
 | **Storage** | `Store`, `H5Store`, `MmapStore` | Format-agnostic data access with multi-segment support |
 | **Producer-Consumer** | `InferenceScheduler`, `Task`, queues | Continuous batching |
 | **AutoModel Registry** | `AutoModel`, `AutoRegressiveLM`, `EmbeddingEncoder` | Model-type dynamic loading |
 
 ## Core Relationships
 
-1. **Config → Training**: `TrainConfig` holds model, dataset, optimizer_fn, scheduler_fn, `parallel_mode`, `executor_kwargs`
+1. **Config → Training**: `TrainConfig` holds `model_fn`, `dataset`, `optimizer_fn`, `scheduler_fn`, `parallel_mode`, `executor_kwargs`
 2. **Training Flow**: `Trainer` → `TrainContextBuilder` → `TrainContext`, uses `BaseStrategy` for loss, `BaseExecutor` for gradient accumulation + model distribution
 3. **Strategy Selection**: `StrategyFactory` creates strategy by `train_type`
 4. **Executor Selection**: `ExecutorFactory.create(cfg.parallel_mode, grad_accum_steps=cfg.grad_accum_steps, **cfg.executor_kwargs)` → `NoneExecutor` / `DDPExecutor` / `FSDPExecutor`
@@ -1089,4 +1193,4 @@ classDiagram
 10. **AutoModel**: `from_pretrained()` loads `config.json` + `model.safetensors`, `_disable_random_init` replaces `nn.init.*` with no-ops
 11. **Protocols**: `OptimizerProtocol` / `SchedulerProtocol` — structural subtyping for `AccumOptimizer` / `AccumScheduler` wrappers
 
-> Document Update Time: 2026-05-28
+> Document Update Time: 2026-05-30

assets/docs/dataflow.md

@@ -5,7 +5,7 @@ This document describes the data pipeline: from raw text to model input tensors.
 ## Overview
 
 ```
-Raw Text → AutoTokenizer → Token IDs → .h5/.bin → Dataset → Sampler → DataLoader → Training/Inference
+Raw Text → AutoTokenizer → Token IDs → .h5/.bin → Store.load() → Store.fetch() → Dataset → Sampler → DataLoader → Training/Inference
 ```
 
 ## Data Preparation
@@ -33,14 +33,21 @@ H5 backend supports shared memory via `.share_memory_()`. Bin (mmap) uses OS pag
 ## Dataset Architecture
 
 ```
-DatasetFactory.load(train_type, load_path, window_size, stride, storage_type)
-  → StoreFactory.create(detect_format(path))
-    → Store._data[Dict[str, List[Tensor]]] + _cum[Dict[str, List[int]]]
-      → BaseDataset.__getitem__(idx)
-        → sliding window [begin, end) via get_index(idx)
+DatasetFactory.load(train_type, load_path, window_size, stride=None, storage_type=None)
+  → BaseDataset.load(load_path, storage_type=None)
+    → detect_format(load_path)
+    → StoreFactory.create(storage_type)
+    → Store.load(load_path)
+      → H5Store._normalize() / MmapStore._normalize()
+        → Store._data[Dict[str, List[Tensor]]] + _cum[Dict[str, List[int]]]
+          → BaseDataset.__getitem__(idx)
+            → get_index(idx) → [begin, end)
+            → Store.fetch(begin, end, keys) → Tensor / Dict[str, Tensor]
 ```
 
-`window_size` = max input length, `stride` = step between consecutive samples (defaults to `window_size`).
+`window_size` = max input length, `stride` = step between consecutive samples (defaults to `window_size`, optional). `storage_type` defaults to `None` (auto-detect via `detect_format`).
+
+`Store.fetch(begin, end, keys)` accepts a single key (`str`) returning a `Tensor`, or a list of keys returning `Dict[str, Tensor]`. Internally uses `bisect` across multi-segment tensors. Raises `RuntimeError("Store not loaded")` if called before `load()`.
 
 ## Sampler
 
@@ -54,4 +61,4 @@ DatasetFactory.load(train_type, load_path, window_size, stride, storage_type)
 
 Standard PyTorch `DataLoader` with configurable `batch_size`, `num_workers`, `pin_memory`, `prefetch_factor`. Sampler produces indices; dataloader fetches tensor batches via `__getitem__`.
 
-> Document Update Time: 2026-05-28
+> Document Update Time: 2026-05-30

assets/docs/inference.md

@@ -12,7 +12,7 @@ RoPE is applied **before** KV cache write, not after — otherwise position enco
 
 ## KVCache System
 
-Six classes working together:
+Six classes (plus two helpers) working together:
 
 ```
 KVCache (facade)
@@ -43,7 +43,8 @@ KVCache (facade)
 BaseSamplingStrategy (ABC)
   ├── TemperatureStrategy
   ├── TopKStrategy
-  └── TopPStrategy
+  ├── TopPStrategy
+  └── SamplingPipeline
 ```
 
 `SamplingPipeline` composes them: Temperature → Top-K → Top-P → softmax → multinomial.  
@@ -73,7 +74,9 @@ Adding a protocol = one builder file, no handler subclassing needed.
 InferenceEngine
   ├── generate(prompt, stream, ...) → str | List[str] | Generator
   ├── generate_with_request(req)    → same
-  └── generate_async(prompt, ...)   → AsyncGenerator
+  ├── generate_async(prompt, ...)   → AsyncGenerator
+  ├── get_stats()                   → Dict
+  └── shutdown()
 ```
 
 `GenerateResult` uses `Condition` for non-streaming (`wait_completion()`) and `Event` for streaming (`wait()`). Stream callback is `cb(token)`.
@@ -124,9 +127,9 @@ Supports `stop_sequences` and streaming via `event: content_block_delta`.
 | Param | Type | Default | Description |
 |-------|------|---------|-------------|
 | `messages` | List[dict] | required | Chat messages (role, content) |
-| `temperature` | float | 1.0 | Sampling temperature (>= 0.0) |
-| `top_p` | float | 1.0 | Nucleus threshold |
 | `top_k` | int | 50 | Top-k count |
+| `top_p` | float | 1.0 | Nucleus threshold |
+| `temperature` | float | 1.0 | Sampling temperature (> 0.0) |
 | `max_tokens` | Optional[int] | None | Max generation length |
 | `stream` | bool | False | Stream output |
 
@@ -142,7 +145,8 @@ engine.generate("Hello", stream=True)           # -> Generator[str]
 engine.generate(["A", "B"], stream=True)        # -> Generator[Tuple[int, str]]
 
 # Async
-await engine.generate_async("Hello", ...)       # -> AsyncGenerator[str]
+async for token in engine.generate_async("Hello", ...):    # -> AsyncGenerator[str]
+    print(token)
 ```
 
-> Document Update Time: 2026-05-28
+> Document Update Time: 2026-05-30

assets/docs/params.md

@@ -75,6 +75,7 @@ export CUDA_VISIBLE_DEVICES=0,1,2,3
 
 nohup python scripts/tools/train.py \
     --nprocs=4 \
+    --parallel_mode=ddp \
     --train_type=seq \
     --data_root_path=/path/to/dataset \
     --param_path=/path/to/model \

assets/docs/preprocessing.md

@@ -0,0 +1,228 @@
+# Preprocessing Pipeline
+
+Declarative JSON-driven data preprocessing. No code needed -- describe your input format and mask rules in a config file, the engine does the rest.
+
+## Philosophy
+
+| Component | Responsibility |
+|-----------|---------------|
+| `tokenizer_config.json` (`chat_template`) | Formatting -- how roles become tokens |
+| `pipeline.json` (`mask`) | Masking -- which roles participate in training |
+
+The two are fully decoupled. A single config file captures the entire pipeline, reusable and version-controllable. Extension is via factory registration (`@MaskBuilderFactory.register`) -- no need to touch existing code.
+
+## Quick Start
+
+### SFT Chat
+
+```json
+{
+  "version": 1,
+  "input": {
+    "type": "chat",
+    "messages_key": "messages"
+  },
+  "mask": {
+    "system": "mask",
+    "user": "mask",
+    "assistant": "train"
+  },
+  "mask_default": "mask",
+  "preprocessing": {
+    "max_seq_len": 2048,
+    "deduplicate": true
+  },
+  "output": {
+    "domain_key": "source",
+    "storage_format": "bin",
+    "max_tokens_per_shard": 100000000
+  }
+}
+```
+
+Three lines of mask rules cover the most common SFT case: train on assistant turns, mask everything else.
+
+### Instruction Tuning
+
+```json
+{
+  "version": 1,
+  "input": {
+    "type": "instruction",
+    "prompt_key": "instruction",
+    "response_key": "output"
+  },
+  "mask": {
+    "prompt": "mask",
+    "response": "train"
+  },
+  "mask_default": "mask",
+  "preprocessing": {
+    "max_seq_len": 2048
+  },
+  "output": {
+    "storage_format": "bin"
+  }
+}
+```
+
+Mask splits at the prompt/response field boundary.
+
+### Pretraining
+
+```json
+{
+  "version": 1,
+  "input": {
+    "type": "text",
+    "text_key": "content"
+  },
+  "mask": {},
+  "preprocessing": {
+    "max_seq_len": 2048,
+    "min_chars": 50
+  },
+  "output": {
+    "storage_format": "bin"
+  }
+}
+```
+
+No mask -- train on all tokens.
+
+### Run
+
+```bash
+python scripts/tools/preprocess.py data/*.jsonl -o output/ -c sft.json
+```
+
+## Configuration Reference
+
+### `input`
+
+| Field | Type | Required | Default | Description |
+|-------|------|----------|---------|-------------|
+| `type` | string | yes | `"chat"` | Format: `"chat"`, `"instruction"`, or `"text"` |
+| `messages_key` | string | no | `"messages"` | JSON key for messages array (chat) |
+| `prompt_key` | string | no | `"prompt"` | JSON key for prompt field (instruction) |
+| `response_key` | string | no | `"response"` | JSON key for response field (instruction) |
+| `text_key` | string | no | `"text"` | JSON key for text field |
+
+### `mask`
+
+A map of `{role_or_field: "mask" | "train"}`. The engine uses this to build `loss_mask`:
+
+- `"mask"` -- tokens in this span are ignored during training (`loss_mask=0`)
+- `"train"` -- tokens in this span contribute to the loss (`loss_mask=1`)
+
+For chat mode, keys are role names (`system`, `user`, `assistant`, ...).
+For instruction mode, keys are `"prompt"` and `"response"`.
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `mask` | dict | `{}` | Role/field to action mapping |
+| `mask_default` | string | `"mask"` | Default action for unlisted roles |
+
+### `preprocessing`
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `max_seq_len` | int | `2048` | Maximum token length; truncated if exceeded |
+| `min_chars` | int | `50` | Minimum character length; dropped if shorter (text mode only) |
+| `max_chars` | int | `2000000` | Maximum character length; dropped if longer (text mode only) |
+| `deduplicate` | bool | `true` | Remove exact duplicates via MD5 of first 200 chars |
+| `max_items` | int or null | `null` | Maximum items to process; `null` = unlimited |
+
+### `output`
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `domain_key` | string or null | `null` | JSON key for domain grouping; `null` = all output to `__default__` |
+| `storage_format` | string | `"bin"` | `"bin"` (mmap, zero-copy) or `"h5"` (HDF5) |
+| `max_tokens_per_shard` | int | `100000000` | Max tokens per output shard |
+
+## Mask Algorithm
+
+### Chat Mode (role-span tracking)
+
+For each message in the `messages` array:
+
+1. Prepend BOS token (position 0, always masked)
+2. Render through the chat template for that single message
+3. Encode the rendered text, record token span `(start, end, role)`
+4. Concatenate all spans — special tokens from the chat template naturally prevent BPE merging across message boundaries
+5. Fill `loss_mask` from the mask rules
+
+**Multi-turn example**:
+
+```
+Data:
+  [system: "You are helpful."]
+  [user: "What is 2+2?"]
+  [assistant: "4"]
+  [user: "What is 3+3?"]
+  [assistant: "6"]
+
+Config:
+  "mask": {"system": "mask", "user": "mask", "assistant": "train"}
+
+Result:
+  tokens:  <bos> [system span] [user span] [assistant:4 span] [user span] [assistant:6 span]
+  mask:      0       0            0              1               0             1
+```
+
+Both assistant turns are trained. All system and user tokens are masked.
+
+### Instruction Mode (field boundary)
+
+Encode the prompt and response fields independently, then split the mask at the field boundary.
+
+- `"prompt": "mask", "response": "train"` -- mask the left half, train the right half
+- `"prompt": "train", "response": "mask"` -- the reverse
+
+### Text Mode (no mask)
+
+Pure tokenization. No `loss_mask` is produced. Used for pretraining.
+
+## Output Layout
+
+```
+output_dir/
+  __default__/              # when domain_key is null
+    meta.json               # {"sequence": {"shape": [N], "dtype": "int64"}, ...}
+    sequence.bin            # int64 raw bytes, mmap-able for zero-copy reads
+    loss_mask.bin           # int64 raw bytes
+  wiki/                     # when domain_key="source" and item["source"]="wiki"
+    meta.json
+    sequence.bin
+    loss_mask.bin
+```
+
+## Extension
+
+Register a custom builder for new formats:
+
+```python
+from astrai.preprocessing.builder import BaseMaskBuilder, MaskBuilderFactory
+
+@MaskBuilderFactory.register("my_format")
+class MyFormatBuilder(BaseMaskBuilder):
+    def build(self, item: dict, config, tokenizer) -> dict | None:
+        # Return {"ids": [...], "loss_mask": [...], "domain": "..."}
+        # Return None to skip this item
+        ...
+```
+
+Then set `"input": {"type": "my_format"}` in your config.
+
+## Compared to Old Pipeline
+
+| Old (`astrai.preprocess.Pipeline`) | New (`astrai.preprocessing.pipeline.Pipeline`) |
+|---|---|
+| Configured via constructor arguments | Configured via JSON file |
+| Hardcoded `_transform_chat` / `_transform_text` | Factory-registered `Builder` with declarative mask rules |
+| Auto-detects format via magic key lists | Explicit `input.type` declaration |
+| Double-encodes (full + prompt), uses length diff for mask | Single-encode with role-span tracking |
+| Only trains the last assistant turn | Configurable: multi-turn, single-turn, or no mask |
+
+> Document Update Time: 2026-05-30

assets/docs/training.md

@@ -1,38 +1,5 @@
 # Training
 
-## Model Architecture
-
-The model uses a decoder-only Transformer with **GQA** (Grouped Query Attention) and optional **MLA** (Multi-head Latent Attention). 1.0 billion parameters, Chinese–English bilingual.
-
-```mermaid
-flowchart TB
-    subgraph Layers["Transformer Layers"]
-        direction TB
-        A[Input Embedding] --> B[Transformer Block\nLayer 1]
-        B --> C[Transformer Block\nLayer ...]
-        C --> D[Transformer Block\nLayer ...]
-        D --> E[RMSNorm]
-        E --> F[Linear]
-        F --> G[SoftMax]
-    end
-
-    subgraph TransformerBlock["Transformer Block"]
-        direction TB
-        H[x] --> I[RMSNorm]
-        I --> J[Linear → Q/K/V]
-        J --> K[Q]; J --> L[K]; J --> M[V]
-        K --> N[RoPE]; L --> O[RoPE]
-        N --> P["Q @ K^T / sqrt(d)"]; O --> P
-        P --> Q[Masked SoftMax]; Q --> R[S @ V]; M --> R
-        R --> S[Linear]; S --> T[+]; H --> T
-        T --> U[RMSNorm]
-        U --> V["Linear (gate)"]; U --> W["Linear (up)"]
-        V --> X[SiLU]; X --> Y[×]; W --> Y
-        Y --> Z["Linear (down)"]; Z --> AA[+]; T --> AA
-        AA --> BB[x']
-    end
-```
-
 ### Autoregression
 
 Given a token sequence, the model predicts the probability of the next token. Each generated token is appended to the input and fed back, repeating until an end-of-sequence token or max length.
@@ -69,14 +36,16 @@ Two-level loop: **epoch** → **batch**. Optimizer step fires every `grad_accum_
 
 ```
 on_train_begin
+  model.train()
   on_epoch_begin
     for batch in dataloader:
       on_batch_begin
       with executor.accumulate(model):
-        loss = strategy(batch)
+        loss = strategy.compute_loss(batch)
+        context.loss = loss.item()
         stand_loss = loss / executor.grad_accum_steps
         executor.backward(stand_loss)
-        iteration += 1
+        context.iteration += 1
         on_batch_end
 
         if executor.sync_gradients:
@@ -94,9 +63,13 @@ on_train_end
 | Hook | Fires | Default callback |
 |------|-------|-----------------|
 | `on_train_begin` | Before training starts | `GradientCheckpointingCallback` |
+| `on_epoch_begin` | Start of each epoch | `ProgressBarCallback` |
+| `on_batch_begin` | Every batch | — |
 | `on_optimizer_step` | Every accumulation window | `GradientClippingCallback`, `ValidationCallback` |
 | `on_batch_end` | Every batch | `CheckpointCallback`, `MetricLoggerCallback`, `ProgressBarCallback` |
-| `on_train_end` | Training ends | `CheckpointCallback`, `MetricLoggerCallback` (final save) |
+| `on_epoch_end` | End of each epoch | `ProgressBarCallback` |
+| `on_error` | On exception during training | `CheckpointCallback`, `MetricLoggerCallback` |
+| `on_train_end` | Training ends (always via finally) | `CheckpointCallback`, `MetricLoggerCallback`, `GradientCheckpointingCallback` |
 
 Default callbacks (in order): `gradient_checkpointing` (activation checkpointing, optional), `checkpoint` (safetensors, rank-0), `metric_logger` (JSONL, rank-0), `progress_bar` (tqdm), `gradient_clipping`, `validation` (periodic validation on val_dataset).
 
@@ -110,7 +83,7 @@ $$
 L_{\text{PT}} = -\sum_{t=1}^{T} \log P(x_t \mid x_{\lt t}; \theta)
 $$
 
-Keys: `input_ids`, `target_ids`
+Keys: `input_ids`, `target_ids`. Optional: `label_smoothing`.
 
 ### SFT (Supervised Fine-Tuning)
 
@@ -120,7 +93,7 @@ $$
 L_{\text{SFT}} = -\sum_{t=P+1}^{P+L} \log P(s_t \mid s_{\lt t}; \theta)
 $$
 
-Keys: `input_ids`, `target_ids`, `loss_mask`
+Keys: `input_ids`, `target_ids`, `loss_mask`. Optional: `label_smoothing`.
 
 ### DPO (Direct Preference Optimization)
 
@@ -130,7 +103,7 @@ $$
 L_{\text{DPO}} = -\mathbb{E}\left[\log\sigma\left(\beta\log\frac{\pi_\theta(y_w\mid x)}{\pi_{\text{ref}}(y_w\mid x)} - \beta\log\frac{\pi_\theta(y_l\mid x)}{\pi_{\text{ref}}(y_l\mid x)}\right)\right]
 $$
 
-Parameters: `beta=0.1`. Keys: `chosen`, `rejected`, `chosen_mask`, `rejected_mask`.
+Parameters: `beta=0.1`, `reduction="mean"`. Keys: `chosen`, `rejected`, `chosen_mask`, `rejected_mask`.
 
 ### GRPO (Group Relative Policy Optimization)
 
@@ -144,7 +117,7 @@ $$
 L_{\text{GRPO}} = -\mathbb{E}\left[\min\left(\frac{\pi_\theta}{\pi_{\text{ref}}}A,\; \text{clip}\left(\frac{\pi_\theta}{\pi_{\text{ref}}}, 1-\epsilon, 1+\epsilon\right)A\right)\right] + \lambda \cdot \mathbb{E}\left[(\log\pi_\theta - \log\pi_{\text{ref}})^2\right]
 $$
 
-Parameters: `group_size=4`, `clip_eps=0.2`, `kl_coef=0.01`, `sync_interval=200`.
+Parameters: `group_size=4`, `clip_eps=0.2`, `kl_coef=0.01`, `sync_interval=200`, `reduction="mean"`.
 
 Keys: `prompts`, `responses`, `masks`, `rewards`.
 
@@ -155,7 +128,7 @@ Keys: `prompts`, `responses`, `masks`, `rewards`.
 | Cosine | `CosineScheduler` | Linear warmup → cosine decay to `min_rate` |
 | SGDR | `SGDRScheduler` | Cosine annealing with warm restarts (`t_mult=2`) |
 
-Created by `SchedulerFactory.create(optimizer, schedule_type, **kwargs)`.
+Created by `SchedulerFactory.create(optimizer, schedule_type, **kwargs)`. Valid types: `"cosine"`, `"sgdr"`. Omit to use no scheduler.
 
 ## Gradient Checkpointing
 
@@ -172,8 +145,8 @@ Callback wraps each `DecoderBlock.forward` with `torch.utils.checkpoint.checkpoi
 
 ```
 Checkpoint(state_dict, epoch, iteration, extra, meta, config)
-  ├── save(save_dir)    rank-0 only: meta.json (epoch/iteration/timestamp) + config.json (model config) + state_dict.safetensors + optional {key}.pt (optimizer.pt, scheduler.pt)
-  └── load(save_dir)    broadcasts metadata from rank-0
+  ├── save(save_dir)    rank-0 only: meta.json (epoch/iteration/timestamp) + config.json (model config) + model.safetensors + optional {key}.pt (optimizer.pt, scheduler.pt)
+  └── load(save_dir, broadcast=False)    loads from local disk; set broadcast=True to broadcast metadata from rank-0
 ```
 
 Optimizer/scheduler state persisted by default via `Checkpoint.extra`.  
@@ -194,7 +167,7 @@ context = (
 - Creates executor via `ExecutorFactory.create(cfg.parallel_mode, grad_accum_steps=cfg.grad_accum_steps, **cfg.executor_kwargs)`
 - Calls `executor.prepare(model, optimizer, dataloader, scheduler)` for model distribution (e.g. DDP) + gradient accumulation wrappers
 - Creates `ResumableDistributedSampler` for shuffle+resume
-- Builds strategy via `StrategyFactory.create(train_type, ...)`
+- Builds strategy via `StrategyFactory.create(train_type, model, device, **kwargs)`
 
 ## Training CLI
 
@@ -203,6 +176,7 @@ export CUDA_VISIBLE_DEVICES=0,1,2,3
 
 nohup python scripts/tools/train.py \
     --nprocs=4 \
+    --parallel_mode=ddp \
     --train_type=seq \
     --data_root_path=/path/to/dataset \
     --param_path=/path/to/model \
@@ -224,4 +198,4 @@ nohup python scripts/tools/train.py \
 
 Full parameter reference at [params.md](params.md).
 
-> Document Update Time: 2026-05-28
+> Document Update Time: 2026-05-30

astrai/config/__init__.py

@@ -4,13 +4,22 @@ from astrai.config.model_config import (
     ConfigFactory,
     EncoderConfig,
 )
+from astrai.config.preprocess_config import (
+    InputConfig,
+    OutputConfig,
+    PipelineConfig,
+    ProcessingConfig,
+)
 from astrai.config.train_config import TrainConfig
 
 __all__ = [
-    # Model configuration
     "BaseModelConfig",
     "AutoRegressiveLMConfig",
     "EncoderConfig",
     "ConfigFactory",
     "TrainConfig",
+    "InputConfig",
+    "OutputConfig",
+    "PipelineConfig",
+    "ProcessingConfig",
 ]

astrai/config/base.py

@@ -1,6 +1,7 @@
 import json
 from dataclasses import MISSING, dataclass, fields
-from typing import Any, Dict, Optional, Self, get_type_hints
+from pathlib import Path
+from typing import Any, Dict, Optional, Self, Union, get_type_hints
 
 
 @dataclass
@@ -83,4 +84,15 @@ class BaseConfig:
             return value
         if isinstance(value, target_type):
             return value
+        if isinstance(value, dict) and issubclass(target_type, BaseConfig):
+            return target_type.from_dict(value)
         raise TypeError
+
+    @classmethod
+    def from_json(cls, path: Union[str, Path]) -> Self:
+        with open(path, "r", encoding="utf-8") as f:
+            return cls.from_dict(json.load(f))
+
+    def to_json(self, path: Union[str, Path]):
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)

astrai/config/preprocess_config.py

@@ -0,0 +1,43 @@
+"""Pipeline configuration for JSONL preprocessing."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, Optional
+
+from astrai.config.base import BaseConfig
+
+
+@dataclass
+class InputConfig(BaseConfig):
+    type: str = "chat"
+    messages_key: str = "messages"
+    prompt_key: str = "prompt"
+    response_key: str = "response"
+    text_key: str = "text"
+
+
+@dataclass
+class ProcessingConfig(BaseConfig):
+    max_seq_len: int = 2048
+    min_chars: int = 50
+    max_chars: int = 2_000_000
+    deduplicate: bool = True
+    max_items: Optional[int] = None
+
+
+@dataclass
+class OutputConfig(BaseConfig):
+    domain_key: Optional[str] = None
+    storage_format: str = "bin"
+    max_tokens_per_shard: int = 100_000_000
+
+
+@dataclass
+class PipelineConfig(BaseConfig):
+    version: int = 1
+    input: InputConfig = field(default_factory=InputConfig)
+    mask: Dict[str, str] = field(default_factory=dict)
+    mask_default: str = "mask"
+    preprocessing: ProcessingConfig = field(default_factory=ProcessingConfig)
+    output: OutputConfig = field(default_factory=OutputConfig)

astrai/inference/api/protocol.py

@@ -138,13 +138,13 @@ class ProtocolHandler:
             yielded = ""
             matched = None
             async for token in agen:
-                ctx.completion_tokens += 1
                 body += token
 
                 matched = checker.check(body)
                 if matched:
                     break
 
+                ctx.completion_tokens += 1
                 yield self.builder.format_chunk(token)
                 yielded += token
 
@@ -168,7 +168,6 @@ class ProtocolHandler:
         matched = None
 
         async for token in agen:
-            ctx.completion_tokens += 1
             chunks.append(token)
             body += token
 
@@ -176,6 +175,8 @@ class ProtocolHandler:
             if matched:
                 break
 
+            ctx.completion_tokens += 1
+
         content = "".join(chunks)
         stop = StopInfo(matched=matched, body=body)
         return self.builder.format_response(ctx, content, stop)

astrai/inference/core/scheduler.py

@@ -71,6 +71,7 @@ class InferenceScheduler:
         )
 
         self._running = False
+        self._fatal_error: Optional[Exception] = None
 
     def add_task(self, prompt: str, **kwargs) -> str:
         return self._task_mgr.add_task(prompt, **kwargs)
@@ -175,6 +176,8 @@ class InferenceScheduler:
                                     t.stream_callback(STOP)
 
         except Exception as e:
+            self._fatal_error = e
+            self._running = False
             logger.error(f"Scheduler loop crashed: {e}", exc_info=True)
             for task in self._task_mgr.get_active_tasks():
                 if task.stream_callback:
@@ -184,7 +187,6 @@ class InferenceScheduler:
                 if task.stream_callback:
                     task.stream_callback(STOP)
             self._task_mgr.clear_queues()
-            raise
 
     def start(self):
         if not self._running:
@@ -199,7 +201,12 @@ class InferenceScheduler:
         if hasattr(self, "_loop_thread"):
             self._loop_thread.join(timeout=2.0)
         for task in self._task_mgr.get_active_tasks():
+            if task.stream_callback:
+                task.stream_callback(STOP)
             self._page_cache.task_free(task.task_id)
+        for task in self._task_mgr.get_waiting_tasks():
+            if task.stream_callback:
+                task.stream_callback(STOP)
         self._task_mgr.clear_queues()
         if torch.cuda.is_available():
             torch.cuda.empty_cache()

astrai/inference/core/task.py

@@ -186,7 +186,10 @@ class TaskManager:
         return bool(self.active_tasks or self.waiting_queue)
 
     def wait_for_tasks(self, timeout: float = 1.0):
-        self._task_event.clear()
+        with self._lock:
+            if self.waiting_queue or self.active_tasks:
+                return
+            self._task_event.clear()
         self._task_event.wait(timeout=timeout)
 
     def get_active_tasks(self) -> List[Task]:

astrai/inference/engine.py

@@ -79,8 +79,8 @@ class GenerationRequest:
             raise ValueError("top_k must be a non-negative integer")
         if not (0.0 <= top_p <= 1.0):
             raise ValueError("top_p must be a float between 0.0 and 1.0")
-        if not (isinstance(temperature, (int, float)) and temperature >= 0):
-            raise ValueError("temperature must be a non-negative number")
+        if not (isinstance(temperature, (int, float)) and temperature > 0):
+            raise ValueError("temperature must be a positive number")
 
         self.messages = messages
         self.top_k = top_k

astrai/inference/sample.py

@@ -44,10 +44,12 @@ class TemperatureStrategy(BaseSamplingStrategy):
     def apply(self, logits, filter_value=-float("inf")):
         t = self.temperature
         if isinstance(t, Tensor):
+            t = t.to(logits.device, non_blocking=True).view(-1, 1)
+            t = torch.clamp(t, min=1e-8)
             if (t != 1.0).any():
-                logits = logits / t.to(logits.device, non_blocking=True).view(-1, 1)
+                logits = logits / t
         elif t != 1.0:
-            logits = logits / t
+            logits = logits / max(t, 1e-8)
         return logits
 
 

astrai/parallel/executor.py

@@ -7,6 +7,7 @@ from typing import Optional, Tuple
 
 import torch
 import torch.nn as nn
+from torch.distributed.fsdp import FullStateDictConfig, StateDictType
 from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
 from torch.nn.parallel import DistributedDataParallel as DDP
 from torch.optim import Optimizer
@@ -115,8 +116,8 @@ class BaseExecutor:
     def backward(self, loss: torch.Tensor):
         loss.backward()
 
-    def unwrap_model(self, model: nn.Module) -> nn.Module:
-        return model
+    def unwrap_model(self, model: nn.Module):
+        return model.state_dict()
 
     @property
     def use_distributed(self) -> bool:
@@ -195,10 +196,10 @@ class DDPExecutor(BaseExecutor):
             return model.no_sync()
         return contextlib.nullcontext()
 
-    def unwrap_model(self, model: nn.Module) -> nn.Module:
+    def unwrap_model(self, model: nn.Module):
         if isinstance(model, DDP):
-            return model.module
-        return model
+            return model.module.state_dict()
+        return model.state_dict()
 
 
 @ExecutorFactory.register("fsdp")
@@ -217,7 +218,6 @@ class FSDPExecutor(BaseExecutor):
         sync_module_states: bool = False,
         forward_prefetch: bool = False,
         limit_all_gathers: bool = True,
-        use_orig_params: bool = False,
         ignored_states=None,
         device_mesh=None,
     ):
@@ -236,7 +236,7 @@ class FSDPExecutor(BaseExecutor):
                 sync_module_states=sync_module_states,
                 forward_prefetch=forward_prefetch,
                 limit_all_gathers=limit_all_gathers,
-                use_orig_params=use_orig_params,
+                use_orig_params=True,
                 ignored_states=ignored_states,
                 device_mesh=device_mesh,
             ).items()
@@ -259,9 +259,13 @@ class FSDPExecutor(BaseExecutor):
             return model.no_sync()
         return contextlib.nullcontext()
 
-    def unwrap_model(self, model: nn.Module) -> nn.Module:
-        if self._original_model is not None:
-            return self._original_model
-        if isinstance(model, FSDP):
-            return model._fsdp_wrapped_module
-        return model
+    def unwrap_model(self, model: nn.Module):
+        if isinstance(model, FSDP) and self.use_distributed:
+            with FSDP.state_dict_type(
+                model,
+                StateDictType.FULL_STATE_DICT,
+                FullStateDictConfig(offload_to_cpu=True, rank0_only=False),
+            ):
+                return model.state_dict()
+
+        return model.state_dict()

astrai/preprocessing/__init__.py

@@ -0,0 +1,19 @@
+from astrai.preprocessing.builder import (
+    BaseMaskBuilder,
+    ChatMaskBuilder,
+    InstructionMaskBuilder,
+    MaskBuilderFactory,
+    TextMaskBuilder,
+)
+from astrai.preprocessing.pipeline import Pipeline, dedup_signature, filter_by_length
+
+__all__ = [
+    "BaseMaskBuilder",
+    "ChatMaskBuilder",
+    "InstructionMaskBuilder",
+    "MaskBuilderFactory",
+    "TextMaskBuilder",
+    "Pipeline",
+    "dedup_signature",
+    "filter_by_length",
+]

astrai/preprocessing/builder.py

@@ -0,0 +1,161 @@
+"""Mask building strategies for preprocessing pipeline.
+
+Each builder knows how to tokenize one input format and construct
+the loss_mask according to declarative mask rules from the config.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import List, Optional
+
+from astrai.factory import BaseFactory
+
+
+class BaseMaskBuilder(ABC):
+    """Convert a JSONL item into token ids and optional loss_mask."""
+
+    @abstractmethod
+    def build(self, item: dict, config, tokenizer) -> Optional[dict]:
+        """Build ``{ids, loss_mask?, domain}`` from a JSONL record.
+
+        Returns ``None`` to skip the item entirely.
+        """
+        ...
+
+
+class MaskBuilderFactory(BaseFactory["BaseMaskBuilder"]):
+    @classmethod
+    def _validate_component(cls, component_cls: type):
+        if not issubclass(component_cls, BaseMaskBuilder):
+            raise TypeError(
+                f"{component_cls.__name__} must inherit from BaseMaskBuilder"
+            )
+
+
+def _extract_domain(item: dict, domain_key: Optional[str]) -> str:
+    if not domain_key:
+        return "__default__"
+    val = item.get(domain_key, "__default__")
+    return val if isinstance(val, str) else "__default__"
+
+
+@MaskBuilderFactory.register("chat")
+class ChatMaskBuilder(BaseMaskBuilder):
+    """Mask by role via message-level tokenisation with role-span tracking.
+
+    For each message, renders the chat template for that single message,
+    encodes individually, and records its token span + role action.
+    The concatenated sequence receives a loss_mask built from span rules.
+    """
+
+    def build(self, item: dict, config, tokenizer) -> Optional[dict]:
+        messages = item.get(config.input.messages_key)
+        if not isinstance(messages, list) or not messages:
+            return None
+
+        all_ids: List[int] = []
+        spans: List[tuple] = []
+
+        if tokenizer.bos_token_id is not None:
+            all_ids.append(tokenizer.bos_token_id)
+
+        for msg in messages:
+            role = msg.get("role", "")
+            action = config.mask.get(role, config.mask_default)
+
+            rendered = tokenizer.apply_chat_template(
+                [msg], tokenize=False, add_generation_prompt=False
+            )
+            ids = tokenizer.encode(rendered, add_special_tokens=False)
+
+            start = len(all_ids)
+            all_ids.extend(ids)
+            spans.append((start, len(all_ids), action))
+
+        if len(all_ids) <= 1:
+            return None
+
+        max_len = config.preprocessing.max_seq_len
+        all_ids = all_ids[:max_len]
+
+        loss_mask = [0] * len(all_ids)
+        for start, end, action in spans:
+            if start >= len(all_ids):
+                break
+            e = min(end, len(all_ids))
+            if action == "train":
+                loss_mask[start:e] = [1] * (e - start)
+
+        return {
+            "ids": all_ids,
+            "loss_mask": loss_mask,
+            "domain": _extract_domain(item, config.output.domain_key),
+        }
+
+
+@MaskBuilderFactory.register("instruction")
+class InstructionMaskBuilder(BaseMaskBuilder):
+    """Mask by prompt / response field boundary.
+
+    Encodes prompt and response independently, then fills mask
+    according to ``prompt`` / ``response`` entries in the mask config.
+    """
+
+    def build(self, item: dict, config, tokenizer) -> Optional[dict]:
+        prompt = str(item.get(config.input.prompt_key, ""))
+        response = str(item.get(config.input.response_key, ""))
+
+        if not prompt.strip() and not response.strip():
+            return None
+
+        prompt_ids = tokenizer.encode(prompt, add_special_tokens=True)
+        response_ids = tokenizer.encode(response, add_special_tokens=False)
+
+        max_len = config.preprocessing.max_seq_len
+        full_ids = (prompt_ids + response_ids)[:max_len]
+
+        prompt_action = config.mask.get("prompt", config.mask_default)
+        response_action = config.mask.get("response", config.mask_default)
+
+        p_len = min(len(prompt_ids), len(full_ids))
+        r_len = len(full_ids) - p_len
+
+        loss_mask = []
+        if prompt_action == "train":
+            loss_mask += [1] * p_len
+        else:
+            loss_mask += [0] * p_len
+
+        if response_action == "train":
+            loss_mask += [1] * r_len
+        else:
+            loss_mask += [0] * r_len
+
+        return {
+            "ids": full_ids,
+            "loss_mask": loss_mask,
+            "domain": _extract_domain(item, config.output.domain_key),
+        }
+
+
+@MaskBuilderFactory.register("text")
+class TextMaskBuilder(BaseMaskBuilder):
+    """Plain tokenisation — no mask, used for pre-training data."""
+
+    def build(self, item: dict, config, tokenizer) -> Optional[dict]:
+        text = item.get(config.input.text_key, "")
+        if not isinstance(text, str) or not text.strip():
+            return None
+
+        pp = config.preprocessing
+        if not (pp.min_chars <= len(text) <= pp.max_chars):
+            return None
+
+        ids = tokenizer.encode(text, add_special_tokens=True)
+        ids = ids[: pp.max_seq_len]
+
+        return {
+            "ids": ids,
+            "domain": _extract_domain(item, config.output.domain_key),
+        }

astrai/preprocessing/pipeline.py

@@ -0,0 +1,134 @@
+"""Config-driven JSONL preprocessing pipeline.
+
+Composes a :class:`BaseMaskBuilder` (selected by ``input.type``) with
+deduplication, sharding, and flush to ``.h5`` / ``.bin`` storage.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from collections import defaultdict
+from typing import List, Optional
+
+import torch
+import tqdm
+
+from astrai.config.preprocess_config import PipelineConfig
+from astrai.dataset.storage import save_bin, save_h5
+from astrai.preprocessing.builder import MaskBuilderFactory
+from astrai.tokenize import AutoTokenizer
+
+
+def filter_by_length(text: str, min_len: int = 50, max_len: int = 2_000_000) -> bool:
+    return min_len <= len(text) <= max_len
+
+
+def dedup_signature(item: dict) -> str:
+    raw = json.dumps(item, sort_keys=True, ensure_ascii=False)
+    return hashlib.md5(raw[:200].encode()).hexdigest()
+
+
+class Pipeline:
+    """Tokenization pipeline driven by a declarative :class:`PipelineConfig`.
+
+    Usage::
+
+        config = PipelineConfig.from_json("sft_pipeline.json")
+        Pipeline(config, ["data.jsonl"], output_dir="out", tokenizer_path="params").run()
+    """
+
+    def __init__(
+        self,
+        config: PipelineConfig,
+        input_paths: List[str],
+        output_dir: str,
+        tokenizer_path: str,
+    ):
+        os.makedirs(output_dir, exist_ok=True)
+        self.config = config
+        self.paths = input_paths
+        self.output_dir = output_dir
+        self.tokenizer_path = tokenizer_path
+
+        self.mask_builder = MaskBuilderFactory.create(config.input.type)
+
+    def transform(self, item: dict) -> Optional[dict]:
+        return self.mask_builder.build(item, self.config, self._tokenizer)
+
+    def run(self):
+        self._tokenizer = AutoTokenizer.from_pretrained(self.tokenizer_path)
+
+        seen: set = set()
+        domains: dict = defaultdict(lambda: defaultdict(list))
+        total_tokens = 0
+        shard_idx: dict[str, int] = defaultdict(int)
+        count = 0
+
+        pp = self.config.preprocessing
+
+        for item in tqdm.tqdm(
+            self._iter_items(), desc="Tokenizing", unit="docs", mininterval=0.5
+        ):
+            if pp.max_items and count >= pp.max_items:
+                break
+
+            if pp.deduplicate:
+                sig = dedup_signature(item)
+                if sig in seen:
+                    continue
+                seen.add(sig)
+
+            result = self.transform(item)
+            if result is None:
+                continue
+
+            ids = result["ids"]
+            if not ids:
+                continue
+
+            domain = result.get("domain", "__default__")
+            domains[domain]["sequence"].append(ids)
+            if "loss_mask" in result:
+                domains[domain]["loss_mask"].append(result["loss_mask"])
+
+            count += 1
+            total_tokens += len(ids)
+
+            if total_tokens >= self.config.output.max_tokens_per_shard:
+                self._flush(domains, shard_idx)
+                domains.clear()
+                total_tokens = 0
+
+        if total_tokens > 0:
+            self._flush(domains, shard_idx)
+
+        print(f"Done. {count} documents tokenized.")
+
+    def _iter_items(self):
+        for path in self.paths:
+            with open(path, "r", encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    yield json.loads(line)
+
+    def _flush(self, domains, shard_idx):
+        for domain, keys in domains.items():
+            idx = shard_idx[domain]
+            tensors = {}
+            for key, ids_list in keys.items():
+                tensors[key] = [torch.tensor(sum(ids_list, []), dtype=torch.long)]
+            chunk_dir = os.path.join(self.output_dir, domain)
+            fmt = self.config.output.storage_format
+            if fmt == "bin":
+                save_bin(chunk_dir, tensors)
+            else:
+                save_h5(chunk_dir, f"data_{idx:04d}", tensors)
+            shard_idx[domain] = idx + 1
+            tqdm.tqdm.write(
+                f"  saved {domain}/shard_{idx:04d}  "
+                f"({tensors['sequence'][0].numel():,} tokens)"
+            )

astrai/trainer/strategy.py

@@ -1,6 +1,5 @@
 """Training strategy implementations with factory pattern."""
 
-import copy
 from abc import ABC, abstractmethod
 from typing import Any, Callable, Dict, Union
 
@@ -8,28 +7,14 @@ import torch
 import torch.nn as nn
 import torch.nn.functional as F
 from torch import Tensor
-from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
-from torch.nn.parallel import DistributedDataParallel as DDP
 
 from astrai.factory import BaseFactory
 
 
-def unwrap_model(model: nn.Module) -> nn.Module:
-    if isinstance(model, DDP):
-        return model.module
-    if isinstance(model, FSDP):
-        return model._fsdp_wrapped_module
-    return model
-
-
-def create_ref_model(model: nn.Module) -> nn.Module:
-    """Create a reference model for DPO/GRPO training.
-
-    Handles DDP-wrapped models safely by unwrapping first,
-    then creating a deep copy with frozen gradients.
-    """
-    original_model = unwrap_model(model)
-    ref_model = copy.deepcopy(original_model)
+def create_ref_model(model_fn, state_dict: dict) -> nn.Module:
+    """Create a frozen reference model from model_fn + full state dict."""
+    ref_model = model_fn()
+    ref_model.load_state_dict(state_dict)
     ref_model.requires_grad_(False)
     ref_model.eval()
     return ref_model
@@ -91,6 +76,8 @@ class BaseStrategy(ABC):
     ):
         self.model = model
         self.device = device
+        self.executor = kwargs.pop("executor", None)
+        self.model_fn = kwargs.pop("model_fn", None)
         self.extra_kwargs = kwargs
 
     @abstractmethod
@@ -230,7 +217,9 @@ class DPOStrategy(BaseStrategy):
         **kwargs,
     ):
         super().__init__(model, device, **kwargs)
-        self.ref_model = create_ref_model(model)
+        self.ref_model = create_ref_model(
+            self.model_fn, self.executor.unwrap_model(model)
+        ).to(device=self.device)
         self.beta = beta
         self.reduction = reduction
 
@@ -284,7 +273,9 @@ class GRPOStrategy(BaseStrategy):
         **kwargs,
     ):
         super().__init__(model, device, **kwargs)
-        self.ref_model = create_ref_model(model)
+        self.ref_model = create_ref_model(
+            self.model_fn, self.executor.unwrap_model(model)
+        ).to(device=self.device)
         self.clip_eps = clip_eps
         self.kl_coef = kl_coef
         self.group_size = group_size
@@ -294,8 +285,7 @@ class GRPOStrategy(BaseStrategy):
 
     def sync_ref_model(self):
         """Copy current model weights to ref model."""
-        ref_state = self.model.state_dict()
-        self.ref_model.load_state_dict(ref_state)
+        self.ref_model.load_state_dict(self.executor.unwrap_model(self.model))
 
     def compute_loss(self, batch: Dict[str, Tensor]) -> Tensor:
         self._step += 1

astrai/trainer/train_callback.py

@@ -146,8 +146,7 @@ class CheckpointCallback(TrainCallback):
         self.last_ckpt_iter = 0
 
     def _save_checkpoint(self, context: TrainContext):
-        unwrapped = context.executor.unwrap_model(context.model)
-        state_dict = unwrapped.state_dict()
+        state_dict = context.executor.unwrap_model(context.model)
         self.last_ckpt_iter = context.iteration
 
         if get_rank() == 0:

astrai/trainer/train_context.py

@@ -162,6 +162,8 @@ class TrainContextBuilder:
             model=context.model,
             train_type=cfg.strategy,
             device=device,
+            executor=executor,
+            model_fn=cfg.model_fn,
             **cfg.extra_kwargs,
         )
 

scripts/tools/evaluate_mmlu.py

@@ -5,9 +5,9 @@ import csv
 import json
 import os
 import shutil
-import urllib.request
-import zipfile
+import tarfile
 
+import requests
 import torch
 import torch.nn.functional as F
 import tqdm
@@ -15,7 +15,7 @@ import tqdm
 from astrai.model import AutoModel
 from astrai.tokenize import AutoTokenizer
 
-MMLU_URL = "https://github.com/hendrycks/test/archive/refs/heads/master.zip"
+MMLU_URL = "https://people.eecs.berkeley.edu/~hendrycks/data.tar"
 MMLU_SUBJECTS = [
     "abstract_algebra",
     "anatomy",
@@ -78,23 +78,37 @@ MMLU_SUBJECTS = [
 
 
 def _download_and_extract(url: str, data_dir: str):
-    zip_path = os.path.join(data_dir, "mmlu.zip")
+    tar_path = os.path.join(data_dir, "data.tar")
     os.makedirs(data_dir, exist_ok=True)
     print(f"Downloading MMLU data from {url}...")
-    urllib.request.urlretrieve(url, zip_path)
+    resp = requests.get(url, stream=True, timeout=300)
+    resp.raise_for_status()
+    total = int(resp.headers.get("content-length", 0))
+    with tqdm.tqdm(total=total, unit="B", unit_scale=True, desc="  Download") as bar:
+        with open(tar_path, "wb") as f:
+            for chunk in resp.iter_content(chunk_size=8192):
+                f.write(chunk)
+                bar.update(len(chunk))
     print("Extracting...")
-    with zipfile.ZipFile(zip_path, "r") as zf:
-        zf.extractall(data_dir)
-    os.remove(zip_path)
+    with tarfile.open(tar_path, "r") as tf:
+        tf.extractall(data_dir)
+    os.remove(tar_path)
 
 
 def download_mmlu(data_dir: str):
     _download_and_extract(MMLU_URL, data_dir)
-    src = os.path.join(data_dir, "test-master", "data")
+    src = os.path.join(data_dir, "data")
     if os.path.exists(src):
         for item in os.listdir(src):
-            os.rename(os.path.join(src, item), os.path.join(data_dir, item))
-        shutil.rmtree(os.path.join(data_dir, "test-master"))
+            src_item = os.path.join(src, item)
+            dst_item = os.path.join(data_dir, item)
+            if os.path.exists(dst_item):
+                if os.path.isdir(dst_item):
+                    shutil.rmtree(dst_item)
+                else:
+                    os.remove(dst_item)
+            os.rename(src_item, dst_item)
+        os.rmdir(src)
     print(f"MMLU data saved to {data_dir}")
 
 
@@ -233,6 +247,7 @@ def main():
     device = args.device
     dtype = getattr(torch, args.dtype)
     model.to(device=device, dtype=dtype)
+    model.eval()
 
     subjects = args.subjects or MMLU_SUBJECTS
     results = {}

scripts/tools/preprocess.py

@@ -0,0 +1,38 @@
+"""CLI: JSONL → tokenized .h5/.bin via config-driven Pipeline."""
+
+import argparse
+
+from astrai.config.preprocess_config import PipelineConfig
+from astrai.preprocessing.pipeline import Pipeline
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Raw JSONL → tokenized .h5/.bin via config-driven Pipeline"
+    )
+    parser.add_argument(
+        "inputs", nargs="+", metavar="JSONL", help="One or more JSONL files"
+    )
+    parser.add_argument("--output_dir", "-o", required=True, help="Output directory")
+    parser.add_argument(
+        "--config", "-c", required=True, help="Path to pipeline config JSON"
+    )
+    parser.add_argument(
+        "--tokenizer_path",
+        default="params",
+        help="Path to tokenizer directory (default: params)",
+    )
+    args = parser.parse_args()
+
+    config = PipelineConfig.from_json(args.config)
+
+    Pipeline(
+        config=config,
+        input_paths=args.inputs,
+        output_dir=args.output_dir,
+        tokenizer_path=args.tokenizer_path,
+    ).run()
+
+
+if __name__ == "__main__":
+    main()

tests/data/test_dataset.py

@@ -1,4 +1,3 @@
-import json
 import os
 
 import numpy as np
@@ -8,7 +7,6 @@ import torch
 from astrai.dataset.dataset import DatasetFactory, SEQDataset
 from astrai.dataset.storage import (
     H5Store,
-    MmapStore,
     StoreFactory,
     detect_format,
     load_bin,

tests/data/test_preprocess.py

@@ -0,0 +1,603 @@
+import json
+import os
+import tempfile
+
+import pytest
+from tokenizers import Tokenizer, models, pre_tokenizers, trainers
+
+from astrai.config.preprocess_config import (
+    InputConfig,
+    OutputConfig,
+    PipelineConfig,
+    ProcessingConfig,
+)
+from astrai.preprocessing.builder import (
+    ChatMaskBuilder,
+    InstructionMaskBuilder,
+    MaskBuilderFactory,
+    TextMaskBuilder,
+)
+from astrai.preprocessing.pipeline import Pipeline, dedup_signature, filter_by_length
+from astrai.tokenize import AutoTokenizer
+
+_SPECIAL_TOKENS = [
+    "<unk>",
+    "<pad>",
+    "<|begin_of_sentence|>",
+    "<|end_of_sentence|>",
+    "<|im_start|>",
+    "<|im_end|>",
+]
+
+_CHAT_TEMPLATE = (
+    "{% for message in messages %}"
+    "{% if message['role'] == 'system' %}"
+    "<|im_start|>system\n{{ message['content'] }}<|im_end|>\n"
+    "{% elif message['role'] == 'user' %}"
+    "<|im_start|>user\n{{ message['content'] }}<|im_end|>\n"
+    "{% elif message['role'] == 'assistant' %}"
+    "<|im_start|>assistant\n{{ message['content'] }}<|im_end|>\n"
+    "{% endif %}"
+    "{% endfor %}"
+    "{% if add_generation_prompt %}<|im_start|>assistant\n{% endif %}"
+)
+
+
+def _build_chat_tokenizer() -> AutoTokenizer:
+    tok = Tokenizer(models.BPE())
+    tok.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False)
+    tr = trainers.BpeTrainer(
+        vocab_size=512,
+        min_frequency=1,
+        special_tokens=_SPECIAL_TOKENS,
+    )
+    train_data = [
+        "hello world",
+        "Hi there!",
+        "You are helpful.",
+        "What is 2+2?",
+        "Tell me a story about dragons and knights.",
+        "Sure, here is a tale.",
+        "Translate to French: Hello",
+        "Bonjour",
+        "Artificial Intelligence is a field of computer science.",
+        "system",
+        "user",
+        "assistant",
+        "<|im_start|>",
+        "<|im_end|>",
+        *[chr(i) for i in range(32, 127)],
+    ]
+    tok.train_from_iterator(train_data, tr)
+
+    auto_tok = AutoTokenizer()
+    auto_tok._tokenizer = tok
+    auto_tok._special_token_map = {
+        "bos_token": "<|begin_of_sentence|>",
+        "eos_token": "<|end_of_sentence|>",
+        "pad_token": "<pad>",
+        "unk_token": "<unk>",
+    }
+    auto_tok.set_chat_template(_CHAT_TEMPLATE)
+    return auto_tok
+
+
+@pytest.fixture(scope="session")
+def chat_tokenizer():
+    return _build_chat_tokenizer()
+
+
+@pytest.fixture
+def temp_dir():
+    d = tempfile.mkdtemp()
+    yield d
+    import shutil
+
+    shutil.rmtree(d, ignore_errors=True)
+
+
+def make_chat_config():
+    return PipelineConfig(
+        input=InputConfig(type="chat", messages_key="messages"),
+        mask={"system": "mask", "user": "mask", "assistant": "train"},
+        mask_default="mask",
+        preprocessing=ProcessingConfig(max_seq_len=2048),
+    )
+
+
+def make_instruction_config():
+    return PipelineConfig(
+        input=InputConfig(
+            type="instruction", prompt_key="prompt", response_key="response"
+        ),
+        mask={"prompt": "mask", "response": "train"},
+        mask_default="mask",
+        preprocessing=ProcessingConfig(max_seq_len=2048),
+    )
+
+
+def make_text_config():
+    return PipelineConfig(
+        input=InputConfig(type="text", text_key="text"),
+        preprocessing=ProcessingConfig(
+            max_seq_len=2048, min_chars=1, max_chars=2_000_000
+        ),
+    )
+
+
+class TestPipelineConfig:
+    def test_default_values(self):
+        config = PipelineConfig()
+        assert config.version == 1
+        assert config.input.type == "chat"
+        assert config.mask == {}
+        assert config.mask_default == "mask"
+        assert config.preprocessing.max_seq_len == 2048
+        assert config.output.storage_format == "bin"
+
+    def test_from_dict_flat(self):
+        data = {
+            "version": 1,
+            "input": {"type": "chat", "messages_key": "msgs"},
+            "mask": {"system": "mask", "assistant": "train"},
+            "mask_default": "mask",
+            "preprocessing": {"max_seq_len": 1024},
+            "output": {"storage_format": "h5"},
+        }
+        config = PipelineConfig.from_dict(data)
+        assert config.input.type == "chat"
+        assert config.input.messages_key == "msgs"
+        assert config.mask == {"system": "mask", "assistant": "train"}
+        assert config.preprocessing.max_seq_len == 1024
+        assert config.output.storage_format == "h5"
+
+    def test_to_dict_roundtrip(self):
+        config = PipelineConfig(
+            input=InputConfig(type="instruction", prompt_key="q", response_key="a"),
+            mask={"prompt": "mask", "response": "train"},
+            mask_default="mask",
+        )
+        d = config.to_dict()
+        config2 = PipelineConfig.from_dict(d)
+        assert config2.input.type == "instruction"
+        assert config2.input.prompt_key == "q"
+        assert config2.mask == {"prompt": "mask", "response": "train"}
+
+    def test_to_json_from_json(self, temp_dir):
+        config = PipelineConfig(
+            input=InputConfig(type="text", text_key="body"),
+            mask={"text": "train"},
+            mask_default="mask",
+        )
+        path = os.path.join(temp_dir, "config.json")
+        config.to_json(path)
+        loaded = PipelineConfig.from_json(path)
+        assert loaded.input.type == "text"
+        assert loaded.input.text_key == "body"
+        assert loaded.mask == {"text": "train"}
+
+
+class TestChatMaskBuilder:
+    def test_simple_chat_mask(self, chat_tokenizer):
+        config = make_chat_config()
+        builder = ChatMaskBuilder()
+        item = {
+            "messages": [
+                {"role": "system", "content": "You are helpful."},
+                {"role": "user", "content": "Hello."},
+                {"role": "assistant", "content": "Hi there!"},
+            ]
+        }
+        result = builder.build(item, config, chat_tokenizer)
+        assert result is not None
+        assert "ids" in result
+        assert "loss_mask" in result
+        assert len(result["ids"]) == len(result["loss_mask"])
+
+        ids = chat_tokenizer.decode(result["ids"], skip_special_tokens=False)
+
+        assert "system" in ids.lower() or "<|im_start|>system" in ids
+        assert "assistant" in ids.lower() or "<|im_start|>assistant" in ids
+
+        total = len(result["ids"])
+        trained = sum(result["loss_mask"])
+        assert trained > 0, "At least assistant tokens should be trained"
+        assert trained < total, "System and user tokens should be masked"
+
+    def test_mask_only_assistant_trained(self, chat_tokenizer):
+        config = make_chat_config()
+        builder = ChatMaskBuilder()
+        item = {
+            "messages": [
+                {"role": "user", "content": "What is 2+2?"},
+                {"role": "assistant", "content": "4"},
+            ]
+        }
+        result = builder.build(item, config, chat_tokenizer)
+        mask = result["loss_mask"]
+        ids = result["ids"]
+
+        assert len(ids) == len(mask)
+
+        trained_positions = [i for i, m in enumerate(mask) if m == 1]
+        assert len(trained_positions) > 0, "At least some tokens should be trained"
+
+        masked_positions = [i for i, m in enumerate(mask) if m == 0]
+        assert len(masked_positions) > 0, "User tokens should be masked"
+
+    def test_chat_all_masked(self, chat_tokenizer):
+        config = PipelineConfig(
+            input=InputConfig(type="chat", messages_key="messages"),
+            mask={"system": "mask", "user": "mask", "assistant": "mask"},
+            mask_default="mask",
+            preprocessing=ProcessingConfig(max_seq_len=2048),
+        )
+        builder = ChatMaskBuilder()
+        item = {
+            "messages": [
+                {"role": "system", "content": "You are helpful."},
+                {"role": "assistant", "content": "Hi there!"},
+            ]
+        }
+        result = builder.build(item, config, chat_tokenizer)
+        assert sum(result["loss_mask"]) == 0
+
+    def test_chat_all_trained(self, chat_tokenizer):
+        config = PipelineConfig(
+            input=InputConfig(type="chat", messages_key="messages"),
+            mask={},
+            mask_default="train",
+            preprocessing=ProcessingConfig(max_seq_len=2048),
+        )
+        builder = ChatMaskBuilder()
+        item = {
+            "messages": [
+                {"role": "system", "content": "You are helpful."},
+                {"role": "assistant", "content": "Hi there!"},
+            ]
+        }
+        result = builder.build(item, config, chat_tokenizer)
+        assert sum(result["loss_mask"]) == len(result["ids"]) - 1
+
+    def test_empty_messages_returns_none(self, chat_tokenizer):
+        config = make_chat_config()
+        builder = ChatMaskBuilder()
+        assert builder.build({"messages": []}, config, chat_tokenizer) is None
+        assert builder.build({}, config, chat_tokenizer) is None
+
+    def test_domain_extraction(self, chat_tokenizer):
+        config = PipelineConfig(
+            input=InputConfig(type="chat", messages_key="messages"),
+            mask={"assistant": "train"},
+            mask_default="mask",
+            preprocessing=ProcessingConfig(max_seq_len=2048),
+            output=OutputConfig(domain_key="source"),
+        )
+        builder = ChatMaskBuilder()
+        item = {
+            "messages": [
+                {"role": "user", "content": "Hi"},
+                {"role": "assistant", "content": "Hello"},
+            ],
+            "source": "wiki",
+        }
+        result = builder.build(item, config, chat_tokenizer)
+        assert result["domain"] == "wiki"
+
+    def test_truncation_to_max_len(self, chat_tokenizer):
+        config = PipelineConfig(
+            input=InputConfig(type="chat", messages_key="messages"),
+            mask={"assistant": "train"},
+            mask_default="mask",
+            preprocessing=ProcessingConfig(max_seq_len=10),
+        )
+        builder = ChatMaskBuilder()
+        item = {
+            "messages": [
+                {
+                    "role": "user",
+                    "content": "Tell me a very long story about dragons and knights and magic.",
+                },
+                {"role": "assistant", "content": "Sure! Here is a tale..."},
+            ]
+        }
+        result = builder.build(item, config, chat_tokenizer)
+        assert len(result["ids"]) <= 10
+        assert len(result["loss_mask"]) == len(result["ids"])
+
+
+class TestInstructionMaskBuilder:
+    def test_basic_instruction_mask(self, test_tokenizer):
+        config = make_instruction_config()
+        builder = InstructionMaskBuilder()
+        item = {"prompt": "Translate to French: Hello", "response": "Bonjour"}
+        result = builder.build(item, config, test_tokenizer)
+        assert result is not None
+        assert len(result["ids"]) == len(result["loss_mask"])
+
+    def test_prompt_masked_response_trained(self, test_tokenizer):
+        config = make_instruction_config()
+        builder = InstructionMaskBuilder()
+        item = {"prompt": "hello", "response": "world"}
+        result = builder.build(item, config, test_tokenizer)
+        mask = result["loss_mask"]
+        ids = result["ids"]
+
+        prompt_ids = test_tokenizer.encode("hello", add_special_tokens=True)
+        response_ids = test_tokenizer.encode("world", add_special_tokens=False)
+
+        p_len = min(len(prompt_ids), len(ids))
+        assert all(m == 0 for m in mask[:p_len])
+
+        if p_len < len(ids):
+            assert all(m == 1 for m in mask[p_len:])
+
+    def test_train_on_prompt(self, test_tokenizer):
+        config = PipelineConfig(
+            input=InputConfig(
+                type="instruction", prompt_key="prompt", response_key="response"
+            ),
+            mask={"prompt": "train", "response": "mask"},
+            mask_default="mask",
+            preprocessing=ProcessingConfig(max_seq_len=2048),
+        )
+        builder = InstructionMaskBuilder()
+        item = {"prompt": "hello", "response": "world"}
+        result = builder.build(item, config, test_tokenizer)
+        mask = result["loss_mask"]
+        ids = result["ids"]
+
+        prompt_ids = test_tokenizer.encode("hello", add_special_tokens=True)
+        p_len = min(len(prompt_ids), len(ids))
+        assert all(m == 1 for m in mask[:p_len])
+
+
+class TestTextMaskBuilder:
+    def test_basic_text(self, test_tokenizer):
+        config = make_text_config()
+        builder = TextMaskBuilder()
+        item = {"text": "Hello world. This is a test document."}
+        result = builder.build(item, config, test_tokenizer)
+        assert result is not None
+        assert "ids" in result
+        assert len(result["ids"]) > 0
+        assert "loss_mask" not in result
+
+    def test_empty_text_returns_none(self, test_tokenizer):
+        config = make_text_config()
+        builder = TextMaskBuilder()
+        assert builder.build({"text": ""}, config, test_tokenizer) is None
+        assert builder.build({"text": "   "}, config, test_tokenizer) is None
+
+    def test_too_short_text(self, test_tokenizer):
+        config = PipelineConfig(
+            input=InputConfig(type="text", text_key="text"),
+            preprocessing=ProcessingConfig(min_chars=100),
+        )
+        builder = TextMaskBuilder()
+        assert builder.build({"text": "short"}, config, test_tokenizer) is None
+
+    def test_truncation(self, test_tokenizer):
+        config = PipelineConfig(
+            input=InputConfig(type="text", text_key="text"),
+            preprocessing=ProcessingConfig(max_seq_len=3, min_chars=1),
+        )
+        builder = TextMaskBuilder()
+        item = {"text": "This is a very long text that should be truncated"}
+        result = builder.build(item, config, test_tokenizer)
+        assert len(result["ids"]) <= 3
+
+
+class TestPipeline:
+    def test_full_chat_pipeline(self, temp_dir, chat_tokenizer):
+        tokenizer_dir = os.path.join(temp_dir, "tok")
+        os.makedirs(tokenizer_dir, exist_ok=True)
+        chat_tokenizer._tokenizer.save(os.path.join(tokenizer_dir, "tokenizer.json"))
+        with open(os.path.join(tokenizer_dir, "tokenizer_config.json"), "w") as f:
+            json.dump(
+                {
+                    "special_tokens": {
+                        "bos_token": "<|begin_of_sentence|>",
+                        "eos_token": "<|end_of_sentence|>",
+                        "pad_token": "<pad>",
+                        "unk_token": "<unk>",
+                        "im_start": "<|im_start|>",
+                        "im_end": "<|im_end|>",
+                    },
+                    "chat_template": _CHAT_TEMPLATE,
+                },
+                f,
+            )
+
+        jsonl_path = os.path.join(temp_dir, "chat.jsonl")
+        with open(jsonl_path, "w", encoding="utf-8") as f:
+            f.write(
+                json.dumps(
+                    {
+                        "messages": [
+                            {"role": "system", "content": "You are helpful."},
+                            {"role": "user", "content": "Hi."},
+                            {"role": "assistant", "content": "Hello!"},
+                        ]
+                    }
+                )
+                + "\n"
+            )
+            f.write(
+                json.dumps(
+                    {
+                        "messages": [
+                            {"role": "user", "content": "What is 2+2?"},
+                            {"role": "assistant", "content": "4"},
+                        ]
+                    }
+                )
+                + "\n"
+            )
+
+        config = PipelineConfig(
+            input=InputConfig(type="chat", messages_key="messages"),
+            mask={"system": "mask", "user": "mask", "assistant": "train"},
+            mask_default="mask",
+            preprocessing=ProcessingConfig(max_seq_len=2048, deduplicate=True),
+            output=OutputConfig(storage_format="bin", domain_key=None),
+        )
+
+        out_dir = os.path.join(temp_dir, "output")
+        Pipeline(
+            config=config,
+            input_paths=[jsonl_path],
+            output_dir=out_dir,
+            tokenizer_path=tokenizer_dir,
+        ).run()
+
+        meta_path = os.path.join(out_dir, "__default__", "meta.json")
+        assert os.path.exists(meta_path)
+        with open(meta_path, "r") as f:
+            meta = json.load(f)
+        assert "sequence" in meta
+        assert "loss_mask" in meta
+
+    def test_full_text_pipeline(self, temp_dir, test_tokenizer):
+        import tempfile as tmp
+
+        tokenizer_dir = os.path.join(temp_dir, "tok")
+        os.makedirs(tokenizer_dir, exist_ok=True)
+
+        test_tokenizer._tokenizer.save(os.path.join(tokenizer_dir, "tokenizer.json"))
+        with open(os.path.join(tokenizer_dir, "tokenizer_config.json"), "w") as f:
+            json.dump(
+                {"special_tokens": {"pad_token": "<pad>", "unk_token": "<unk>"}}, f
+            )
+
+        jsonl_path = os.path.join(temp_dir, "text.jsonl")
+        with open(jsonl_path, "w", encoding="utf-8") as f:
+            f.write(
+                json.dumps(
+                    {
+                        "text": "Hello world this is a test document with enough characters to pass the minimum length filter."
+                    }
+                )
+                + "\n"
+            )
+            f.write(
+                json.dumps(
+                    {
+                        "text": "Another document for testing purposes with sufficient length to be processed."
+                    }
+                )
+                + "\n"
+            )
+
+        config = PipelineConfig(
+            input=InputConfig(type="text", text_key="text"),
+            preprocessing=ProcessingConfig(
+                max_seq_len=2048, min_chars=10, deduplicate=True
+            ),
+            output=OutputConfig(storage_format="bin"),
+        )
+
+        out_dir = os.path.join(temp_dir, "output")
+        Pipeline(
+            config=config,
+            input_paths=[jsonl_path],
+            output_dir=out_dir,
+            tokenizer_path=tokenizer_dir,
+        ).run()
+
+        meta_path = os.path.join(out_dir, "__default__", "meta.json")
+        assert os.path.exists(meta_path)
+        with open(meta_path, "r") as f:
+            meta = json.load(f)
+        assert "sequence" in meta
+        assert "loss_mask" not in meta
+
+    def test_full_instruction_pipeline(self, temp_dir, test_tokenizer):
+        tokenizer_dir = os.path.join(temp_dir, "tok")
+        os.makedirs(tokenizer_dir, exist_ok=True)
+        test_tokenizer._tokenizer.save(os.path.join(tokenizer_dir, "tokenizer.json"))
+        with open(os.path.join(tokenizer_dir, "tokenizer_config.json"), "w") as f:
+            json.dump(
+                {"special_tokens": {"pad_token": "<pad>", "unk_token": "<unk>"}}, f
+            )
+
+        jsonl_path = os.path.join(temp_dir, "instruct.jsonl")
+        with open(jsonl_path, "w", encoding="utf-8") as f:
+            f.write(
+                json.dumps(
+                    {
+                        "prompt": "Tell me a joke",
+                        "response": "Why did the chicken cross the road?",
+                    }
+                )
+                + "\n"
+            )
+            f.write(
+                json.dumps(
+                    {
+                        "prompt": "What is AI?",
+                        "response": "Artificial Intelligence is a field of computer science.",
+                    }
+                )
+                + "\n"
+            )
+
+        config = PipelineConfig(
+            input=InputConfig(
+                type="instruction", prompt_key="prompt", response_key="response"
+            ),
+            mask={"prompt": "mask", "response": "train"},
+            mask_default="mask",
+            preprocessing=ProcessingConfig(max_seq_len=2048),
+            output=OutputConfig(storage_format="bin"),
+        )
+
+        out_dir = os.path.join(temp_dir, "output")
+        Pipeline(
+            config=config,
+            input_paths=[jsonl_path],
+            output_dir=out_dir,
+            tokenizer_path=tokenizer_dir,
+        ).run()
+
+        meta_path = os.path.join(out_dir, "__default__", "meta.json")
+        assert os.path.exists(meta_path)
+        with open(meta_path, "r") as f:
+            meta = json.load(f)
+        assert "sequence" in meta
+        assert "loss_mask" in meta
+
+
+class TestUtility:
+    def test_filter_by_length(self):
+        assert filter_by_length("hello world", min_len=5)
+        assert not filter_by_length("hi", min_len=5)
+        assert not filter_by_length("x" * 100, max_len=50)
+        assert filter_by_length("just right", min_len=5, max_len=20)
+
+    def test_dedup_signature(self):
+        a = {"key": "value", "number": 1}
+        b = {"number": 1, "key": "value"}
+        assert dedup_signature(a) == dedup_signature(b)
+        c = {"key": "different"}
+        assert dedup_signature(a) != dedup_signature(c)
+
+
+class TestFactoryRegistration:
+    def test_registered_builders(self):
+        names = MaskBuilderFactory._registry.list_names()
+        assert "chat" in names
+        assert "instruction" in names
+        assert "text" in names
+
+    def test_create_chat_builder(self):
+        builder = MaskBuilderFactory.create("chat")
+        assert isinstance(builder, ChatMaskBuilder)
+
+    def test_create_instruction_builder(self):
+        builder = MaskBuilderFactory.create("instruction")
+        assert isinstance(builder, InstructionMaskBuilder)
+
+    def test_create_text_builder(self):
+        builder = MaskBuilderFactory.create("text")
+        assert isinstance(builder, TextMaskBuilder)