reference/code/TAD/prepare/dream_gen_math_traj.py

"""

Generate trajectories using the Dream diffusion language model on GSM8k / competition_math datasets.

Core logic:
  - Each diffusion step decodes only 1 token (steps = max_new_tokens)
  - If the model's final answer is correct, collect the complete trajectory (token sequence at each step)
  - Trajectories are written in JSONL format, consistent with trajectory_generation.py

Usage example:
  python dream_gen_math_traj.py \
      --model_name Dream-org/Dream-v0-Instruct-7B \
      --output_path outputs/gsm8k_dream_traj.jsonl \
      --dataset_name gsm8k \
      --dataset_config main \
      --dataset_split train \
      --max_new_tokens 256 \
      --num_samples 1 \
      --temperature 0.2 \
      --top_p 0.95 \
      --limit 100
"""

import argparse
import json
import os
import random
import re
import sys

import numpy as np
import torch
from datasets import load_dataset
from tqdm import tqdm
from transformers import AutoModel, AutoTokenizer

# -- Reuse Parser / is_equiv --
# Prefer importing from sibling eval directory; fallback to inline minimal version
_DLLM_EVAL_PATH = os.path.join(
    os.path.dirname(__file__),
    "..",
    "TAD",
    "eval",
)
if os.path.isdir(_DLLM_EVAL_PATH):
    sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
    from dllm_inference_acceleration_main.eval.parsers import Parser, is_equiv  # type: ignore
else:
    import re as _re

    def _remove_boxed(s):
        if "\\boxed " in s:
            return s[len("\\boxed "):]
        left = "\\boxed{"
        try:
            assert s[: len(left)] == left and s[-1] == "}"
            return s[len(left): -1]
        except Exception:
            return s

    def _last_boxed_only_string(string):
        idx = string.rfind("\\boxed")
        if "\\boxed " in string:
            return "\\boxed " + string.split("\\boxed ")[-1].split("$")[0]
        if idx < 0:
            idx = string.rfind("\\fbox")
            if idx < 0:
                return string
        i, right_brace_idx, num_open = idx, None, 0
        while i < len(string):
            if string[i] == "{":
                num_open += 1
            if string[i] == "}":
                num_open -= 1
                if num_open == 0:
                    right_brace_idx = i
                    break
            i += 1
        return string[idx: right_brace_idx + 1] if right_brace_idx is not None else None

    class Parser:
        @classmethod
        def extract_answer_gsm8k(cls, text):
            try:
                m = _re.search(r"####\s*\$?([\d,]+(?:\.\d+)?)", text)
                if m:
                    return float(m.group(1).replace(",", ""))
            except Exception:
                pass
            return None

        @classmethod
        def extract_answer_boxed(cls, text):
            try:
                return _remove_boxed(_last_boxed_only_string(text))
            except Exception:
                return text

    def _strip_string(s):
        s = s.replace("\n", "").replace("\\!", "").replace("\\\\", "\\")
        s = s.replace("tfrac", "frac").replace("dfrac", "frac")
        s = s.replace("\\left", "").replace("\\right", "")
        s = s.replace("^{\\circ}", "").replace("^\\circ", "")
        s = s.replace("\\$", "").replace("\\%", "").replace("\%", "")
        s = s.replace(" .", " 0.").replace("{.", "{0.")
        if not s:
            return s
        if s[0] == ".":
            s = "0" + s
        if len(s.split("=")) == 2 and len(s.split("=")[0]) <= 2:
            s = s.split("=")[1]
        s = s.replace(" ", "")
        return s

    def is_equiv(str1, str2, verbose=False):
        if isinstance(str1, float) or isinstance(str2, float):
            try:
                return abs(float(str1) - float(str2)) < 1e-6
            except Exception:
                return False
        if str1 is None and str2 is None:
            return True
        if str1 is None or str2 is None:
            return False
        try:
            return _strip_string(str(str1)) == _strip_string(str(str2))
        except Exception:
            return str1 == str2


# -- Helper functions --

def select_device():
    if torch.cuda.is_available():
        return "cuda"
    mps_backend = getattr(torch.backends, "mps", None)
    if mps_backend is not None and mps_backend.is_available():
        return "mps"
    return "cpu"


def build_prompt(tokenizer, question: str) -> str:
    messages = [{"role": "user", "content": question}]
    return tokenizer.apply_chat_template(
        messages, add_generation_prompt=True, tokenize=False
    )


def extract_last_number(text: str):
    nums = re.findall(r"-?\d{1,3}(?:,\d{3})+(?:\.\d+)?|-?\d+(?:\.\d+)?", text)
    return nums[-1] if nums else None


def extract_pred(text: str):
    pred = Parser.extract_answer_boxed(text)
    if pred is None:
        pred = Parser.extract_answer_gsm8k(text)
    if pred is None:
        return extract_last_number(text)
    pred_num = extract_last_number(pred)
    return pred_num if pred_num is not None else pred


def extract_example(example, question_key: str, answer_key: str):
    """Extract question, answer, and dataset type from a dataset sample."""
    if question_key and answer_key:
        return example[question_key], example[answer_key], "custom"
    if "question" in example and "answer" in example:
        return example["question"], example["answer"], "gsm8k"
    if "problem" in example and "solution" in example:
        return example["problem"], example["solution"], "math"
    raise ValueError(
        "Cannot infer question/answer fields, please specify via --question_key and --answer_key."
    )


# -- Trajectory generation (core) --

@torch.no_grad()
def generate_dream_trajectory(
    model,
    tokenizer,
    input_ids: torch.Tensor,
    attention_mask: torch.Tensor,
    max_new_tokens: int,
    temperature: float,
    top_p: float,
    alg: str,
    alg_temp: float,
    block_length: int=32,
):
    """
    Call Dream model's diffusion_generate interface with steps = max_new_tokens,
    i.e., each diffusion step decodes only 1 token.

    Returns:
        output_text  : str, final generated text (truncated after eos)
        traj_dict    : dict, keys are "step0", "step1", ..., values are token id lists for the generation region
                       (consistent with trajectory_generation.py format)
    """
    output = model.diffusion_generate(
        input_ids,
        attention_mask=attention_mask,
        max_new_tokens=max_new_tokens,
        output_history=True,          # Collect intermediate states
        return_dict_in_generate=True,
        steps=max_new_tokens,         # Each step generates only 1 token
        temperature=temperature,
        top_p=top_p,
        alg=alg,
        alg_temp=alg_temp,
        block_length=block_length,
    )

    # Decode final output
    prompt_len = input_ids.shape[1]
    final_ids = output.sequences[0]          # shape: (prompt_len + gen_len,)
    gen_ids = final_ids[prompt_len:].tolist()
    output_text = tokenizer.decode(gen_ids, skip_special_tokens=False)
    # Truncate at first eos
    eos_token = tokenizer.eos_token
    if eos_token and eos_token in output_text:
        output_text = output_text.split(eos_token)[0]

    # Organize trajectory: output.history is a list, each element shape=(batch, full_seq_len)
    # Only keep the generation region (after prompt), consistent with trajectory_generation.py
    traj_dict = {}
    history = output.history  # List[Tensor(batch, seq_len)]
    for step_idx, h in enumerate(history):
        # h: (batch_size, prompt_len + gen_len), take sample 0's generation region
        gen_part = h[0, prompt_len:].tolist()
        traj_dict[f"step{step_idx}"] = gen_part

    return output_text, traj_dict


# -- Main --

def main():
    parser = argparse.ArgumentParser(
        description="Generate math reasoning trajectories using Dream diffusion language model"
    )
    parser.add_argument(
        "--model_name",
        type=str,
        default="",
        help="Dream model path or HuggingFace Hub ID",
    )
    parser.add_argument("--output_path", type=str, required=True, help="Output JSONL file path")
    parser.add_argument("--max_new_tokens", type=int, default=256, help="Max generated tokens (also the number of diffusion steps)")
    parser.add_argument("--block_length", type=int, default=32, help="Block length")
    parser.add_argument("--num_samples", type=int, default=1, help="Number of samples per problem")
    parser.add_argument("--temperature", type=float, default=0.2, help="Sampling temperature")
    parser.add_argument("--top_p", type=float, default=0.95, help="Top-p for nucleus sampling")
    parser.add_argument("--alg", type=str, default="entropy", help="Dream decoding algorithm (entropy / origin etc.)")
    parser.add_argument("--alg_temp", type=float, default=0.0, help="Dream decoding algorithm temperature")
    parser.add_argument("--seed", type=int, default=42, help="Random seed")
    parser.add_argument("--limit", type=int, default=0, help="Max samples to process, 0 means all")
    parser.add_argument(
        "--dataset_name",
        type=str,
        default="gsm8k",
        help="Dataset name",
    )
    parser.add_argument("--dataset_config", type=str, default=None, help="Dataset config")
    parser.add_argument("--dataset_split", type=str, default="train", help="Dataset split")
    parser.add_argument("--question_key", type=str, default="", help="Custom question field name")
    parser.add_argument("--answer_key", type=str, default="", help="Custom answer field name")
    args = parser.parse_args()

    # -- Random seed --
    random.seed(args.seed)
    np.random.seed(args.seed)
    torch.manual_seed(args.seed)

    # -- Device and model --
    device = select_device()
    dtype_map = {"cuda": torch.bfloat16, "mps": torch.float16, "cpu": torch.float32}
    dtype = dtype_map[device]
    print(f"[INFO] Device: {device}  dtype={dtype}")

    print(f"[INFO] Loading model: {args.model_name}")
    model = AutoModel.from_pretrained(
        args.model_name, torch_dtype=dtype, trust_remote_code=True
    ).to(device).eval()

    tokenizer = AutoTokenizer.from_pretrained(
        args.model_name, trust_remote_code=True, padding_side="left"
    )

    # -- Dataset --
    print(f"[INFO] Loading dataset: {args.dataset_name}/{args.dataset_config} split={args.dataset_split}")
    if args.dataset_config is not None:
      ds = load_dataset(args.dataset_name, args.dataset_config, split=args.dataset_split)
    else:
      ds = load_dataset(args.dataset_name, split=args.dataset_split)

    # -- Output file --
    out_dir = os.path.dirname(args.output_path)
    if out_dir:
        os.makedirs(out_dir, exist_ok=True)
    out_f = open(args.output_path, "w", encoding="utf-8")

    total = len(ds) if args.limit <= 0 else min(args.limit, len(ds))
    solved = 0

    for idx in tqdm(range(total), desc=f"dataset={args.dataset_name}"):
        raw_q, raw_a, kind = extract_example(ds[idx], args.question_key, args.answer_key)

        # Extract ground-truth answer
        if kind == "gsm8k":
            answer_gt = Parser.extract_answer_gsm8k(raw_a)
        else:
            # math / custom: prefer boxed, otherwise raw text
            answer_gt = Parser.extract_answer_boxed(raw_a)

        # Build question text: provide reference answer to the model, let it re-solve in its own way
        question = (
            f"Question: {raw_q}\n"
            f"Reference Answer: {raw_a}\n"
            f"After understanding the reference answer, please try to solve this problem "
            f"using your own approach below and output a detailed solution process:"
        )

        # Build prompt and tokenize (batch_size=1, Dream does not support batch trajectory collection)
        prompt_str = build_prompt(tokenizer, question)
        encoded = tokenizer(
            [prompt_str],
            add_special_tokens=False,
            padding=True,
            return_tensors="pt",
        )
        input_ids = encoded["input_ids"].to(device)
        attention_mask = encoded["attention_mask"].to(device)

        counted = False  # Only count solved once per problem

        for sample_id in range(args.num_samples):
            output_text, traj_dict = generate_dream_trajectory(
                model=model,
                tokenizer=tokenizer,
                input_ids=input_ids,
                attention_mask=attention_mask,
                max_new_tokens=args.max_new_tokens,
                temperature=args.temperature,
                top_p=args.top_p,
                alg=args.alg,
                alg_temp=args.alg_temp,
                block_length=args.block_length,
            )
            print(f"output_text:{output_text}")

            pred = extract_pred(output_text)
            print(f"[idx={idx} sample={sample_id}] pred={pred}  gt={answer_gt}")

            if is_equiv(pred, answer_gt):
                if not counted:
                    solved += 1
                    counted = True

                # Collect trajectory record (format consistent with trajectory_generation.py)
                record = {
                    "dataset_name": args.dataset_name,
                    "dataset_config": args.dataset_config,
                    "dataset_split": args.dataset_split,
                    "index": idx,
                    "sample_id": sample_id,
                    "prompt": prompt_str,
                    "question": question,
                    "answer_gt": answer_gt,
                    "pred": pred,
                    "prompt_len": int(input_ids.shape[1]),
                    "trajectory": traj_dict,
                    "output_text": output_text,
                }
                out_f.write(json.dumps(record, ensure_ascii=False) + "\n")
                out_f.flush()

    # -- Summary --
    accuracy = solved / total if total > 0 else 0.0
    print(f"Accuracy_any_success: {solved}/{total} = {accuracy * 100:.2f}%")
    out_f.write(
        json.dumps(
            {
                "summary": "any_success",
                "solved": solved,
                "total": total,
                "accuracy": accuracy,
            },
            ensure_ascii=False,
        )
        + "\n"
    )
    out_f.close()


if __name__ == "__main__":
    main()