use std::{cmp::Ordering, collections::{BinaryHeap, BTreeMap, HashMap}, sync::atomic::AtomicU64};
use crate::{MyError, MyResult, spec::Spec, traits::{CanStrToWord, CanToWord, CanTrain, Train}};
use super::*;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InitialAlphabet {
RawBytes,
ByteLevel,
}
impl Default for InitialAlphabet {
fn default() -> Self {
Self::RawBytes
}
}
impl InitialAlphabet {
fn bytes(self) -> Vec<u8> {
match self {
Self::RawBytes => (0u8..=255).collect(),
Self::ByteLevel => byte_level_alphabet_bytes(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TieBreak {
SmallestPairId,
LargestContent,
}
impl Default for TieBreak {
fn default() -> Self {
Self::SmallestPairId
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct BpeTrainerConfig {
pub initial_alphabet: InitialAlphabet,
pub tie_break: TieBreak,
}
impl BpeTrainerConfig {
pub fn hf_byte_level() -> Self {
Self {
initial_alphabet: InitialAlphabet::ByteLevel,
tie_break: TieBreak::SmallestPairId,
}
}
}
fn byte_level_alphabet_bytes() -> Vec<u8> {
let mut pairs = (0u8..=255)
.map(|byte| (byte, byte_to_unicode(byte)))
.collect::<Vec<_>>();
pairs.sort_by_key(|(_, ch)| *ch);
pairs.into_iter().map(|(byte, _)| byte).collect()
}
fn byte_to_unicode(byte: u8) -> char {
if (b'!'..=b'~').contains(&byte)
|| (0xA1..=0xAC).contains(&byte)
|| (0xAE..=0xFF).contains(&byte)
{
return byte as char;
}
let mut n = 0u32;
for b in 0u8..=255 {
if (b'!'..=b'~').contains(&b)
|| (0xA1..=0xAC).contains(&b)
|| (0xAE..=0xFF).contains(&b)
{
continue;
}
if b == byte {
return char::from_u32(256 + n).unwrap();
}
n += 1;
}
unreachable!("all bytes are covered")
}
#[derive(Debug, Default)]
pub struct BpeTrainer<C, I> {
pub start_vocab_idx: AtomicU64,
pub _byte_vocab_start_idx: Option<u64>,
pub byte_vocab: HashMap<u8, I>,
pub config: BpeTrainerConfig,
pub special_tokens: Vec<String>,
pub vocab: BTreeMap<I, Word<C>>,
pub merges: Vec<Merge<C, I>>,
pub pre_merges: HashMap<(I, I), Merge<C, I>>,
merge_heap: BinaryHeap<MergeCandidate<C, I>>,
pub words: Vec<PreToken<C, I>>,
}
#[derive(Debug, Clone)]
struct MergeCandidate<C, I> {
freq: Freq,
tp: (I, I),
content: (Word<C>, Word<C>),
tie_break: TieBreak,
}
impl<C, I> MergeCandidate<C, I> {
fn from_merge(merge: &Merge<C, I>, tie_break: TieBreak) -> Self
where
I: Copy,
{
Self {
freq: merge.data.freq,
tp: merge.tp,
content: merge.content.clone(),
tie_break,
}
}
}
impl<C: Ord, I: Ord> Ord for MergeCandidate<C, I> {
fn cmp(&self, other: &Self) -> Ordering {
match self.tie_break {
TieBreak::SmallestPairId => self
.freq
.cmp(&other.freq)
.then_with(|| other.tp.cmp(&self.tp)),
TieBreak::LargestContent => self
.freq
.cmp(&other.freq)
.then_with(|| self.content.cmp(&other.content)),
}
}
}
impl<C: Ord, I: Ord> PartialOrd for MergeCandidate<C, I> {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl<C: Ord, I: Ord> PartialEq for MergeCandidate<C, I> {
fn eq(&self, other: &Self) -> bool {
self.cmp(other) == Ordering::Equal
}
}
impl<C: Ord, I: Ord> Eq for MergeCandidate<C, I> {}
impl<C, I: IdxLike> BpeTrainer<C, I>
where
Word<C>: WordDebugExt,
C: CanStrToWord + CanToWord<u8>,
{
pub fn from_words<Iter: IntoIterator<Item = (S, Freq)>, S: AsRef<str>>(words: Iter, special_tokens: &[String]) -> Self
where
C: CharToIdx<I>,
I: HasChar<C>,
{
Self::from_words_with_config(words, special_tokens, BpeTrainerConfig::default())
}
pub fn from_words_with_config<Iter: IntoIterator<Item = (S, Freq)>, S: AsRef<str>>(
words: Iter, special_tokens: &[String], config: BpeTrainerConfig,
) -> Self
where
C: CharToIdx<I>,
I: HasChar<C>,
{
let vocab_start_idx = special_tokens.len() as u64;
let sp_set = special_tokens.iter().map(String::as_str).collect::<BTreeSet<_>>();
let tokens = Self::_words_to_tokens(words, vocab_start_idx, &sp_set, None);
Self::new_with_config(tokens, special_tokens.to_vec(), config)
}
pub fn new(words: Vec<PreToken<C, I>>, special_tokens: Vec<String>) -> Self {
Self::new_with_config(words, special_tokens, BpeTrainerConfig::default())
}
pub fn new_with_config(words: Vec<PreToken<C, I>>, special_tokens: Vec<String>, config: BpeTrainerConfig) -> Self {
let mut bpe = Self::empty();
bpe.config = config;
bpe._vocab_insert_special_tokens(special_tokens);
bpe._vocab_insert_all_single_byte();
bpe.words = words;
bpe
}
pub fn _vocab_insert_all_single_byte(&mut self) -> I {
let start_idx = self.start_vocab_idx.fetch_add(256, std::sync::atomic::Ordering::AcqRel);
let vocab = &mut self.vocab;
self.byte_vocab.clear();
for (offset, byte) in self.config.initial_alphabet.bytes().into_iter().enumerate() {
let idx = I::from_u64(offset as u64 + start_idx);
if byte < 128 {
vocab.insert(idx, (byte as char).to_string().to_word());
} else {
vocab.insert(idx, byte.to_word());
}
self.byte_vocab.insert(byte, idx);
}
self._byte_vocab_start_idx = Some(start_idx);
I::from_u64(start_idx + 256)
}
pub fn _words_to_tokens<Iter: IntoIterator<Item = (S, Freq)>, S: AsRef<str>>(
words: Iter, vocab_start_idx: u64, special_tokens: &BTreeSet<&str>, byte_vocab: Option<&HashMap<u8, I>>,
) -> Vec<PreToken<C, I>>
where
C: CharToIdx<I>,
{
let mut tokens = Vec::new();
for (w, freq) in words.into_iter() {
let w = w.as_ref();
if special_tokens.contains(w) {
continue;
}
let src = w.to_word();
let idxs = src.iter().map(|b| b.char_to_idx(vocab_start_idx, byte_vocab)).collect::<Vec<_>>();
let pre_token = PreToken {
src: src.clone(),
idxs,
freq: freq as Freq,
};
tokens.push(pre_token);
}
tokens
}
}
impl<C: CanStrToWord, I: IdxLike> BpeTrainer<C, I>
where
Word<C>: WordDebugExt,
{
pub fn _vocab_insert_special_tokens(&mut self, special_tokens: Vec<String>) -> I {
let length = special_tokens.len();
let start_idx = self.start_vocab_idx.fetch_add(length as u64, std::sync::atomic::Ordering::AcqRel);
let vocab = &mut self.vocab;
for (i, token) in special_tokens.iter().enumerate() {
vocab.insert(I::from_u64(i as u64 + start_idx), token.as_str().to_word());
}
self.special_tokens.extend(special_tokens);
I::from_u64(start_idx + length as u64)
}
pub fn save_vocab_json<W: std::io::Write>(&self, spec: &dyn Spec<C, I>, mut w: W) -> MyResult<()> {
spec.encode_vocab(&mut w, &self.vocab)
}
pub fn save_merges_txt<W: std::io::Write>(&self, spec: &dyn Spec<C, I>, mut w: W) -> MyResult<()> {
spec.encode_merges(&mut w, &self.merges)
}
}
impl<C, I> BpeTrainer<C, I> {
pub fn empty() -> Self {
Self {
start_vocab_idx: AtomicU64::new(0),
_byte_vocab_start_idx: None,
byte_vocab: HashMap::new(),
config: BpeTrainerConfig::default(),
vocab: BTreeMap::new(),
merges: Vec::new(),
pre_merges: HashMap::new(),
merge_heap: BinaryHeap::new(),
special_tokens: Vec::new(),
words: Vec::new(),
}
}
}
impl<C, I: IdxLike> BpeTrainer<C, I>
where
Word<C>: WordDebugExt,
I: HasChar<C>,
C: CanStrToWord + Ord,
{
pub fn _build_pre_merges(&mut self) {
debug!("Initializing BPE training with {} words", self.words.len());
self.pre_merges.clear();
self.merge_heap.clear();
let vocab_get = |i: I| {
self.vocab.get(&i).cloned().or_else(|| i.idx_to_word()).ok_or_else(|| MyError::OovIdx(i.to_u64()))
};
let i_none = I::from_u64(u64::MAX);
let w_none = char::from_u32(0x10FFFF).unwrap().to_string().to_word();
for (i, word) in self.words.iter().enumerate() {
for j in word.idxs.iter() {
if self.vocab.contains_key(j) {
continue;
}
let tp = (i_none, *j);
let merge = self.pre_merges.entry(tp).or_insert_with(|| {
let content = (
w_none.clone(),
vocab_get(*j).unwrap(),
);
Merge::new(tp, content).with_target(*j)
});
merge.data.freq += word.freq;
}
for (j1, j2) in word.idxs.iter().copied().zip(word.idxs.iter().skip(1).copied()) {
let tp = (j1, j2);
let merge = self.pre_merges.entry(tp).or_insert_with(|| {
let content = (
vocab_get(j1).unwrap(),
vocab_get(j2).unwrap(),
);
Merge::new(tp, content)
});
merge.add(i as u64, word.freq);
}
}
self.rebuild_merge_heap();
}
fn _set_vocab_idx(&mut self, start_idx: I) {
self.start_vocab_idx.store(start_idx.to_u64(), std::sync::atomic::Ordering::Release);
}
fn _add_vocab_idx(&self) -> I {
I::from_u64(self.start_vocab_idx.fetch_add(1, std::sync::atomic::Ordering::AcqRel))
}
fn push_merge_candidate(&mut self, tp: (I, I)) {
let Some(merge) = self.pre_merges.get(&tp) else {
return;
};
if merge.data.freq <= 0 {
return;
}
self.merge_heap.push(MergeCandidate::from_merge(merge, self.config.tie_break));
}
fn rebuild_merge_heap(&mut self) {
self.merge_heap = self
.pre_merges
.values()
.filter(|merge| merge.data.freq > 0)
.map(|merge| MergeCandidate::from_merge(merge, self.config.tie_break))
.collect();
}
fn update_pre_merges(&mut self, merge: &Merge<C, I>, changes: BTreeMap<(I, I), MergeData>) {
let changed_tps = changes.keys().copied().collect::<Vec<_>>();
_update_merge_map(&mut self.pre_merges, merge, changes, Some(&self.vocab));
for tp in changed_tps {
if tp != merge.tp {
self.push_merge_candidate(tp);
}
}
}
fn merge(&mut self, merge: &Merge<C, I>, target_idx: I) -> BTreeMap<(I, I), MergeData> {
_merge(&mut self.words, merge, target_idx)
}
fn _get_largest_merge(&mut self) -> Option<Merge<C, I>> {
while let Some(candidate) = self.merge_heap.pop() {
if candidate.freq <= 0 {
continue;
}
let Some(merge) = self.pre_merges.get(&candidate.tp) else {
continue;
};
if merge.data.freq != candidate.freq {
continue;
}
if merge.content != candidate.content {
continue;
}
return Some(merge.clone());
}
None
}
pub fn _step(&mut self, merge: Merge<C, I>) -> I where C: Clone {
let target_idx = self._add_vocab_idx();
if merge.target.is_some() {
self.vocab.insert(target_idx, merge.content.1.clone());
self.pre_merges.remove(&merge.tp);
return target_idx;
}
let changes = self.merge(&merge, target_idx);
let merge = merge.with_target(target_idx);
let merged_word = merge.merged_content();
self.vocab.insert(target_idx, merged_word);
assert_eq!(-changes.get(&merge.tp).map(|i| i.freq).unwrap_or(0), merge.data.freq);
metrics::histogram!("bpe_trainer.changes").record(changes.len() as f64);
self.update_pre_merges(&merge, changes);
self.pre_merges.remove(&merge.tp);
metrics::histogram!("bpe_trainer.occurs_in").record(merge.data.occurs_in.len() as f64);
metrics::histogram!("bpe_trainer.freq").record(merge.data.freq as f64);
self.merges.push(merge);
target_idx
}
pub fn finish(self) -> MyResult<BpeEncoder<C>>
where
C: Ord + Clone + Cachable + CharSplit,
{
let merges = self.merges
.into_iter()
.map(|m| {
let tp = (m.tp.0.to_u64() as Idx, m.tp.1.to_u64() as Idx);
let target = m.target.unwrap().to_u64() as Idx;
(tp, target)
})
.collect();
let vocab = self.vocab.into_iter().map(|(i, w)| (i.to_u64() as Idx, w)).collect();
BpeEncoder::new(vocab, merges, self.special_tokens)
}
pub fn _metrics(&self) {
metrics::counter!("bpe_trainer.vocab_size").absolute(self.vocab.len() as u64);
metrics::gauge!("bpe_trainer.pre_merges_count").set(self.pre_merges.len() as f64);
metrics::gauge!("bpe_trainer.words_count").set(self.words.len() as f64);
}
}
impl<C, I> Train for BpeTrainer<C, I>
where
Self: CanTrain<C, I>,
{
fn new(special_tokens: Vec<String>) -> Self {
Self::new(vec![], special_tokens)
}
fn add_words(&mut self, words: &mut dyn Iterator<Item = (&str, Freq)>) {
let special_tokens = self.special_tokens.iter().map(String::as_str).collect::<BTreeSet<_>>();
let vocab_start_idx = self._byte_vocab_start_idx.unwrap();
self.words = Self::_words_to_tokens(words, vocab_start_idx, &special_tokens, Some(&self.byte_vocab));
}
fn vocab_size(&self) -> usize {
self.vocab.len()
}
fn init_training(&mut self) {
self._build_pre_merges();
self._metrics();
}
fn step(&mut self) -> MyResult<()> {
let merge = self._get_largest_merge();
if let Some(merge) = merge {
self._step(merge);
if self.vocab_size() % 100 == 0 {
self._metrics();
}
Ok(())
} else {
Err(MyError::TrainStep)
}
}
}
#[cfg(test)]
mod tests {
use crate::{pretokenizer::DEFAULT_EOT, spec::gpt2::Gpt2Spec};
use super::*;
fn _test_bpe_merge(pretokens: &[(&str, Freq)], merges: &[((&str, &str), Vec<(&str, &str, MergeData)>)]) {
fn pretoken(s: &str, freq: Freq) -> PreToken<u8, Idx> {
let idxs = s.bytes().map(|b| b as Idx - 'a' as Idx).collect::<Vec<_>>();
PreToken {
src: s.to_word(),
idxs,
freq,
}
}
fn lookup(bpe: &BpeTrainer<u8, Idx>, s: &str) -> Option<Idx> {
bpe.vocab.iter().find_map(|(i, w)| {
if w.as_ref() == s.as_bytes() {
Some(*i)
} else {
None
}
})
}
fn display(bpe: &BpeTrainer<u8, Idx>, changes: &BTreeMap<(Idx, Idx), MergeData>) -> String {
let mut parts = Vec::new();
let target = ("__target__").to_word();
for (tp, data) in changes.iter() {
let left = bpe.vocab.get(&tp.0).unwrap_or(&target).debug_display();
let right = bpe.vocab.get(&tp.1).unwrap_or(&target).debug_display();
parts.push(format!("({:?}, {:?}, MergeData::new({}).occurs_in({:?}))", left, right, data.freq, data.occurs_in_vec()));
}
format!("{{\n {}\n}}", parts.join(",\n "))
}
let mut bpe = BpeTrainer::default();
bpe.vocab.extend(
('a' ..= 'z').enumerate().map(|(i, c)| (i as Idx, c.to_string().to_word()))
);
bpe._set_vocab_idx(100);
bpe.words.extend(
pretokens.iter().map(|(s, f)| pretoken(s, *f))
);
bpe.init_training();
for (m, expected) in merges {
let merge_tp = (
lookup(&bpe, m.0).unwrap(), lookup(&bpe, m.1).unwrap()
);
let merge = bpe.pre_merges.get(&merge_tp).unwrap().clone();
let target = bpe._add_vocab_idx();
let changes = bpe.merge(&merge, target);
assert_eq!(merge.data.freq, -changes.get(&merge_tp).cloned().unwrap().freq);
if expected.is_empty() {
continue;
}
let expected = expected.into_iter().map(|(a, b, data)| {
let tp_idx = (lookup(&bpe, a).unwrap_or(target), lookup(&bpe, b).unwrap_or(target));
(tp_idx, data.clone())
}).collect::<BTreeMap<_, _>>();
assert_eq!(changes, expected, "\nExpected changes:\n{}\nActual changes:\n{}", display(&bpe, &expected), display(&bpe, &changes));
}
}
#[test]
fn test_bpe_merge() {
_test_bpe_merge(&[("abcd", 5), ("abcdbcd", 30), ("abcbcdab", 200)], &[(("b", "c"), vec![
("a", "b", MergeData::new(-235).add_occurs_in([0, 1])),
("a", "bc", MergeData::new(235).add_occurs_in([0, 1, 2])),
("b", "c", MergeData::new(-465).add_occurs_in([0, 1, 2])),
("c", "b", MergeData::new(-200).add_occurs_in([2])),
("c", "d", MergeData::new(-265).add_occurs_in([0, 1, 2])),
("d", "b", MergeData::new(-30).add_occurs_in([1])),
("d", "bc", MergeData::new(30).add_occurs_in([1])),
("bc", "b", MergeData::new(0).add_occurs_in([2])),
("bc", "d", MergeData::new(265).add_occurs_in([0, 1, 2])),
("bc", "bc", MergeData::new(200).add_occurs_in([2])),
])]);
_test_bpe_merge(&[("wherever", 10)],
&[(("h", "e"), vec![
("e", "r", MergeData::new(-10).add_occurs_in([])),
("h", "e", MergeData::new(-10).add_occurs_in([0])),
("w", "h", MergeData::new(-10).add_occurs_in([0])),
("w", "he", MergeData::new(10).add_occurs_in([0])),
("he", "r", MergeData::new(10).add_occurs_in([0])),
])]);
_test_bpe_merge(&[("aaa", 10), ("aaaa", 1)],
&[(("a", "a"), vec![
("a", "a", MergeData::new(-23).add_occurs_in([0, 1])),
("aa", "a", MergeData::new(10).add_occurs_in([0, 1])),
("aa", "aa", MergeData::new(1).add_occurs_in([1])),
])]);
}
#[test]
fn test_bpe_step() {
let mut bpe = BpeTrainer::<u8, Idx>::from_words(vec![
("ababc", 5),
("ababcbabc", 30),
("abcbabcab", 200),
], &vec![]);
assert!(bpe.words.len() > 0);
bpe.init_training();
assert!(bpe.pre_merges.len() > 0);
for _ in 0..3 {
bpe.step().unwrap();
}
let result_vocab = bpe.vocab.into_iter().map(|(i, w)| (i, w.debug_display())).skip(256).collect::<Vec<_>>();
assert_eq!(
result_vocab,
vec![
(256, "ab".to_string()),
(257, "abc".to_string()),
(258, "babc".to_string()),
]
);
let result_merges = bpe.merges.into_iter().map(|m| {
let left = m.content.0.debug_display();
let right = m.content.1.debug_display();
(left, right, m.data.freq)
}).collect::<Vec<_>>();
assert_eq!(
result_merges,
vec![
("a".to_string(), "b".to_string(), 700),
("ab".to_string(), "c".to_string(), 465),
("b".to_string(), "abc".to_string(), 230),
]
);
}
#[test]
fn test_bpe_step_tie_breaks_by_smallest_pair_id() {
let mut bpe = BpeTrainer::<u8, Idx>::from_words(vec![
("ab", 1),
("cd", 1),
], &vec![]);
bpe.init_training();
bpe.step().unwrap();
let merge = bpe.merges.last().unwrap();
assert_eq!(merge.content.0.debug_display(), "a");
assert_eq!(merge.content.1.debug_display(), "b");
}
#[test]
fn test_bpe_step_can_tie_break_by_largest_content() {
let mut bpe = BpeTrainer::<u8, Idx>::new_with_config(
vec![],
vec![],
BpeTrainerConfig {
initial_alphabet: InitialAlphabet::RawBytes,
tie_break: TieBreak::LargestContent,
},
);
bpe.add_words(&mut vec![
("ab", 1),
("cd", 1),
].into_iter());
bpe.init_training();
bpe.step().unwrap();
let merge = bpe.merges.last().unwrap();
assert_eq!(merge.content.0.debug_display(), "c");
assert_eq!(merge.content.1.debug_display(), "d");
}
#[test]
fn test_bpe_from_words() {
const NAME: &str = "tinystories_sample_5M";
let input = std::fs::read_to_string(format!("fixtures/_words.{NAME}.json")).unwrap();
let words: BTreeMap<String, Freq> = serde_json::from_str(&input).unwrap();
let mut bpe = BpeTrainer::from_words(words, &vec![DEFAULT_EOT.to_string()]);
bpe.init_training();
let vocab_size = match NAME {
"tinystories_sample_5M" => 2000,
_ => 10000,
};
while bpe.vocab.len() < vocab_size {
bpe.step().unwrap();
}
std::fs::create_dir_all(format!("out/models/{NAME}")).ok();
bpe.save_vocab_json(&Gpt2Spec, std::fs::File::create(format!("out/models/{NAME}/vocab.json")).unwrap()).unwrap();
bpe.save_merges_txt(&Gpt2Spec, std::fs::File::create(format!("out/models/{NAME}/merges.txt")).unwrap()).unwrap();
let merges_txt = std::fs::read_to_string(format!("out/models/{NAME}/merges.txt")).unwrap();
let merges_expect_txt = std::fs::read_to_string(format!("fixtures/merges.{NAME}.txt")).unwrap();
assert_eq!(merges_txt, merges_expect_txt);
}
#[test]
fn test_bpe_from_words_uni() {
const NAME: &str = "TinyStories_all_data_zh_1M-sample";
let spec = crate::spec::uni::UniSpec;
let input = std::fs::read_to_string(format!("fixtures/_words.{NAME}.json")).unwrap();
let words: BTreeMap<String, Freq> = serde_json::from_str(&input).unwrap();
let mut bpe = BpeTrainer::<Character, CharIdx>::from_words_with_config(
words,
&vec![DEFAULT_EOT.to_string()],
BpeTrainerConfig {
initial_alphabet: InitialAlphabet::RawBytes,
tie_break: TieBreak::LargestContent,
},
);
bpe.init_training();
let vocab_size = match NAME {
"tinystories_sample_5M" | "TinyStories_all_data_zh_1M-sample" => 2000,
_ => 10000,
};
while bpe.vocab.len() < vocab_size {
bpe.step().unwrap();
}
std::fs::create_dir_all(format!("out/models/{NAME}")).ok();
bpe.save_vocab_json(&spec, std::fs::File::create(format!("out/models/{NAME}/vocab.uni.json")).unwrap()).unwrap();
bpe.save_merges_txt(&spec, std::fs::File::create(format!("out/models/{NAME}/merges.uni.txt")).unwrap()).unwrap();
let merges_txt = std::fs::read_to_string(format!("out/models/{NAME}/merges.uni.txt")).unwrap();
let merges_expect_txt = std::fs::read_to_string(format!("fixtures/merges.{NAME}.uni.txt")).unwrap();
let merges = merges_txt.trim_end().lines().collect::<Vec<_>>();
assert_eq!(merges, merges_expect_txt.lines().take(merges.len()).collect::<Vec<_>>());
}
}