| """Generate intra-sentence code-switched Nigerian sentences (tagged spans) with Qwen3-30B-A3B-Instruct-2507-FP8 (Apache-2.0), offline vLLM. |
| -> /scratch/asr/synth/cs_text.jsonl {id, tagged, pair, topic} |
| /scratch/vllm/bin/python gen_cs_text.py [n_prompts_per_pair]""" |
| import json, os, random, re, sys |
| from vllm import LLM, SamplingParams |
| random.seed(5) |
| OUT = "/scratch/asr/synth"; os.makedirs(OUT, exist_ok=True) |
| NAME = {"eng": "English", "ibo": "Igbo", "yor": "Yoruba", "hau": "Hausa", "pcm": "Nigerian Pidgin"} |
| |
| PAIRS = [("ibo", "eng"), ("eng", "ibo"), ("yor", "eng"), ("eng", "yor"), ("hau", "eng"), ("eng", "hau"), |
| ("pcm", "ibo"), ("pcm", "yor"), ("pcm", "hau"), ("eng", "pcm"), ("pcm", "eng")] |
| TOPICS = ["greeting a friend and asking about family", "a phone call about being late", "buying food at the market and bargaining", |
| "taking a bus, keke or okada in traffic", "school, exams and lecturers", "work, a boss and a deadline", "church or mosque on the weekend", |
| "football, a Premier League match", "a bank transfer that failed", "fixing a phone or laptop", "a wedding or naming ceremony", |
| "cooking jollof rice or soup at home", "rain, heat and NEPA light going off", "a hospital visit", "a WhatsApp group chat argument", |
| "a job interview", "rent and the landlord", "music, Afrobeats and a concert", "a birthday party plan", "giving directions in town", |
| "asking an AI assistant for help", "a business deal with a customer", "complaining about fuel prices", "studying for JAMB or WAEC", |
| "a village visit for Christmas", "a doctor's advice", "ordering food delivery", "a startup and a product launch"] |
| STY = ["casual and friendly", "excited", "polite and respectful", "complaining", "joking", "in a hurry", "telling a story"] |
|
|
| def prompt(m, e, topic, sty, n=15): |
| sw = (f"Most of each sentence is in {NAME[m]}, with {NAME[e]} words or phrases switched in mid-sentence, the way Nigerians really talk." |
| if m != "eng" else f"Most of each sentence is in English, with {NAME[e]} words, phrases or clauses switched in mid-sentence, the way Nigerians really talk.") |
| return (f"Write {n} different spoken sentences a Nigerian might say about: {topic}. Tone: {sty}. {sw}\n" |
| f"Rules: every sentence switches language at least once INSIDE the sentence (not just at the start or end); 6 to 24 words; " |
| f"natural everyday speech; use correct standard spelling ({'Yoruba with tone marks and dots' if 'yor' in (m, e) else ''}" |
| f"{'Igbo with dotted vowels (ị ọ ụ ṅ)' if 'ibo' in (m, e) else ''}{'Hausa with hooked letters (ɓ ɗ ƙ ƴ) where needed' if 'hau' in (m, e) else ''}); " |
| f"write numbers as words; no emojis, no hashtags, no quotation marks.\n" |
| f"Wrap every span in its language tag: <{m}>...</{m}> and <{e}>...</{e}>. Example format: " |
| f"<{m}>...</{m}> <{e}>...</{e}> <{m}>...</{m}>\n" |
| f"Output ONLY a JSON array of {n} strings.") |
|
|
| N = int(sys.argv[1]) if len(sys.argv) > 1 else 45 |
| jobs = [] |
| for m, e in PAIRS: |
| k = N if "pcm" not in (m, e) or (m, e) in (("pcm", "eng"), ("eng", "pcm")) else int(N * 0.6) |
| for i in range(k): |
| t, s = random.choice(TOPICS), random.choice(STY); jobs.append(dict(pair=f"{m}+{e}", topic=t, msg=prompt(m, e, t, s))) |
| llm = LLM("Qwen/Qwen3-30B-A3B-Instruct-2507-FP8", gpu_memory_utilization=float(os.environ.get("GPU_UTIL", "0.42")), max_model_len=4096, |
| enforce_eager=False) |
| sp = SamplingParams(temperature=0.9, top_p=0.95, max_tokens=1800, seed=0) |
| outs = llm.chat([[{"role": "user", "content": j["msg"]}] for j in jobs], sp) |
| TAG = re.compile(r"<(eng|ibo|yor|hau|pcm)>(.*?)</\1>", re.S) |
| seen, n_ok, n_bad = set(), 0, 0 |
| with open(f"{OUT}/cs_text.jsonl", "w") as f: |
| for j, o in zip(jobs, outs): |
| txt = o.outputs[0].text; i0, i1 = txt.find("["), txt.rfind("]") |
| try: arr = json.loads(txt[i0: i1 + 1]) |
| except Exception: n_bad += 1; continue |
| for s in arr: |
| if not isinstance(s, str): continue |
| spans = [(l, re.sub(r"\s+", " ", t).strip()) for l, t in TAG.findall(s)] |
| spans = [(l, t) for l, t in spans if t] |
| langs = [l for l, _ in spans]; nw = sum(len(t.split()) for _, t in spans) |
| if len(spans) < 2 or len(set(langs)) < 2 or not (5 <= nw <= 30) or set(langs) - set(j["pair"].split("+")): n_bad += 1; continue |
| if any(len(t) > 200 for _, t in spans) or re.search(r"[<>\"#@]", "".join(t for _, t in spans)): n_bad += 1; continue |
| key = " ".join(t for _, t in spans).lower() |
| if key in seen: continue |
| seen.add(key); tagged = " ".join(f"<{l}>{t}</{l}>" for l, t in spans) |
| f.write(json.dumps(dict(id=f"cs{n_ok:05d}", tagged=tagged, pair=j["pair"], topic=j["topic"]), ensure_ascii=False) + "\n"); n_ok += 1 |
| print("CS_TEXT_DONE ok", n_ok, "bad", n_bad, "prompts", len(jobs)) |
|
|
