حرّاف
مدل زبانی فارسی، ساخته‌شده از صفر برای زبان فارسی
گفت‌وگوجای خالیکلمه‌ی بعدیاصلاح متنبردار متن
"""
CSS = """
.gradio-container { direction: rtl; max-width: 980px !important; margin: auto; }
textarea, input, .prose, .message, .md { direction: rtl; text-align: right; }
.hero { text-align: center; padding: 28px 16px 20px; border-radius: 18px; margin-bottom: 8px; color: #fff;
background: linear-gradient(135deg, #3730a3 0%, #4f46e5 45%, #0ea5e9 100%); }
.hero, .hero * { color: #ffffff !important; }
.hero-title { font-size: 2.4em; font-weight: 800; letter-spacing: 1px; }
.hero-sub { opacity: .92; margin-top: 4px; }
.badges { margin-top: 14px; display: flex; flex-wrap: wrap; gap: 8px; justify-content: center; }
.badges span { background: rgba(255,255,255,.16); border: 1px solid rgba(255,255,255,.3); padding: 3px 12px;
border-radius: 999px; font-size: .85em; }
.prompt textarea { font-size: 0.82em; opacity: 0.75; }
table, td, th, td *, th * { font-family: Vazirmatn, Tahoma, sans-serif !important; letter-spacing: normal !important; }
footer { visibility: hidden; }
"""
QUESTIONS = ["قله‌ی دماوند در کدام استان است؟", "درباره‌ی حافظ مفصل توضیح بده.", "سوره‌ی یوسف چند آیه دارد؟",
"مریم ۴۷ کتاب داشت و ۱۹ تا را به دوستش داد. چند کتاب برایش ماند؟", "یه سوال از من بپرس",
"کی تو رو ساخته؟", "sghl o,fd?", "سوره‌ی دوم قرآن چیست؟"]
WITH_OPTIONS = [["پایتخت ایران کدام شهر است؟", "اصفهان\nتهران\nشیراز\nتبریز"],
["حاصل ۲۵ ضرب در ۴ چند است؟", "۸۰\n۹۰\n۱۰۰\n۱۲۰"]]
PASSAGE = [["کوه سبلان با ارتفاع ۴۸۱۱ متر، سومین قله‌ی بلند ایران است و در استان اردبیل قرار دارد. دریاچه‌ی کوچکی در "
"دهانه‌ی آتشفشانی آن تشکیل شده است.", "سبلان در کدام استان است؟"]]
GAPS = ["حافظ شاعر بزرگ ___ است.", "پایتخت ایران شهر ___ است.", "من هر روز صبح ___ می‌خورم.",
"قرآن ۱۱۴ ___ دارد.", "بلندترین کوه ایران ___ است."]
EMBED_LINES = "تهران پایتخت ایران است.\nپایتخت ایران تهران است.\nحافظ شاعری شیرازی بود.\nدیوان حافظ پر از غزل است.\nامروز هوا بارانی است."
SENTENCES = ["من سبح زود به مدرسح رفتم و درص خواندم.", "امروز حوا خیلی خوب است و ما به پارک رفطیم.",
"این کطاب را دیروز از کطابخانه گرفتم."]
def fetch():
"""The folder with the model and its code: HARRAF_LOCAL, or the private repository."""
if os.environ.get("HARRAF_LOCAL"):
return os.environ["HARRAF_LOCAL"]
from huggingface_hub import snapshot_download
token = (os.environ.get("HARRAF_TOKEN") or os.environ.get("HF_TOKEN") or "").strip()
seen = sorted(k for k in os.environ if "HARRAF" in k.upper() or "TOKEN" in k.upper()) # names only, never values
print(f"model repository: {REPO}; settings found: {seen or 'none'}", flush=True)
if not token:
raise RuntimeError("no token: add the secret HARRAF_TOKEN (Settings -> Variables and secrets -> New secret)")
# into the Hub's cache, outside the app's folder (nothing the page serves), and blocked in launch() as well
return snapshot_download(REPO, token=token)
def cpus():
"""The cores this container may use: a Space's cgroup limit (2 on free CPU), not the host's count that
os.cpu_count() gives; too many threads on 2 cores make every step crawl."""
try:
quota, period = open("/sys/fs/cgroup/cpu.max").read().split()
if quota != "max":
return max(1, int(int(quota) / int(period)))
except Exception:
pass
try:
return max(1, len(os.sched_getaffinity(0)))
except Exception:
return max(1, min(os.cpu_count() or 1, 4))
class Harraf:
def __init__(self, folder):
sys.path.insert(0, folder)
from tokenizers import Tokenizer
from farsitree.data import EOS, normalize
from farsitree.harraf2 import build_model
from farsitree.morph import Segmenter
self.normalize, self.Segmenter = normalize, Segmenter
torch.set_num_threads(cpus())
zero = spaces is not None and os.environ.get("SPACES_ZERO_GPU") # ZeroGPU: put on «cuda» at start
self.dev = "cuda" if zero or torch.cuda.is_available() else "cpu"
ck = torch.load(os.path.join(folder, "model.pt"), map_location="cpu")
model = build_model(dict(ck["config"]))
model.load_state_dict({k: v.float() if v.is_floating_point() else v for k, v in ck["model"].items()})
self.model = model.to(self.dev).eval()
self.tok = Tokenizer.from_file(os.path.join(folder, "tokenizer.json"))
self.seg = Segmenter.load(os.path.join(folder, "morph.json"))
self.eos = self.tok.token_to_id(EOS)
self.lock = threading.Lock()
print(f"Harraf: {model.num_params() / 1e6:.0f}M parameters on {self.dev}, {cpus()} cores", flush=True)
def encode(self, text):
return self.tok.encode(self.seg(self.normalize(text))).ids
def decode(self, ids):
text = self.Segmenter.join(self.tok.decode(ids)).replace("+", "").replace("�", "")
return "".join(c for c in text if unicodedata.category(c) != "Cc")
@torch.no_grad()
def logits(self, ids):
return self.model(torch.tensor([ids[-self.model.cfg.max_seq_len:]], device=self.dev))["logits"][0, -1].float()
@staticmethod
def pick(logits, temperature, top_p=0.9):
"""Greedy at temperature 0; else sampled from the likeliest tokens that together make top_p."""
if temperature <= 0:
return int(logits.argmax())
p = torch.softmax(logits / temperature, -1)
sp, si = torch.sort(p, descending=True)
keep = torch.cumsum(sp, 0) - sp < top_p
sp = sp * keep
return int(si[torch.multinomial(sp / sp.sum(), 1)])
@torch.no_grad()
def write(self, prompt, n=160, temperature=0.0):
"""The answer, word by word (a generator): up to the end-of-text token, «پرسش:» or n tokens."""
n = max(8, min(int(n), self.model.cfg.max_seq_len // 2))
ids = [self.eos] + self.encode(prompt)
ids = ids[-(self.model.cfg.max_seq_len - n):]
start = len(ids)
with self.lock:
for i in range(n):
logits = self.logits(ids)
t = self.pick(logits, temperature)
if t == self.eos:
break
ids.append(t)
text = self.decode(ids[start:])
if "پرسش:" in text:
break
if i % 3 == 2:
yield text.strip()
yield self.decode(ids[start:]).split("پرسش:")[0].strip()
@torch.no_grad()
def next_tokens(self, text, k=10):
ids = [self.eos] + (self.encode(text) if text.strip() else [])
with self.lock:
p = torch.softmax(self.logits(ids), -1)
top = torch.topk(p, k)
rows = []
for prob, i in zip(top.values.tolist(), top.indices.tolist()):
piece = "⏎ پایان متن" if i == self.eos else self.tok.decode([i]).replace("�", "")
rows.append([piece.replace("+", "ـ").strip() or "␣", f"{100 * prob:.1f}٪".translate(FA_DIGITS)])
return rows
class Tools:
"""What else the model can be asked, with no other program: how likely it finds a text, what fits a blank,
and a vector for a text (the mean of its last hidden states)."""
REFERENCE = ["امروز هوا آفتابی است.", "قیمت نان دوباره بالا رفت.", "حافظ در شیراز به دنیا آمد.", "تیم ملی فوتبال برد.",
"آب در صد درجه می‌جوشد.", "کتاب را روی میز گذاشتم.", "این رستوران غذای خوبی دارد.", "دانشگاه تهران قدیمی است.",
"او هر روز ورزش می‌کند.", "قطار ساعت هشت حرکت می‌کند.", "باران تمام شب بارید.", "رایانه‌ام خراب شده است."]
def __init__(self, h):
self.h = h
self.center = None # made the first time a vector is asked for, not at start
def _center(self):
if self.center is None:
self.center = False # while it is made, vectors are taken plain
self.center = torch.stack([self.vector(t) for t in self.REFERENCE]).mean(0) # what all texts share
return self.center
@torch.no_grad()
def logprob(self, text):
"""The model's log-probability of text (after the start token), and the number of its tokens."""
ids = [self.h.eos] + self.h.encode(text)
ids = ids[-self.h.model.cfg.max_seq_len:]
with self.h.lock:
logits = self.h.model(torch.tensor([ids], device=self.h.dev))["logits"][0, :-1].float()
lp = torch.log_softmax(logits, -1).gather(1, torch.tensor(ids[1:], device=self.h.dev)[:, None])
return float(lp.sum()), len(ids) - 1
def words(self, text, limit=60):
"""(word, how likely the model found it there) for each word: low means unexpected here."""
ws = text.split()[:limit]
out, prev = [], 0.0
for i in range(len(ws)):
lp, _ = self.logprob(" ".join(ws[:i + 1]))
out.append((ws[i], math.exp(lp - prev)))
prev = lp
return out
@torch.no_grad()
def fill(self, text, k=16, show=6):
"""Words for «___» in text, by how likely the whole text is with each (the model reads both sides)."""
left, _, right = text.partition("___")
ids = [self.h.eos] + self.h.encode(left.rstrip()) if left.strip() else [self.h.eos]
with self.h.lock:
p = torch.softmax(self.h.logits(ids), -1)
cands = []
for t in torch.topk(p, k).indices.tolist():
if t == self.h.eos:
continue
seq = [t]
with self.h.lock:
for _ in range(3): # a word may be more than one token
nxt = int(self.h.logits(ids + seq).argmax())
if nxt == self.h.eos or " " in self.h.decode(seq + [nxt]).strip():
break
seq.append(nxt)
w = self.h.decode(seq).strip().split(" ")[0].strip("،.؟!:")
if w and w not in cands:
cands.append(w)
scored = []
for w in cands:
full = (left.rstrip() + " " + w + " " + right.lstrip()).strip()
scored.append((w, self.logprob(full)[0]))
if not scored:
return []
top = max(x for _, x in scored)
z = sum(math.exp(x - top) for _, x in scored)
return sorted(((w, math.exp(x - top) / z) for w, x in scored), key=lambda x: -x[1])[:show]
@torch.no_grad()
def vector(self, text):
ids = [self.h.eos] + self.h.encode(text)
ids = ids[-self.h.model.cfg.max_seq_len:]
with self.h.lock:
hid = self.h.model(torch.tensor([ids], device=self.h.dev), return_hidden=True)["hidden"][0, 1:].float()
v = hid.mean(0)
v = v / v.norm()
center = self._center() if self.center is not False else None
if center is not None and center is not False: # without what every text has in common, the rest stands out
v = v - center
v = v / v.norm()
return v
def with_options(question, options):
"""«… گزینه‌ها: ۱) … ۲) …» when choices are given."""
q = question.strip()
opts = [o.strip() for o in (options or "").split("\n") if o.strip()]
if opts:
q += " گزینه‌ها: " + " ".join(f"{str(i).translate(FA_DIGITS)}) {o}" for i, o in enumerate(opts, 1))
return q
def frame(history, question, options):
"""The text given to the model, in the frame it was taught on."""
parts = [f"پرسش: {q} پاسخ: {a}" for q, a in history]
return " ".join(parts + [f"پرسش: {with_options(question, options)} پاسخ:"])
H = T = None
ERROR = None
MODEL_DIR = None
try:
MODEL_DIR = fetch()
H = Harraf(MODEL_DIR)
T = Tools(H)
except Exception as e: # the page still opens and says why
ERROR = f"مدل بارگذاری نشد: {type(e).__name__}: {str(e)[:300]}"
print(ERROR, flush=True)
@GPU(duration=120)
def ask(question, options, passage, keep, chat, pairs, n, temperature):
question, options, passage = question or "", options or "", passage or "" # Gradio 6 sends None for an empty box
chat, pairs = chat or [], pairs or []
if not question.strip():
yield chat, pairs, "", ""
return
if H is None:
yield chat + [{"role": "user", "content": question}, {"role": "assistant", "content": ERROR}], pairs, "", ""
return
if passage.strip(): # answering from a text: the frame «