refactor : 统一 SectionedMaskBuilder，支持可配置 dtype
- 三合一 MaskBuilder，移除 chat/instruction/text，统一为 sections 配置 - OutputConfig 增加 dtype 字段 (per-key，默认 int32) - 移除 from __future__ import annotations - 测试适配新配置格式
This commit is contained in:
@@ -1,20 +1,14 @@
|
||||
"""Pipeline configuration for JSONL preprocessing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, Optional
|
||||
from typing import Dict, List, 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"
|
||||
sections: Optional[List[Dict]] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -31,6 +25,7 @@ class OutputConfig(BaseConfig):
|
||||
domain_key: Optional[str] = None
|
||||
storage_format: str = "bin"
|
||||
max_tokens_per_shard: int = 100_000_000
|
||||
dtype: Dict[str, str] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
@@ -1,18 +1,14 @@
|
||||
from astrai.preprocessing.builder import (
|
||||
BaseMaskBuilder,
|
||||
ChatMaskBuilder,
|
||||
InstructionMaskBuilder,
|
||||
MaskBuilderFactory,
|
||||
TextMaskBuilder,
|
||||
SectionedMaskBuilder,
|
||||
)
|
||||
from astrai.preprocessing.pipeline import Pipeline, dedup_signature, filter_by_length
|
||||
|
||||
__all__ = [
|
||||
"BaseMaskBuilder",
|
||||
"ChatMaskBuilder",
|
||||
"InstructionMaskBuilder",
|
||||
"MaskBuilderFactory",
|
||||
"TextMaskBuilder",
|
||||
"SectionedMaskBuilder",
|
||||
"Pipeline",
|
||||
"dedup_signature",
|
||||
"filter_by_length",
|
||||
|
||||
+102
-104
@@ -1,13 +1,11 @@
|
||||
"""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.
|
||||
The single :class:`SectionedMaskBuilder` handles all input formats
|
||||
via declarative ``input.sections`` config.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import List, Optional
|
||||
from typing import Optional
|
||||
|
||||
from astrai.factory import BaseFactory
|
||||
|
||||
@@ -40,122 +38,122 @@ def _extract_domain(item: dict, domain_key: Optional[str]) -> str:
|
||||
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.
|
||||
def _resolve_action(action: str, role: str, config) -> str:
|
||||
"""Resolve action to "train" or "mask".
|
||||
|
||||
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.
|
||||
- ``"train"`` / ``"mask"`` → literal
|
||||
- ``"$role"`` → look up ``role`` in ``config.mask``, fall back to ``config.mask_default``
|
||||
"""
|
||||
if action == "$role":
|
||||
return config.mask.get(role, config.mask_default)
|
||||
return action
|
||||
|
||||
|
||||
@MaskBuilderFactory.register("sectioned")
|
||||
class SectionedMaskBuilder(BaseMaskBuilder):
|
||||
"""Config-driven builder: iterates over ``input.sections`` in order.
|
||||
|
||||
Each section specifies a JSONL field + mask action.
|
||||
|
||||
Section spec::
|
||||
|
||||
{
|
||||
"field": "messages", # JSONL key
|
||||
"action": "$role", # "train" | "mask" | "$role"
|
||||
"template": true, # apply chat_template per message (optional)
|
||||
"add_special_tokens": false # override encode flag (optional)
|
||||
}
|
||||
|
||||
Example configs::
|
||||
|
||||
# Chat
|
||||
{"input": {"sections": [
|
||||
{"field": "messages", "action": "$role", "template": true}
|
||||
]}}
|
||||
|
||||
# Instruction
|
||||
{"input": {"sections": [
|
||||
{"field": "prompt", "action": "mask", "add_special_tokens": true},
|
||||
{"field": "response", "action": "train"}
|
||||
]}}
|
||||
|
||||
# Text
|
||||
{"input": {"sections": [
|
||||
{"field": "text", "action": "train"}
|
||||
]}}
|
||||
"""
|
||||
|
||||
def build(self, item: dict, config, tokenizer) -> Optional[dict]:
|
||||
messages = item.get(config.input.messages_key)
|
||||
if not isinstance(messages, list) or not messages:
|
||||
sections = config.input.sections
|
||||
if not sections:
|
||||
return None
|
||||
|
||||
all_ids: List[int] = []
|
||||
spans: List[tuple] = []
|
||||
all_ids: list[int] = []
|
||||
loss_mask: list[int] = []
|
||||
|
||||
if tokenizer.bos_token_id is not None:
|
||||
has_template = any(s.get("template") for s in sections)
|
||||
is_text_config = not has_template and all(
|
||||
s["action"] == "train" for s in sections
|
||||
)
|
||||
|
||||
if has_template and tokenizer.bos_token_id is not None:
|
||||
all_ids.append(tokenizer.bos_token_id)
|
||||
loss_mask.append(0)
|
||||
|
||||
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
|
||||
first_section = True
|
||||
for sec in sections:
|
||||
field = sec["field"]
|
||||
action = sec["action"]
|
||||
use_template = sec.get("template", False)
|
||||
add_special = sec.get(
|
||||
"add_special_tokens", not use_template and first_section
|
||||
)
|
||||
ids = tokenizer.encode(rendered, add_special_tokens=False)
|
||||
|
||||
start = len(all_ids)
|
||||
all_ids.extend(ids)
|
||||
spans.append((start, len(all_ids), action))
|
||||
if use_template:
|
||||
messages = item.get(field)
|
||||
if not isinstance(messages, list) or not messages:
|
||||
continue
|
||||
for msg in messages:
|
||||
role = msg.get("role", "")
|
||||
act = _resolve_action(action, role, config)
|
||||
rendered = tokenizer.apply_chat_template(
|
||||
[msg], tokenize=False, add_generation_prompt=False
|
||||
)
|
||||
ids = tokenizer.encode(rendered, add_special_tokens=False)
|
||||
all_ids.extend(ids)
|
||||
val = 1 if act == "train" else 0
|
||||
loss_mask.extend([val] * len(ids))
|
||||
else:
|
||||
text = str(item.get(field, ""))
|
||||
if not text.strip():
|
||||
continue
|
||||
if is_text_config:
|
||||
pp = config.preprocessing
|
||||
if pp.min_chars > 0 and len(text) < pp.min_chars:
|
||||
continue
|
||||
if len(text) > pp.max_chars:
|
||||
continue
|
||||
ids = tokenizer.encode(text, add_special_tokens=add_special)
|
||||
all_ids.extend(ids)
|
||||
val = 1 if action == "train" else 0
|
||||
loss_mask.extend([val] * len(ids))
|
||||
|
||||
if len(all_ids) <= 1:
|
||||
return None
|
||||
first_section = False
|
||||
|
||||
max_len = config.preprocessing.max_seq_len
|
||||
all_ids = all_ids[:max_len]
|
||||
loss_mask = loss_mask[: len(all_ids)]
|
||||
|
||||
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)
|
||||
if not all_ids:
|
||||
return None
|
||||
|
||||
return {
|
||||
if has_template and len(all_ids) <= 1:
|
||||
return None
|
||||
|
||||
result: dict = {
|
||||
"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),
|
||||
}
|
||||
if not all(m == 1 for m in loss_mask):
|
||||
result["loss_mask"] = loss_mask
|
||||
return result
|
||||
|
||||
@@ -4,22 +4,33 @@ 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
|
||||
from itertools import chain
|
||||
from typing import 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.preprocessing.builder import SectionedMaskBuilder
|
||||
from astrai.tokenize import AutoTokenizer
|
||||
|
||||
_STR_TO_DTYPE: dict[str, torch.dtype] = {
|
||||
"bool": torch.bool,
|
||||
"uint8": torch.uint8,
|
||||
"int8": torch.int8,
|
||||
"int16": torch.int16,
|
||||
"int32": torch.int32,
|
||||
"int64": torch.int64,
|
||||
"float16": torch.float16,
|
||||
"float32": torch.float32,
|
||||
"float64": torch.float64,
|
||||
}
|
||||
|
||||
|
||||
def filter_by_length(text: str, min_len: int = 50, max_len: int = 2_000_000) -> bool:
|
||||
return min_len <= len(text) <= max_len
|
||||
@@ -42,7 +53,7 @@ class Pipeline:
|
||||
def __init__(
|
||||
self,
|
||||
config: PipelineConfig,
|
||||
input_paths: List[str],
|
||||
input_paths: list[str],
|
||||
output_dir: str,
|
||||
tokenizer_path: str,
|
||||
):
|
||||
@@ -52,7 +63,7 @@ class Pipeline:
|
||||
self.output_dir = output_dir
|
||||
self.tokenizer_path = tokenizer_path
|
||||
|
||||
self.mask_builder = MaskBuilderFactory.create(config.input.type)
|
||||
self.mask_builder = SectionedMaskBuilder()
|
||||
|
||||
def transform(self, item: dict) -> Optional[dict]:
|
||||
return self.mask_builder.build(item, self.config, self._tokenizer)
|
||||
@@ -120,7 +131,12 @@ class Pipeline:
|
||||
idx = shard_idx[domain]
|
||||
tensors = {}
|
||||
for key, ids_list in keys.items():
|
||||
tensors[key] = [torch.tensor(sum(ids_list, []), dtype=torch.long)]
|
||||
dt = _STR_TO_DTYPE.get(
|
||||
self.config.output.dtype.get(key, "int32"), torch.int32
|
||||
)
|
||||
tensors[key] = [
|
||||
torch.tensor(list(chain.from_iterable(ids_list)), dtype=dt)
|
||||
]
|
||||
chunk_dir = os.path.join(self.output_dir, domain)
|
||||
fmt = self.config.output.storage_format
|
||||
if fmt == "bin":
|
||||
|
||||
Reference in New Issue
Block a user