|
|
|
|
|
""" |
|
|
Generate a synthetic UltraChat-style SFT dataset. |
|
|
|
|
|
Each JSONL line has: |
|
|
{ |
|
|
"prompt": "<first user message>", |
|
|
"messages": [{"content": "...", "role": "user"}, {"content": "...", "role": "assistant"}, ...], |
|
|
"prompt_id": "<sha256 hex of prompt>" |
|
|
} |
|
|
|
|
|
Usage: |
|
|
python scripts/generate_ultrachat_sft.py --out sample_data/train_sft.jsonl --n 500 |
|
|
|
|
|
Only uses Python standard library. |
|
|
""" |
|
|
from __future__ import annotations |
|
|
|
|
|
import argparse |
|
|
import hashlib |
|
|
import json |
|
|
import os |
|
|
import random |
|
|
import sys |
|
|
from typing import Any, Dict, List, Sequence |
|
|
|
|
|
TOPICS = [ |
|
|
"travel planning", |
|
|
"cooking and recipes", |
|
|
"software engineering", |
|
|
"data science", |
|
|
"machine learning", |
|
|
"mathematics", |
|
|
"history", |
|
|
"literature", |
|
|
"productivity", |
|
|
"fitness and health", |
|
|
"economics", |
|
|
"marketing", |
|
|
"photography", |
|
|
"music theory", |
|
|
"language learning", |
|
|
"gardening", |
|
|
"home improvement", |
|
|
"career advice", |
|
|
"public speaking", |
|
|
"time management", |
|
|
"networking", |
|
|
"resume writing", |
|
|
"interview prep", |
|
|
"cloud computing", |
|
|
"kubernetes", |
|
|
"devops", |
|
|
"cybersecurity", |
|
|
"robotics", |
|
|
"electronics", |
|
|
"astronomy", |
|
|
"climate and environment", |
|
|
"education", |
|
|
"parenting", |
|
|
"pet care", |
|
|
"mental wellness", |
|
|
"philosophy", |
|
|
"ethics", |
|
|
"UX design", |
|
|
"UI design", |
|
|
"copywriting", |
|
|
"game design", |
|
|
"board games", |
|
|
"coffee brewing", |
|
|
"tea brewing", |
|
|
"urban planning", |
|
|
"transportation", |
|
|
"sports science", |
|
|
"statistics", |
|
|
] |
|
|
|
|
|
USER_OPENERS = [ |
|
|
"Could you help me with {topic}?", |
|
|
"What are the key steps to get started with {topic}?", |
|
|
"Give me a concise plan to improve at {topic} over 4 weeks.", |
|
|
"Explain a few common pitfalls in {topic} and how to avoid them.", |
|
|
"Draft a checklist for beginners in {topic}.", |
|
|
"Compare two approaches commonly used in {topic} and when to choose each.", |
|
|
"I have 2 hours to learn about {topic} today—what should I do?", |
|
|
"What metrics matter most in {topic} and how do I track them?", |
|
|
"Summarize an actionable framework for {topic} with examples.", |
|
|
] |
|
|
|
|
|
FOLLOW_UPS = [ |
|
|
"Nice. Can you add 3 concrete examples?", |
|
|
"Could you turn that into a step-by-step guide?", |
|
|
"What are likely failure modes and mitigations?", |
|
|
"How would a beginner apply this in a weekend project?", |
|
|
"Please give a small template I can reuse.", |
|
|
"How do I measure progress over a month?", |
|
|
] |
|
|
|
|
|
DEEPENERS = [ |
|
|
"Great. Any tips to make it resilient under constraints?", |
|
|
"How would you adapt this for a small team?", |
|
|
"What are ethical considerations I should keep in mind?", |
|
|
"What's an underrated practice here and why?", |
|
|
] |
|
|
|
|
|
def make_answer(topic: str, opener: str) -> str: |
|
|
bullets: List[str] = [ |
|
|
f"Define your objective in {topic} (outcome, constraints, timeline).", |
|
|
f"Map key concepts and tools in {topic}; pick one stack and stick to it for 2-4 weeks.", |
|
|
f"Practice with a small, scoped project; iterate with feedback.", |
|
|
f"Track 1-3 metrics that reflect real progress; review weekly.", |
|
|
f"Document what worked, what didn't, and the next experiment.", |
|
|
] |
|
|
return ( |
|
|
f"Here's a pragmatic path for {topic} based on your request:\n" + |
|
|
"\n".join([f"- {b}" for b in bullets]) + |
|
|
"\n\nCommon pitfalls:\n- Starting too big; keep scope small.\n- Tool hopping; commit to one stack.\n- No feedback loop; schedule reviews.\n- Unclear metrics; define success upfront." |
|
|
) |
|
|
|
|
|
|
|
|
def make_examples(topic: str) -> str: |
|
|
ex: List[str] = [ |
|
|
f"Example 1 — 90-minute sprint: Learn one core concept in {topic} and apply it to a toy task.", |
|
|
f"Example 2 — Weekend project: Build a tiny demo that proves a single capability in {topic}.", |
|
|
f"Example 3 — Peer review: Share results, gather feedback, and improve one dimension.", |
|
|
] |
|
|
return "\n".join(ex) |
|
|
|
|
|
|
|
|
def make_template(topic: str) -> str: |
|
|
return ( |
|
|
f"Template — One-page plan for {topic}:\n" |
|
|
f"Goal: <1-2 sentences>\n" |
|
|
f"Scope: <what's in / out>\n" |
|
|
f"Resources: <3 links or docs>\n" |
|
|
f"Milestones (weekly): <targets>\n" |
|
|
f"Metrics: <quant/qual measures>\n" |
|
|
f"Risks & mitigations: <top 3>\n" |
|
|
) |
|
|
|
|
|
|
|
|
def rng_pick(r: random.Random, items: Sequence[str]) -> str: |
|
|
return items[r.randrange(0, len(items))] |
|
|
|
|
|
|
|
|
def make_dialogue(r: random.Random) -> Dict[str, Any]: |
|
|
topic = rng_pick(r, TOPICS) |
|
|
opener_tpl = rng_pick(r, USER_OPENERS) |
|
|
opener = opener_tpl.format(topic=topic) |
|
|
|
|
|
a1 = make_answer(topic, opener) |
|
|
|
|
|
u2 = rng_pick(r, FOLLOW_UPS) |
|
|
a2 = make_examples(topic) if "example" in u2.lower() else ( |
|
|
make_template(topic) if "template" in u2.lower() else |
|
|
"Here is a step-by-step variant focusing on small wins first, then scope up.\n1) Clarify outcome.\n2) Pick one method.\n3) Build tiny demo.\n4) Review metrics.\n5) Iterate." |
|
|
) |
|
|
|
|
|
if r.random() < 0.6: |
|
|
u3 = rng_pick(r, DEEPENERS) |
|
|
a3 = ( |
|
|
"Resilience tips:\n- Use checklists and pre-mortems.\n- Automate a single weak link each week.\n- Keep a rollback plan.\n- Prefer boring, well-documented tools.\n\nEthics: stay transparent, reduce bias, and respect privacy." |
|
|
) |
|
|
messages: List[Dict[str, str]] = [ |
|
|
{"content": opener, "role": "user"}, |
|
|
{"content": a1, "role": "assistant"}, |
|
|
{"content": u2, "role": "user"}, |
|
|
{"content": a2, "role": "assistant"}, |
|
|
{"content": u3, "role": "user"}, |
|
|
{"content": a3, "role": "assistant"}, |
|
|
] |
|
|
else: |
|
|
messages = [ |
|
|
{"content": opener, "role": "user"}, |
|
|
{"content": a1, "role": "assistant"}, |
|
|
{"content": rng_pick(r, FOLLOW_UPS), "role": "user"}, |
|
|
{"content": a2, "role": "assistant"}, |
|
|
] |
|
|
|
|
|
prompt = messages[0]["content"] |
|
|
prompt_id = hashlib.sha256(prompt.encode("utf-8")).hexdigest() |
|
|
return {"prompt": prompt, "messages": messages, "prompt_id": prompt_id} |
|
|
|
|
|
|
|
|
def generate(n: int, seed: int) -> List[Dict[str, Any]]: |
|
|
r = random.Random(seed) |
|
|
return [make_dialogue(r) for _ in range(n)] |
|
|
|
|
|
|
|
|
def main(argv: List[str]) -> int: |
|
|
parser = argparse.ArgumentParser() |
|
|
parser.add_argument("--out", type=str, default="sample_data/train_sft.jsonl") |
|
|
parser.add_argument("--n", type=int, default=500) |
|
|
parser.add_argument("--seed", type=int, default=42) |
|
|
args = parser.parse_args(argv) |
|
|
|
|
|
data = generate(args.n, args.seed) |
|
|
out_path = args.out |
|
|
os.makedirs(os.path.dirname(out_path), exist_ok=True) |
|
|
with open(out_path, "w", encoding="utf-8") as f: |
|
|
for row in data: |
|
|
f.write(json.dumps(row, ensure_ascii=False) + "\n") |
|
|
|
|
|
print(f"Wrote {len(data)} rows to {out_path}") |
|
|
return 0 |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
raise SystemExit(main(sys.argv[1:])) |
|
|
|
