پروژه: توکنایزر و آزمایشگاه دما

"""توکنایزر BPE بایتی — همان ایده‌ای که GPT و Llama به کار می‌برند، در ۵۰ خط.

ایده: از بایت‌های UTF-8 شروع کن (۲۵۶ نماد). هر بار پرتکرارترین جفت
کنار هم را پیدا کن و آن را یک نماد تازه کن. آن‌قدر ادامه بده تا
واژگان به اندازهٔ دلخواه برسد.

نتیجهٔ مهم برای فارسی: توکنایزر فقط جفت‌هایی را یاد می‌گیرد که در
دادهٔ آموزشی‌اش پرتکرار بوده‌اند. اگر داده بیشتر انگلیسی باشد،
متن فارسی تکه‌تکه‌تر — و گران‌تر — می‌شود.
"""
import json
from collections import Counter
from pathlib import Path


def _merge(ids: list[int], pair: tuple[int, int], new_id: int) -> list[int]:
    """همهٔ رخدادهای pair را در ids با new_id جایگزین می‌کند."""
    out, i = [], 0
    while i < len(ids):
        if i < len(ids) - 1 and (ids[i], ids[i + 1]) == pair:
            out.append(new_id)
            i += 2
        else:
            out.append(ids[i])
            i += 1
    return out


class BPETokenizer:
    def __init__(self) -> None:
        self.merges: dict[tuple[int, int], int] = {}   # (a, b) -> نماد تازه

    def train(self, text: str, vocab_size: int = 512) -> "BPETokenizer":
        ids = list(text.encode("utf-8"))
        for new_id in range(256, vocab_size):
            pairs = Counter(zip(ids, ids[1:]))
            if not pairs:
                break
            pair, count = pairs.most_common(1)[0]
            if count < 2:                      # جفتی که تکرار نشده، ارزش نماد ندارد
                break
            self.merges[pair] = new_id
            ids = _merge(ids, pair, new_id)
        return self

    def encode(self, text: str) -> list[int]:
        ids = list(text.encode("utf-8"))
        while len(ids) >= 2:
            # جفتی را ادغام کن که زودتر یاد گرفته شده (همان ترتیب آموزش)
            pair = min(set(zip(ids, ids[1:])), key=lambda p: self.merges.get(p, float("inf")))
            if pair not in self.merges:
                break
            ids = _merge(ids, pair, self.merges[pair])
        return ids

    def decode(self, ids: list[int]) -> str:
        vocab = {i: bytes([i]) for i in range(256)}
        for (a, b), i in self.merges.items():   # ترتیب درج = ترتیب یادگیری
            vocab[i] = vocab[a] + vocab[b]
        return b"".join(vocab[i] for i in ids).decode("utf-8", errors="replace")

    def count(self, text: str) -> int:
        return len(self.encode(text))

    def save(self, path: Path) -> None:
        path.write_text(json.dumps([[a, b, i] for (a, b), i in self.merges.items()]), encoding="utf-8")

    @classmethod
    def load(cls, path: Path) -> "BPETokenizer":
        tok = cls()
        tok.merges = {(a, b): i for a, b, i in json.loads(path.read_text(encoding="utf-8"))}
        return tok


_DEFAULT: BPETokenizer | None = None


def default_tokenizer() -> BPETokenizer:
    """توکنایزر ذهن‌یار: آموزش‌دیده روی متن درس‌ها، یک بار ساخته و ذخیره می‌شود."""
    global _DEFAULT
    if _DEFAULT is None:
        from .corpus import DATA, load
        path = DATA / "tokenizer.json"
        if path.exists():
            _DEFAULT = BPETokenizer.load(path)
        else:
            text = "\n".join(load("fa") + load("en"))[:40_000]
            _DEFAULT = BPETokenizer().train(text, vocab_size=1024)
            _DEFAULT.save(path)
    return _DEFAULT

def count_tokens(text: str) -> int:
    """شمارش واقعی با توکنایزر BPE خود ذهن‌یار (فصل ۲)."""
    from ..tokenizer import default_tokenizer
    return max(1, default_tokenizer().count(text))

"""نمونه‌برداری: از لوجیت تا توکن بعدی.

همهٔ پیچ‌های درس نمونه‌برداری اینجاست: سافت‌مکس با دما، top-k،
top-p و انتخاب تصادفی با بذر ثابت.
"""
import math
import random


def softmax(logits: dict[str, float], temperature: float = 1.0) -> dict[str, float]:
    """لوجیت ← احتمال. دمای صفر یعنی حریصانه: همهٔ احتمال به بزرگ‌ترین."""
    if temperature <= 0:
        best = max(logits, key=logits.get)
        return {t: (1.0 if t == best else 0.0) for t in logits}
    scaled = {t: x / temperature for t, x in logits.items()}
    top = max(scaled.values())                      # برای جلوگیری از سرریز عددی
    exps = {t: math.exp(x - top) for t, x in scaled.items()}
    total = sum(exps.values())
    return {t: e / total for t, e in exps.items()}


def top_k(probs: dict[str, float], k: int) -> dict[str, float]:
    kept = dict(sorted(probs.items(), key=lambda kv: -kv[1])[:k])
    total = sum(kept.values())
    return {t: p / total for t, p in kept.items()}


def top_p(probs: dict[str, float], p: float) -> dict[str, float]:
    kept, running = {}, 0.0
    for t, q in sorted(probs.items(), key=lambda kv: -kv[1]):
        kept[t] = q
        running += q
        if running >= p:                            # به p رسیدیم؛ بقیه بیرون
            break
    total = sum(kept.values())
    return {t: q / total for t, q in kept.items()}


def sample(probs: dict[str, float], rng: random.Random) -> str:
    r, running = rng.random(), 0.0
    for t, q in probs.items():
        running += q
        if r < running:
            return t
    return t                                        # گرد کردن عددی

"""یک مدل زبانی واقعی — کوچک، ولی با همان سازوکار مدل‌های بزرگ.

«دوکلمه‌ای» (bigram) است: احتمال کلمهٔ بعدی را فقط از روی کلمهٔ قبلی
حساب می‌کند. مدل‌های بزرگ همین کار را با کل متن قبلی و میلیاردها
پارامتر می‌کنند؛ ولی خروجی هر دو یکی است: یک جدول لوجیت برای توکن
بعدی، که نمونه‌برداری از آن یکی را انتخاب می‌کند.

برای خوانا بودن خروجی، واحد این مدل «کلمه» است نه توکن BPE.
"""
import math
import random
from collections import Counter, defaultdict

from . import sampling

START, END = "<s>", "</s>"


class BigramLM:
    def __init__(self, sentences: list[str]) -> None:
        self.counts: dict[str, Counter] = defaultdict(Counter)
        for s in sentences:
            words = [START] + s.split() + [END]
            for a, b in zip(words, words[1:]):
                self.counts[a][b] += 1

    def prob(self, prev: str, word: str, alpha: float = 0.1) -> float:
        """احتمال word پس از prev، با هموارسازی (فصل ۳).

        بدون هموارسازی، هر جفتی که در داده ندیده احتمال صفر می‌گیرد و
        سرگشتگی بی‌نهایت می‌شود. alpha به هر جفت ممکن کمی شانس می‌دهد.
        """
        vocab = len(self.counts) + 1
        row = self.counts[prev]
        return (row[word] + alpha) / (sum(row.values()) + alpha * vocab)

    def logits(self, prev: str) -> dict[str, float]:
        """لوجیت هر کلمهٔ بعدی = لگاریتم تعداد دفعه‌هایی که پس از prev آمده."""
        return {w: math.log(n) for w, n in self.counts[prev].items()}

    def next_probs(self, prev: str, temperature: float = 1.0,
                   k: int | None = None, p: float | None = None) -> dict[str, float]:
        probs = sampling.softmax(self.logits(prev), temperature)
        if k:
            probs = sampling.top_k(probs, k)
        if p:
            probs = sampling.top_p(probs, p)
        return probs

    def generate(self, start: str, max_words: int = 14, temperature: float = 1.0,
                 k: int | None = None, p: float | None = None, seed: int = 0) -> str:
        rng = random.Random(seed)                       # بذر ثابت = خروجی تکرارپذیر
        words, prev = [start], start
        for _ in range(max_words):
            if not self.counts[prev]:
                break
            prev = sampling.sample(self.next_probs(prev, temperature, k, p), rng)
            if prev == END:                             # شرط توقف: پایان جمله
                break
            words.append(prev)
        return " ".join(words)

"""خروجی ساختاریافته — سه راه سبک از درس نمونه‌برداری، کنار هم.

۱. پرامپت: از مدل JSON می‌خواهیم.
۲. پس‌پردازش: خطاهای تکراری را با کد درست می‌کنیم.
۳. چند بار پرسیدن: اگر هنوز خراب بود، دوباره می‌پرسیم.
"""
import json
import re

from .providers import Message, Provider


def repair_json(text: str) -> dict | None:
    """رایج‌ترین خطاها را درست می‌کند: قاب ```، متن اضافه، کروشهٔ جاافتاده."""
    text = re.sub(r"```(?:json)?", "", text)
    start = text.find("{")
    if start < 0:
        return None
    text = text[start:].strip()
    # کروشه‌های باز را می‌شماریم (بیرون از رشته‌ها) و هرچه باز مانده می‌بندیم
    stack, in_str, escape = [], False, False
    for ch in text:
        if in_str:
            if escape:
                escape = False
            elif ch == "\\":
                escape = True
            elif ch == '"':
                in_str = False
        elif ch == '"':
            in_str = True
        elif ch in "{[":
            stack.append("}" if ch == "{" else "]")
        elif ch in "}]" and stack:
            stack.pop()
    if in_str:
        text += '"'
    text += "".join(reversed(stack))
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        return None
    return data if isinstance(data, dict) else None


def ask_json(provider: Provider, question: str, keys: list[str], attempts: int = 3):
    """تا attempts بار می‌پرسد؛ اولین پاسخی را که JSON درست با همهٔ کلیدها بود برمی‌گرداند."""
    prompt = (f"{question}\nفقط یک شیء JSON برگردان با کلیدهای: {', '.join(keys)}. "
              "هیچ متن دیگری ننویس.")
    log = []
    for i in range(1, attempts + 1):
        raw = provider.complete([Message("user", prompt)], temperature=0.7 + 0.1 * i).text
        data = repair_json(raw)
        ok = data is not None and all(k in data for k in keys)
        log.append((i, raw, ok))
        if ok:
            return data, log
    return None, log

"""پروژهٔ فصل ۲ — توکنایزر، مدل زبانی و آزمایشگاه دما.

اجرا:
    python -m projects.ch02_sampling.run
"""
import json

from zehnyar import _console
from zehnyar.corpus import load, parallel, sentences
from zehnyar.lm import BigramLM
from zehnyar.providers import Completion
from zehnyar.providers.mock import MockProvider
from zehnyar.structured import ask_json
from zehnyar.tokenizer import BPETokenizer

_console.setup()


def part1_tokenizer() -> None:
    print("\n═══ ۱. توکنایزر: فارسی چند برابر توکن می‌خورد؟ ═══")
    print(f"«سلام» = {len('سلام'.encode())} بایت   |   «hello» = {len('hello'.encode())} بایت")

    en, fa = "\n".join(load("en")), "\n".join(load("fa"))
    # مثل اینترنت: بیشتر انگلیسی، کمی فارسی — در برابر دادهٔ متوازن
    internet = en + "\n" + fa[: len(en) // 20]
    balanced = en + "\n" + fa[: len(en)]
    tok_internet = BPETokenizer().train(internet, vocab_size=600)
    tok_balanced = BPETokenizer().train(balanced, vocab_size=600)

    print(f"\n{'':4}{'فارسی':>8}{'انگلیسی':>9}   (توکنایزر «اینترنتی»)")
    fa_i = en_i = fa_b = en_b = 0
    for n, (f, e) in enumerate(parallel(), start=1):
        a, b = tok_internet.count(f), tok_internet.count(e)
        fa_i, en_i = fa_i + a, en_i + b
        fa_b, en_b = fa_b + tok_balanced.count(f), en_b + tok_balanced.count(e)
        print(f"{n:>3}.{a:>8}{b:>9}")
    print(f"\nتوکنایزر «اینترنتی»: فارسی {fa_i / en_i:.1f} برابر انگلیسی")
    print(f"توکنایزر «متوازن»:   فارسی {fa_b / en_b:.1f} برابر انگلیسی")


def part2_model() -> BigramLM:
    print("\n═══ ۲. مدل زبانی: بعد از «مدل» چه می‌آید؟ ═══")
    lm = BigramLM(sentences(load("fa")))
    for t in (0.5, 1.0, 2.0):
        probs = lm.next_probs("مدل", temperature=t)
        top = sorted(probs.items(), key=lambda kv: -kv[1])[:4]
        line = "  ".join(f"{w} {p:.0%}" for w, p in top)
        print(f"دما {t}:  {line}")
    return lm


def part3_lab(lm: BigramLM) -> None:
    print("\n═══ ۳. آزمایشگاه دما ═══")
    for t in (0.2, 1.0, 1.8):
        print(f"\nدما {t}:")
        for seed in (1, 2):
            print(f"  بذر {seed}: {lm.generate('مدل', temperature=t, seed=seed)}")
    print("\nدما 1.8 ولی با top-p = 0.9:")
    for seed in (1, 2):
        print(f"  بذر {seed}: {lm.generate('مدل', temperature=1.8, p=0.9, seed=seed)}")


class SloppyJSONProvider(MockProvider):
    """شبیه‌ساز سه خطای رایج مدل‌ها در JSON — به همین ترتیب:
    ۱. قاب ``` و کروشهٔ پایانی جاافتاده   ۲. پیش‌درآمد مؤدبانه و بی‌JSON   ۳. درست
    """
    model = "sloppy-json-v1"
    ANSWERS = [
        '```json\n{"topic": "نمونه‌برداری", "level": "میانی"',
        "حتماً! این درس دربارهٔ نمونه‌برداری است و سطحش میانی است.",
        '{"topic": "نمونه‌برداری", "level": "میانی"}',
    ]

    def __init__(self) -> None:
        self.calls = 0

    def complete(self, messages, **params) -> Completion:
        text = self.ANSWERS[self.calls % len(self.ANSWERS)]
        self.calls += 1
        return Completion(text=text, model=self.model, provider=self.name)


def part4_structured() -> None:
    print("\n═══ ۴. خروجی ساختاریافته ═══")
    for start in (0, 1):
        bot = SloppyJSONProvider()
        bot.calls = start
        data, log = ask_json(bot, "موضوع و سطح این درس چیست؟", ["topic", "level"])
        for i, raw, ok in log:
            print(f"  تلاش {i}: {'✓' if ok else '✗'}  {raw.splitlines()[0][:48]}")
        print(f"  ← نتیجه: {json.dumps(data, ensure_ascii=False)}\n")


def main() -> None:
    print("=" * 50)
    print(" ذهن‌یار v0.2 — فصل ۲")
    print("=" * 50)
    part1_tokenizer()
    lm = part2_model()
    part3_lab(lm)
    part4_structured()


if __name__ == "__main__":
    main()

python -m projects.ch02_sampling.run

==================================================
 ذهن‌یار v0.2 — فصل ۲
==================================================

═══ ۱. توکنایزر: فارسی چند برابر توکن می‌خورد؟ ═══
«سلام» = 8 بایت   |   «hello» = 5 بایت

       فارسی  انگلیسی   (توکنایزر «اینترنتی»)
  1.      37        6
  2.      51       15
  3.      42       16
  4.      49       19
  5.      57       24
  6.      42       16
  7.      55       23
  8.      49       12

توکنایزر «اینترنتی»: فارسی 2.9 برابر انگلیسی
توکنایزر «متوازن»:   فارسی 1.1 برابر انگلیسی

═══ ۲. مدل زبانی: بعد از «مدل» چه می‌آید؟ ═══
دما 0.5:  را 55%  زبانی 7%  از 4%  یاد 2%
دما 1.0:  را 12%  زبانی 4%  از 3%  یاد 3%
دما 2.0:  را 4%  زبانی 2%  از 2%  یاد 2%

═══ ۳. آزمایشگاه دما ═══

دما 0.2:
  بذر 1: مدل را به کار شما را به کار را به آن را به کار نمی‌کند
  بذر 2: مدل را به آن را به کار را به کار را به مدل را به

دما 1.0:
  بذر 1: مدل به‌طور خودکار — و برای همین تحلیل Common Crawl است: فاصلهٔ نمرهٔ Gemini —
  بذر 2: مدل بود: مدل با تعداد گزینه‌ها ثابت کردن آن ساخته شده‌اند

دما 1.8:
  بذر 1: مدل تا در سطح انگیزه وجود دارد و روزبه‌روز کمتر دربارهٔ انتخاب کنید
  بذر 2: مدل ۱۰۰ میلیون عکس ۵ از همیشه چندسَر است: این نکته دوباره به زبانی فقط

دما 1.8 ولی با top-p = 0.9:
  بذر 1: مدل پایه قوی‌تر را انتخاب شما معمولاً عمر کوتاهی دارند: seq2seq حدود ۵٫۲ × ۲
  بذر 2: مدل «دستوری» را از روی عدد آخر راه می‌اندازید؛ مدل بستگی دارد

═══ ۴. خروجی ساختاریافته ═══
  تلاش 1: ✓  ```json
  ← نتیجه: {"topic": "نمونه‌برداری", "level": "میانی"}

  تلاش 1: ✗  حتماً! این درس دربارهٔ نمونه‌برداری است و سطحش م
  تلاش 2: ✓  {"topic": "نمونه‌برداری", "level": "میانی"}
  ← نتیجه: {"topic": "نمونه‌برداری", "level": "میانی"}