推理模型：思维链与慢思考范式

问题：一个书架有 3 层，每层放 12 本书，又搬走了 5 本，还剩多少本？

【普通提示】直接问：请给出答案。
模型输出：31 本

【思维链提示】请一步步思考后再给出最终答案。
模型输出：
1. 书架总共有 3 层，每层 12 本，所以总数是 3 × 12 = 36 本。
2. 搬走了 5 本，所以剩下 36 - 5 = 31 本。
最终答案：31 本

from openai import OpenAI

client = OpenAI()

question = "一个班有 30 名学生，男生比女生多 6 人，男生有多少人？"

# 关键在于结尾的触发句，把模型从"直接作答"推向"逐步推理"
prompt = f"{question}\n请一步步思考，最后用『答案：』开头给出结论。"

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": prompt}],
    temperature=0,
)
print(resp.choices[0].message.content)

few_shot = """问题：停车场有 15 辆车，开走了 6 辆又来了 4 辆，现在有几辆？
推理：先算开走后剩下 15 - 6 = 9 辆，再加上新来的 4 辆，得到 9 + 4 = 13 辆。
答案：13

问题：一盒鸡蛋有 12 个，打碎了 3 个，又买了 2 盒，一共有几个完好的鸡蛋？
推理：原来完好 12 - 3 = 9 个，买 2 盒即 2 × 12 = 24 个，合计 9 + 24 = 33 个。
答案：33
"""

new_question = "问题：图书馆有 200 本书，借出 45 本，归还 12 本，现在有几本？"

prompt = few_shot + new_question + "\n推理："

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": prompt}],
    temperature=0,
)
print(resp.choices[0].message.content)

from collections import Counter
import re

def extract_answer(text: str) -> str:
    # 约定模型用「答案：xxx」收尾，这里抓取最后一个答案标记
    matches = re.findall(r"答案[:：]\s*(.+)", text)
    return matches[-1].strip() if matches else text.strip().splitlines()[-1]

def self_consistency(question: str, n: int = 5, temperature: float = 0.8) -> str:
    prompt = f"{question}\n请一步步思考，最后用『答案：』开头给出结论。"
    answers = []
    for _ in range(n):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,  # 必须 > 0 才能采样出不同路径
        )
        answers.append(extract_answer(resp.choices[0].message.content))
    # 多数投票，返回出现次数最多的答案
    winner, count = Counter(answers).most_common(1)[0]
    print(f"各路径答案分布：{Counter(answers)}，投票胜出：{winner}（{count}/{n}）")
    return winner

print(self_consistency("37 × 24 等于多少？", n=5))

import subprocess, tempfile, os

def score_code(code: str, test_snippet: str) -> int:
    # 打分器：把候选代码和测试拼起来跑，通过得 1 分，否则 0 分（充当 ORM）
    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
        f.write(code + "\n" + test_snippet)
        path = f.name
    try:
        r = subprocess.run(["python", path], capture_output=True, timeout=10)
        return 1 if r.returncode == 0 else 0
    finally:
        os.remove(path)

def best_of_n(task: str, test_snippet: str, n: int = 8) -> str:
    candidates = []
    for _ in range(n):
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": task}],
            temperature=0.9,
        )
        code = resp.choices[0].message.content
        candidates.append((score_code(code, test_snippet), code))
    # 按分数降序，取最优候选
    candidates.sort(key=lambda x: x[0], reverse=True)
    best_score, best_code = candidates[0]
    print(f"最佳候选得分：{best_score}/1")
    return best_code

task = "写一个 Python 函数 is_prime(n) 判断素数，只返回函数定义。"
test = "assert is_prime(7) and not is_prime(8) and not is_prime(1)"
print(best_of_n(task, test, n=8))

def propose_steps(state: str, k: int = 3) -> list[str]:
    # 让模型基于当前推理状态，提出 k 个可能的下一步
    prompt = f"当前推理进展：\n{state}\n请提出 {k} 个不同的下一步思路，每行一个。"
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
    )
    return [s for s in resp.choices[0].message.content.splitlines() if s.strip()][:k]

def evaluate(state: str) -> float:
    # 让模型给当前路径打一个 0 到 1 的"有希望程度"分（充当启发式评估器）
    prompt = f"评估这条推理路径有多可能导向正确答案，只输出 0 到 1 的小数：\n{state}"
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    try:
        return float(resp.choices[0].message.content.strip().split()[0])
    except ValueError:
        return 0.0

def tree_of_thoughts(question: str, depth: int = 3, beam: int = 2) -> str:
    frontier = [question]  # 当前保留的推理路径集合
    for _ in range(depth):
        candidates = []
        for state in frontier:
            for step in propose_steps(state):
                new_state = state + "\n-> " + step
                candidates.append((evaluate(new_state), new_state))
        # 按评估分排序，只保留 beam 条最有希望的路径继续展开
        candidates.sort(key=lambda x: x[0], reverse=True)
        frontier = [s for _, s in candidates[:beam]]
    return frontier[0]

print(tree_of_thoughts("用 4, 6, 8, 3 通过加减乘除凑出 24。"))

resp = client.chat.completions.create(
    model="o3-mini",
    messages=[{"role": "user", "content": "证明：任意三个连续整数之积能被 6 整除。"}],
    reasoning_effort="high",   # low / medium / high，控制思考深度
)

msg = resp.choices[0].message
usage = resp.usage
print("最终回答：", msg.content)
# 推理 token 单独统计，属于输出计费的一部分
print("推理 token：", usage.completion_tokens_details.reasoning_tokens)
print("可见输出 token：", usage.completion_tokens)

def route_effort(question: str) -> str:
    # 用轻量启发式（长度、关键词）决定给多少思考预算
    hard_signals = ["证明", "为什么", "推导", "规划", "优化", "复杂度", "调试"]
    if any(k in question for k in hard_signals) or len(question) > 120:
        return "high"
    if len(question) > 40:
        return "medium"
    return "low"

def ask(question: str) -> str:
    effort = route_effort(question)
    resp = client.chat.completions.create(
        model="o3-mini",
        messages=[{"role": "user", "content": question}],
        reasoning_effort=effort,
    )
    print(f"[effort={effort}]")
    return resp.choices[0].message.content

print(ask("今天星期几用一句话回答"))         # 走 low
print(ask("请推导快速排序的平均时间复杂度"))   # 走 high

curl https://api.openai.com/v1/chat/completions -H "Authorization: Bearer ${OPENAI_API_KEY}" -H "Content-Type: application/json" -d '{"model":"o3-mini","reasoning_effort":"medium","messages":[{"role":"user","content":"9.11 和 9.9 哪个大？"}]}'

import json

def build_distill_dataset(questions: list[str], out_path: str):
    records = []
    for q in questions:
        resp = client.chat.completions.create(
            model="deepseek-reasoner",   # 教师模型
            messages=[{"role": "user", "content": q}],
        )
        msg = resp.choices[0].message
        reasoning = getattr(msg, "reasoning_content", "")  # 教师的思维链
        answer = msg.content

        # 只保留答对的样本（拒绝采样）：这里用一个占位校验函数
        if not verify(q, answer):
            continue

        # 学生模型的训练样本：把思维链包进 <think> 标签，教它模仿"先想再答"
        target = f"<think>\n{reasoning}\n</think>\n\n{answer}"
        records.append({"prompt": q, "completion": target})

    with open(out_path, "w", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print(f"共产出 {len(records)} 条蒸馏样本 -> {out_path}")

def verify(question: str, answer: str) -> bool:
    # 实战中替换成真校验：数学题验算、代码题跑测试、或用另一模型判分
    return "答案" in answer

# build_distill_dataset(["计算 128 的所有正因子之和", "..."], "distill.jsonl")