tinymind-native-8b-remote-handoff/bundle/scripts/build_omni_round_curriculum.py

from __future__ import annotations

from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]


SEEDS = [
    (
        "world_context",
        "ผู้ใช้ถามข้อเท็จจริงปัจจุบัน ระบบควรวางขั้นตอนอย่างไร",
        "อ่านเวลา runtime ก่อน แยกสิ่งที่ต้องค้นเว็บจากสิ่งที่ตอบจากหลักฐานเดิมได้ ค้นหาแหล่งที่เกี่ยวข้อง fetch และ hash หลักฐาน ถ้าหลักฐานไม่พอให้ตอบแบบมีข้อจำกัดและบันทึก gate ว่าไม่ใช่ความรู้ในน้ำหนักโมเดล",
    ),
    (
        "omni_file",
        "เมื่อผู้ใช้ส่งไฟล์ไม่รู้ชนิดเข้ามา ผู้ช่วยควรทำอะไรเป็นลำดับ",
        "คัดลอกหรือดาวน์โหลดเข้า workspace แยกก่อน คำนวณ sha256 อ่าน magic bytes เดา mime/suffix เลือก parser ตาม registry แปลงเฉพาะสิ่งที่ decoder มีจริง และรายงาน limitation ถ้ารูปแบบยังไม่รองรับ",
    ),
    (
        "pure_signal",
        "อธิบาย PureSignal decoder ให้เป็นกฎ transferable",
        "ทุกไฟล์ต้องสร้างสัญญาณอย่างน้อยห้ากลุ่ม ได้แก่ byte fingerprint, structure signal, language/code signal, modality handoff และ fusion vector ทุก stage ต้องมี use field เพื่อพิสูจน์ว่าไม่มีงานศูนย์เปล่า",
    ),
    (
        "tool_grounding",
        "ถ้าต้องเรียกเครื่องมือให้ตอบแบบ parser อ่านได้แน่นอน",
        "เลือกเครื่องมือจาก policy สร้าง arguments เป็น object เท่านั้น ตรวจชนิดข้อมูลก่อนส่ง บันทึก request id และ hash ผลลัพธ์ ห้ามปน markdown หรือคำเกริ่นใน payload ที่ parser ต้องอ่าน",
    ),
    (
        "thai_natural",
        "ตอบภาษาไทยให้เป็นธรรมชาติและไม่หลอนต้องทำอย่างไร",
        "ใช้ประโยคสั้นชัดก่อน แล้วค่อยขยายเหตุผล แยกข้อเท็จจริงจากข้อสันนิษฐาน ถ้าข้อมูลไม่พอให้บอกว่าหลักฐานยังไม่พอ ไม่ใช่ตอบมั่วเพื่อให้ดูมั่นใจ",
    ),
    (
        "code_systems",
        "แนวทางแก้บั๊กโค้ดระบบให้ปลอดภัยและตรวจซ้ำได้",
        "ทำ reproduction ที่เล็กที่สุด อ่าน error จริง ระบุ invariant ของระบบ แก้เฉพาะจุดที่เกี่ยวข้อง เพิ่ม test ที่จับ regression และไม่รันคำสั่งทำลายข้อมูลโดยไม่มีเหตุผล",
    ),
    (
        "math_reasoning",
        "แก้โจทย์คณิตหรือพิสูจน์ขอบเขตให้คมต้องทำอย่างไร",
        "ตั้งตัวแปรและเงื่อนไขให้ครบ ใช้สมการที่ตรวจสอบได้ทีละบรรทัด ระบุขอบเขตและกรณีสุดขอบ แล้วสรุปว่าข้อพิสูจน์พึ่งสมมติฐานใดบ้าง",
    ),
    (
        "self_assessment",
        "หลังตอบแล้วระบบควรตรวจตัวเองอย่างไร",
        "ตรวจว่าคำตอบมีหลักฐานไหม มีช่องว่างไหม รูปแบบตรงคำสั่งไหม มี claim เกินหลักฐานไหม และระบุข้อมูลใดที่จะทำให้คำตอบผิดในอนาคต",
    ),
    (
        "multimodal_reasoning",
        "เมื่อภาพ เสียง วิดีโอ และเอกสารให้สัญญาณคนละแบบ ระบบควรรวมหลักฐานอย่างไร",
        "แปลงทุก modality เป็น evidence frame ก่อน แล้วแยกสิ่งที่มาจาก metadata, extracted text, transcript, visual frame และ byte signal จากนั้นรวมเฉพาะ claim ที่มีหลักฐานร่วมและระบุช่องว่างที่ต้องใช้ specialist decoder",
    ),
    (
        "document_tables",
        "อ่านเอกสารที่มีตารางและข้อความยาวให้แม่นต้องทำอย่างไร",
        "ต้องแยก layout, heading, paragraph, table row, table column และ citation ออกจากกันก่อน ไม่ควร flatten จนสูญเสียความสัมพันธ์ แล้วตอบโดยอ้าง page หรือ row ที่เกี่ยวข้อง",
    ),
    (
        "audio_video_timeline",
        "สรุปเสียงหรือวิดีโออย่างมีหลักฐานต้องมีขั้นตอนอะไร",
        "ต้องดึง metadata เวลา แยกช่วงเวลา สร้าง transcript หรือ frame summary แล้วผูก claim กับ timestamp ถ้าไม่มี ASR หรือ frame decoder ให้รายงานข้อจำกัดแทนการเดาเนื้อหา",
    ),
    (
        "retrieval_memory",
        "ทำให้จำบริบทยาวโดยไม่ให้ KV โตตามความยาวควรทำอย่างไร",
        "เก็บ recent window แบบ exact เพื่อคุม syntax และเก็บบริบทเก่าเป็น chunk/hash/anchor ใน ledger จากนั้น retrieve หลักฐานที่เกี่ยวข้องแล้ว regenerate K/V เฉพาะส่วนที่ต้องใช้",
    ),
    (
        "data_purity",
        "คัดข้อมูลให้เพียวบริสุทธิ์และไม่ทำให้โมเดลจำข้อสอบควรทำอย่างไร",
        "ต้อง dedupe, secret scan, license check, source manifest, holdout split ด้วย hash และ semantic overlap audit ก่อนเทรน หากแหล่งข้อมูลใกล้ benchmark ต้อง quarantine ไม่เข้า main mix",
    ),
    (
        "agent_planning",
        "Agent ที่ยืดหยุ่นแต่ไม่หลุดคำสั่งควรวางแผนอย่างไร",
        "แยก intent, constraints, allowed tools, evidence needed, execution risk และ rollback plan ก่อนลงมือ ทุก action ต้องมี observation แล้วค่อยสรุป ไม่ถือว่า tool สำเร็จเพียงเพราะวางแผนไว้",
    ),
    (
        "security_boundary",
        "ระบบต้องรับไฟล์หรือโค้ดที่เสี่ยงโดยไม่เกิดอันตรายอย่างไร",
        "ต้องห้าม execute active content โดย default ทำ static inspection, hash, sandbox containment, path allowlist และ redaction ก่อนเทรนหรือส่งต่อให้โมเดล",
    ),
    (
        "natural_dialogue",
        "ทำให้คำตอบเป็นธรรมชาติแต่ยังแม่นและตรวจสอบได้ต้องทำอย่างไร",
        "เริ่มจากคำตอบตรงคำถาม ใช้ภาษาเหมือนมนุษย์ ไม่ใส่โครงแข็งเกินจำเป็น แล้วค่อยให้หลักฐาน ข้อจำกัด และทางเลือกต่อเมื่อช่วยให้ผู้ใช้ตัดสินใจได้ดีขึ้น",
    ),
]


VARIANTS = [
    ("brief", "ตอบสั้น ตรง และมีหลักฐานกำกับ"),
    ("deep", "ตอบลึกเป็นขั้นตอนพร้อม gate"),
    ("failure", "อธิบายกรณีที่ต้องปฏิเสธการสรุปเพราะหลักฐานไม่พอ"),
    ("repair", "เสนอวิธีซ่อมเมื่อ output หลุดรูปแบบ"),
    ("audit", "ทำรายการตรวจว่าทุกขั้นมีประโยชน์ ไม่เกิดงานศูนย์เปล่า"),
    ("edge", "เน้น edge cases และข้อผิดพลาดที่มักเกิดจริง"),
    ("contrast", "เปรียบเทียบวิธีที่ถูกกับวิธีที่ทำให้หลอน"),
    ("operator", "เขียนเป็น runbook ที่ทำซ้ำและตรวจหลักฐานได้"),
]


def _sha(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _row(axis: str, user: str, answer: str, variant: str, idx: int) -> dict:
    prompt = f"{user}\nโหมด: {variant}\nห้ามตอบเกินหลักฐานและต้องบอกข้อจำกัดถ้ามี"
    response = f"{answer}\nSelf-check: ตรวจหลักฐาน รูปแบบ และ claim gate ก่อนสรุปทุกครั้ง."
    if idx % 4 == 0:
        response += "\nZero-waste: ขั้นตอนที่ไม่มี output ใช้ต่อได้ต้องตัดออกหรือรวมกับขั้นที่มีหลักฐาน."
    if idx % 5 == 0:
        response += "\nRealtime: ข้อมูลเปลี่ยนเร็วต้องผ่าน runtime/web evidence ไม่ใช่จำจากน้ำหนักโมเดล."
    fp = _sha(f"{axis}:{variant}:{idx}:{prompt}:{response}")
    return {
        "messages": [
            {
                "role": "system",
                "content": "You are TinyMind omni-round training core. Learn transferable robust behavior, not fixed answers.",
            },
            {"role": "user", "content": prompt},
            {"role": "assistant", "content": response},
        ],
        "source": "omni_round_curriculum_v1",
        "metadata": {
            "domain": axis,
            "variant": variant,
            "fingerprint_sha256": fp,
            "loss_weight": 2.4,
            "quality_tags": ["omni_round", "targeted", "no_probe_prompt", "zero_waste"],
        },
        "quality_governor": {
            "domain": axis,
            "source": "omni_round_curriculum_v1",
            "semantic_sha256": f"omni_round:{axis}:{variant}:{fp}",
            "purity_density_score": 0.99,
            "estimated_tokens": max(64, (len(prompt) + len(response)) // 4),
        },
    }


def build(out_dir: str | Path, *, repeats: int = 80) -> dict:
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    rows = []
    for idx in range(repeats):
        for axis, user, answer in SEEDS:
            variant, instruction = VARIANTS[idx % len(VARIANTS)]
            rows.append(_row(axis, f"{instruction}: {user}", answer, variant, idx))
    jsonl_path = out / "omni_round_curriculum.jsonl"
    jsonl_path.write_text("\n".join(json.dumps(row, ensure_ascii=False, sort_keys=True) for row in rows) + "\n", encoding="utf-8")
    manifest = {
        "schema": "tinymind.omni_round_curriculum.v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "jsonl_path": str(jsonl_path),
        "rows": len(rows),
        "axes": sorted({axis for axis, _, _ in SEEDS}),
        "variants": [name for name, _ in VARIANTS],
        "claim_gate": {
            "designed_for_targeted_omni_round_training": True,
            "contains_exact_probe_prompts": False,
            "world_best_claim_allowed": False,
        },
    }
    manifest_path = out / "omni_round_curriculum_manifest.json"
    manifest["json_path"] = str(manifest_path)
    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return manifest


if __name__ == "__main__":
    print(json.dumps(build(ROOT / "reports" / "omni_round_curriculum_latest"), ensure_ascii=False, indent=2, sort_keys=True))