推理模型从零构建 第3章：如何评估推理模型

import re

def get_last_boxed(text):
    """提取文本中最后一个 \\boxed{...} 的内容，支持嵌套大括号"""
    boxed_start_idx = text.rfind(r"\boxed")
    if boxed_start_idx == -1:
        return None

    current_idx = boxed_start_idx + len(r"\boxed")
    # 跳过空白，找到开始的 {
    while current_idx < len(text) and text[current_idx].isspace():
        current_idx += 1
    if current_idx >= len(text) or text[current_idx] != "{":
        return None

    # 用计数器处理嵌套大括号
    current_idx += 1
    brace_depth = 1
    content_start_idx = current_idx
    while current_idx < len(text) and brace_depth > 0:
        if text[current_idx] == "{":
            brace_depth += 1
        elif text[current_idx] == "}":
            brace_depth -= 1
        current_idx += 1

    if brace_depth != 0:
        return None
    return text[content_start_idx:current_idx - 1]


RE_NUMBER = re.compile(r"-?(?:\d+/\d+|\d+(?:\.\d+)?(?:[eE][+-]?\d+)?)")

def extract_final_candidate(text, fallback="number_then_full"):
    """优先提取 \\boxed 内容；否则 fallback 到最后一个数字"""
    result = ""
    if text:
        boxed = get_last_boxed(text.strip())
        if boxed:
            result = boxed.strip().strip("$ ")
        elif fallback in ("number_then_full", "number_only"):
            m = RE_NUMBER.findall(text)
            if m:
                result = m[-1]  # 取最后一个数字
            elif fallback == "number_then_full":
                result = text   # 实在没有就返回全文
    return result


# 测试
model_answer = r"... \boxed{\dfrac{14}{3}}"
print(extract_final_candidate(model_answer))  # \dfrac{14}{3}

LATEX_FIXES = [  # LaTeX 格式替换规则
    (r"\\left\s*", ""),          # 去掉 \left
    (r"\\right\s*", ""),         # 去掉 \right
    (r"\\,|\\!|\\;|\\:", ""),    # 去掉间距命令
    (r"\\cdot", "*"),            # \cdot → *
    (r"\\dfrac", r"\\frac"),     # \dfrac → \frac（统一）
    (r"\\tfrac", r"\\frac"),     # \tfrac → \frac
]

def normalize_text(text):
    if not text:
        return ""
    # 去掉 chat 特殊 token
    text = re.sub(r"<\|[^>]+?\|>", "", text).strip()

    # 去掉 \text{...} 包裹
    match = re.match(r"^\\text\{(?P.+?)\}$", text)
    if match:
        text = match.group("x")

    # 去掉数学分隔符 \( \) \[ \]
    text = re.sub(r"\\\(|\\\)|\\\[|\\\]", "", text)

    # 应用 LaTeX 替换规则
    for pat, rep in LATEX_FIXES:
        text = re.sub(pat, rep, text)

    # \frac{a}{b} → (a)/(b)
    text = re.sub(
        r"\\frac\s*\{([^{}]+)\}\s*\{([^{}]+)\}",
        lambda m: f"({m.group(1)})/({m.group(2)})", text
    )
    # \sqrt{x} → sqrt(x)
    text = re.sub(
        r"\\sqrt\s*\{([^}]*)\}",
        lambda m: f"sqrt({m.group(1)})", text
    )
    # ^ → **（指数）
    text = text.replace("^", "**")
    # 去掉千分位逗号：1,234 → 1234
    text = re.sub(r"(?<=\d),(?=\d\d\d(\D|$))", "", text)

    return text.replace("{", "").replace("}", "").strip().lower()


# 示例
print(normalize_text(r"\dfrac{14}{3}"))   # (14)/(3)
print(normalize_text(r"$\sqrt{8}/2$"))    # sqrt(8)/2

from sympy.parsing import sympy_parser as spp
from sympy import simplify

def sympy_parser(expr):
    """将字符串解析为 SymPy 符号表达式"""
    if expr is None or len(expr) > 2000:
        return None  # 防止 badly-trained 模型的超长垃圾输出
    try:
        return spp.parse_expr(
            expr,
            transformations=(
                *spp.standard_transformations,
                spp.implicit_multiplication_application,  # "2x" → 2*x
            ),
            evaluate=True,  # 2+3 → 5
        )
    except (SyntaxError, TypeError, ValueError):
        return None

def equality_check(expr_gtruth, expr_pred):
    """判断两个表达式是否数学等价"""
    if expr_gtruth == expr_pred:  # 快速路径：字符串相同
        return True
    gtruth = sympy_parser(expr_gtruth)
    pred = sympy_parser(expr_pred)
    if gtruth is not None and pred is not None:
        try:
            return simplify(gtruth - pred) == 0  # 差为0则等价
        except (TypeError,):
            pass
    return False

# 测试
print(equality_check("(14)/(3)", "14/3"))      # True
print(equality_check("0.5", "(1)/(2)"))         # True
print(equality_check("(14)/(3)", "(15)/(3)"))   # False

def grade_answer(pred_text, gt_text):
    """完整评分：标准化 → 拆分 → 逐部分等价判定"""
    if pred_text is None or gt_text is None:
        return False

    gt_parts = split_into_parts(normalize_text(gt_text))
    pred_parts = split_into_parts(normalize_text(pred_text))

    # 部分数量必须一致
    if not gt_parts or not pred_parts or len(gt_parts) != len(pred_parts):
        return False

    # 每个部分都必须数学等价
    return all(
        equality_check(gt, pred)
        for gt, pred in zip(gt_parts, pred_parts)
    )

# 元组答案测试
grade_answer(r"(14/3, 2/3)", "(14/3, 4/6)")  # True（4/6 = 2/3）
grade_answer("(2, 1)", "(1, 2)")               # False（顺序不同）

def render_prompt(prompt):
    return (
        "You are a helpful math assistant.\n"
        "Answer the question and write the final result on a new line as:\n"
        "\\boxed{ANSWER}\n\n"
        f"Question:\n{prompt}\n\nAnswer:"
    )

def evaluate_math500_stream(model, tokenizer, device, math_data,
                             max_new_tokens=2048):
    num_correct = 0
    for i, row in enumerate(math_data, start=1):
        # 1. 构造 prompt
        prompt = render_prompt(row["problem"])
        # 2. 生成回答
        gen_text = generate_text_stream_concat(
            model, tokenizer, prompt, device,
            max_new_tokens=max_new_tokens
        )
        # 3. 提取答案
        extracted = extract_final_candidate(gen_text)
        # 4. 评分
        is_correct = grade_answer(extracted, row["answer"])
        num_correct += int(is_correct)

    acc = num_correct / len(math_data)
    print(f"Accuracy: {acc*100:.1f}% ({num_correct}/{len(math_data)})")
    return acc