codelion's picture
Curate a corpus: clean, calibrate, create, compose on a sample, plus a recipe and runner for your own machines
4f740eb verified
Raw History Blame Contribute Delete
9.27 kB
"""Create: rephrase good source documents into new training documents, or write new ones where sources are thin.
Rephrase styles are the five from Sutra (weights 30/25/20/15/10, after BeyondWeb), plus math and code styles.
A fidelity check flags rephrasings that introduce numbers or names the source does not contain, which matters when the
training data must respect a knowledge cutoff.
"""
from __future__ import annotations
import random
import re
from .llm import Endpoint, EndpointError
NO_PREAMBLE = ("Output ONLY the rewritten document. No preamble, no meta-commentary, no phrases like \"Here is\", "
"\"Let me\", \"Sure\" or \"Okay\". Do not add facts, numbers, names or dates that are not in the original.")
STYLES = {
"conversational": ("Teacher-student dialogue", 30, """Rewrite the content as a natural conversation between a teacher and a student.
The student asks curious, natural questions; the teacher explains clearly with concrete examples taken from the content.
Keep all key information and the same technical accuracy. Keep a similar length (within 20%).
Use the labels **Teacher**: and **Student**: and start directly with "**Teacher**:"."""),
"tutorial": ("Step-by-step tutorial", 25, """Rewrite the content as a clear step-by-step tutorial.
Start with "## Tutorial: <topic>", then an overview, prerequisites, numbered steps with what to do and how to check it,
a short practice exercise and key takeaways. Keep all key information accurate. Keep a similar length (within 20%)."""),
"qa": ("Questions and answers", 20, """Rewrite the content as 4 to 7 questions and thorough answers that together cover every key point,
moving from basic (what) to deeper (why, how) to applied. Start with "## Understanding <topic>: Questions and Answers".
Use **Q:** and **A:** labels. Keep a similar total length (within 20%)."""),
"dense_reference": ("Dense reference notes", 15, """Rewrite the content as compact reference notes at 70 to 85 percent of the original length.
Remove filler and repetition. Use headings and bullet points: core definition, key properties, relationships,
essential examples, formulas or procedures if any, pitfalls."""),
"pedagogical": ("Textbook section", 10, """Rewrite the content as a well-structured textbook section: learning objectives, the connection to what
the reader already knows, the core explanation, one worked example with step-by-step reasoning, a common question
with its answer, and a quick self-check question. Keep a similar length."""),
"worked_examples": ("Worked problems (math)", 0, """Rewrite the content as 2 to 4 worked problems that use exactly the ideas in it. For each: state the problem,
solve it step by step with the reasoning for each step, and check the answer. Use LaTeX for mathematics."""),
"code_walkthrough": ("Code walkthrough (code)", 0, """Rewrite the content as an explained code walkthrough: what the code does, then the code in a fenced block
(keep it correct and runnable; you may tidy names and add comments), then an explanation of the key lines and one
example of using it."""),
}
KIND_STYLES = {"math": ["worked_examples", "pedagogical", "qa", "tutorial"],
"code": ["code_walkthrough", "tutorial", "qa", "dense_reference"]}
def styles_for(kind: str) -> list[str]:
return KIND_STYLES.get(kind, ["conversational", "tutorial", "qa", "dense_reference", "pedagogical"])
def pick_style(kind: str, rng: random.Random, allowed: list[str] | None = None) -> str:
opts = [s for s in (allowed or styles_for(kind)) if s in STYLES]
weights = [max(STYLES[s][1], 10) for s in opts]
return rng.choices(opts, weights=weights, k=1)[0]
def chunks(text: str, max_chars: int = 6000) -> list[str]:
"""Split a long document at paragraph breaks so each rewrite sees a piece it can keep at a similar length."""
if len(text) <= max_chars:
return [text]
out, cur = [], ""
for para in re.split(r"\n\s*\n", text):
if cur and len(cur) + len(para) + 2 > max_chars:
out.append(cur)
cur = ""
cur = (cur + "\n\n" + para) if cur else para
while len(cur) > max_chars: # one giant paragraph
out.append(cur[:max_chars])
cur = cur[max_chars:]
if cur.strip():
out.append(cur)
return [c for c in out if len(c.split()) >= 80] or [text[:max_chars]]
def rephrase_prompt(text: str, style: str, language: str = "") -> str:
lang = f"\nWrite it in {language}." if language and language.lower() != "same" else ""
return f"""{STYLES[style][2]}{lang}
{NO_PREAMBLE}
Original content:
<<<
{text[:8000]}
>>>"""
LEAD = re.compile(r"^(here('s| is)|sure|okay|certainly|let me|i('ll| will))\b", re.I)
NUM = re.compile(r"\b\d[\d,.]*\b")
# a capitalized word that follows a lowercase word mid-sentence: names, places, products. Words at the start of a
# sentence, a line or a markdown label are skipped, so "Teacher:" or "Great question" never count.
HEADINGS = re.compile(r"^#+ .*$|\*\*[^*\n]{1,80}\*\*", re.M)
NAME = re.compile(r"\b[a-z][a-z'-]*[,;]? ([A-Z][a-zA-Z]{2,}(?: [A-Z][a-zA-Z]{2,})*)")
def fidelity(src: str, out: str, template: str = "") -> dict:
"""Share of numbers and mid-sentence names in the rewrite that never appear in the source. A rough proxy for
invented facts; the threshold is a choice, not a law."""
s_low = (src + " " + template).lower()
body = HEADINGS.sub(" ", out) # headings and bold labels the style asks for are not new facts
nums = {n.rstrip(".,") for n in NUM.findall(out)}
names = set(NAME.findall(body))
new_nums = [n for n in nums if n and n not in src]
new_names = [n for n in names if n.lower() not in s_low]
total = len(nums) + len(names)
share = (len(new_nums) + len(new_names)) / total if total else 0.0
return {"new_numbers": sorted(new_nums)[:8], "new_names": sorted(new_names)[:8], "new_share": round(share, 3),
"checked": total}
def rephrase(ep: Endpoint, text: str, style: str, language: str = "", max_new_share: float = 0.35) -> dict:
"""One rewrite. Returns {"text", "style", "fidelity", "ok", "why"}."""
def ok(s):
return len(s.split()) >= 60 and not LEAD.search(s.strip()[:40])
try:
out = ep.chat(rephrase_prompt(text, style, language), max_tokens=3000, temperature=0.7, retries=3, validate=ok)
except EndpointError as e:
return {"text": "", "style": style, "ok": False, "why": f"model call failed: {e}"[:200]}
fid = fidelity(text, out, STYLES[style][2])
new = len(fid["new_numbers"]) + len(fid["new_names"])
# invented numbers are the clearest sign of made-up content; one stray number or name is not a pattern
good = len(fid["new_numbers"]) < 3 and (new < 2 or fid["new_share"] <= max_new_share)
return {"text": out, "style": style, "fidelity": fid, "ok": good, "why": "" if good else "adds facts the source does not contain"}
# ---------------------------------------------------------------- new documents where sources are thin
DOC_TYPES = { # a subset of Sutra's content types that reads well as pretraining text
"concept_introduction": "an introduction to the concept: definition, intuition, why it matters, how it works, two examples, a common misconception, and how it connects to related ideas",
"reasoning_demonstration": "a worked demonstration of reasoning about a concrete problem in this area, thinking step by step and checking the result",
"worked_examples": "two or three worked examples of increasing difficulty, each solved step by step",
"common_misconceptions": "an explanation of the most common misconceptions about the concept, why people hold them, and the correct understanding",
"problem_set": "a short problem set of four problems with full solutions",
"case_study": "a realistic case study that applies the concept, with the decisions and their reasons",
"technical_documentation": "clear technical documentation of the concept or tool, with a usage example",
"historical_context": "the history of how the idea developed, who contributed what, and why it mattered",
}
def invent_prompt(topic: str, doc_type: str, level: int, request: str, language: str = "English") -> str:
return f"""You are writing high-quality pretraining text for a language model. Collection: "{request}".
Write {DOC_TYPES[doc_type]}.
Concept: {topic}
Difficulty: {level} on a 1 (introductory) to 10 (expert) scale. Language: {language}.
Write 350 to 700 words of dense, accurate prose with markdown headings where useful. Every sentence should teach
something. Do not invent statistics, quotes or citations. {NO_PREAMBLE.split('. Do not add')[0]}."""
def invent(ep: Endpoint, topic: str, doc_type: str, level: int, request: str, language: str = "English") -> dict:
def ok(s):
return len(s.split()) >= 150 and not LEAD.search(s.strip()[:40])
try:
out = ep.chat(invent_prompt(topic, doc_type, level, request, language), max_tokens=2500, temperature=0.8, retries=3, validate=ok)
except EndpointError as e:
return {"text": "", "ok": False, "why": f"model call failed: {e}"[:200]}
return {"text": out, "ok": True, "doc_type": doc_type, "topic": topic, "level": level}