Test-Time Compute 深度解析：让模型「思考更久」的工程实践

# 传统范式：提升能力 = 更大模型 + 更多训练数据
performance = f(model_size, training_compute)

# TTC 范式：提升能力 = 推理阶段投入更多计算
performance = f(model_size, training_compute, inference_compute)

import openai

def solve_with_cot(problem: str, client: openai.OpenAI) -> str:
    """使用 Chain-of-Thought 提示解决复杂问题"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "你是一个严谨的推理专家。解决问题时，你必须：\n"
                    "1. 明确列出已知条件\n"
                    "2. 分步推导，每步说明依据\n"
                    "3. 检验最终答案的合理性"
                )
            },
            {
                "role": "user",
                "content": f"请一步一步推理解决以下问题：\n\n{problem}"
            }
        ],
        temperature=0.0
    )
    return response.choices[0].message.content

import asyncio
from collections import Counter
from typing import List

async def self_consistency_solve(
    problem: str,
    client: openai.AsyncOpenAI,
    num_samples: int = 5,
    temperature: float = 0.7
) -> dict:
    """Self-Consistency: 多路径采样 + 多数投票"""
    
    async def sample_one() -> str:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "逐步推理，最终答案用 \\boxed{} 包裹。"},
                {"role": "user", "content": problem}
            ],
            temperature=temperature
        )
        return response.choices[0].message.content
    
    # 并行采样 N 条独立推理路径
    paths = await asyncio.gather(*[sample_one() for _ in range(num_samples)])
    
    # 提取最终答案并投票
    answers = [extract_boxed_answer(path) for path in paths]
    vote_counts = Counter(answers)
    best_answer = vote_counts.most_common(1)[0][0]
    confidence = vote_counts[best_answer] / num_samples
    
    return {
        "answer": best_answer,
        "confidence": confidence,
        "num_paths": num_samples,
        "vote_distribution": dict(vote_counts)
    }


def extract_boxed_answer(text: str) -> str:
    """从 LaTeX \\boxed{} 格式中提取答案"""
    import re
    match = re.search(r'\\boxed\{(.+?)\}', text)
    return match.group(1).strip() if match else text.strip().split('\n')[-1]

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ThoughtNode:
    """推理树节点"""
    thought: str
    score: float = 0.0
    children: list = field(default_factory=list)
    parent: Optional['ThoughtNode'] = None
    depth: int = 0

class TreeOfThought:
    """Tree-of-Thought 推理搜索引擎"""
    
    def __init__(self, client: openai.OpenAI, max_depth: int = 3, branch_factor: int = 3):
        self.client = client
        self.max_depth = max_depth
        self.branch_factor = branch_factor
    
    def solve(self, problem: str) -> ThoughtNode:
        root = ThoughtNode(thought=f"问题：{problem}")
        self._expand(root, problem)
        return self._best_leaf(root)
    
    def _expand(self, node: ThoughtNode, problem: str):
        """递归展开推理树"""
        if node.depth >= self.max_depth:
            return
        
        # 生成多个候选思路
        candidates = self._generate_thoughts(problem, node)
        
        # 评估每个思路的质量
        for thought_text in candidates:
            child = ThoughtNode(
                thought=thought_text,
                parent=node,
                depth=node.depth + 1
            )
            child.score = self._evaluate_thought(problem, child)
            node.children.append(child)
        
        # 只展开得分最高的分支（束搜索）
        node.children.sort(key=lambda x: x.score, reverse=True)
        for child in node.children[:2]:  # beam width = 2
            self._expand(child, problem)
    
    def _generate_thoughts(self, problem: str, node: ThoughtNode) -> list:
        """生成当前节点的候选下一步思路"""
        path = self._get_path(node)
        prompt = (
            f"问题：{problem}\n\n"
            f"已有推理步骤：\n{path}\n\n"
            f"请提出 {self.branch_factor} 种不同的下一步推理方向，"
            f"每种用 [思路N] 标记。"
        )
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.8
        )
        return self._parse_thoughts(response.choices[0].message.content)
    
    def _evaluate_thought(self, problem: str, node: ThoughtNode) -> float:
        """使用 LLM 作为评估器打分 (0-1)"""
        path = self._get_path(node)
        prompt = (
            f"评估以下推理路径的质量（0-1 分）：\n"
            f"问题：{problem}\n推理：{path}\n"
            f"评分标准：逻辑连贯性、是否朝正确方向推进、是否有明显错误。\n"
            f"只输出一个 0-1 之间的数字。"
        )
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0
        )
        try:
            return float(response.choices[0].message.content.strip())
        except ValueError:
            return 0.5
    
    def _get_path(self, node: ThoughtNode) -> str:
        """回溯获取从根到当前节点的完整路径"""
        path = []
        current = node
        while current.parent:
            path.append(current.thought)
            current = current.parent
        return "\n→ ".join(reversed(path)) if path else "(起始)"
    
    def _best_leaf(self, root: ThoughtNode) -> ThoughtNode:
        """找到得分最高的叶子节点"""
        best = root
        stack = [root]
        while stack:
            node = stack.pop()
            if not node.children and node.score > best.score:
                best = node
            stack.extend(node.children)
        return best
    
    def _parse_thoughts(self, text: str) -> list:
        import re
        thoughts = re.findall(r'\[思路\d+\]\s*(.+?)(?=\[思路|$)', text, re.DOTALL)
        return [t.strip() for t in thoughts] if thoughts else [text]

# DeepSeek R1 的训练哲学（伪代码）
class DeepSeekR1Training:
    """
    R1 训练的核心创新：
    1. 不使用昂贵的 Reward Model（不像 RLHF）
    2. 使用 GRPO (Group Relative Policy Optimization)
    3. 推理过程完全可见（<think>...</think> 标签）
    """
    
    def grpo_step(self, problem, model):
        # 采样一组候选答案
        group = [model.generate(problem) for _ in range(16)]
        
        # 使用规则验证器（而非 Reward Model）打分
        scores = [self.rule_verifier(problem, answer) for answer in group]
        
        # 组内相对排序作为奖励信号
        baseline = sum(scores) / len(scores)
        advantages = [s - baseline for s in scores]
        
        # 策略梯度更新
        model.update(group, advantages)
    
    def rule_verifier(self, problem, answer):
        """基于规则的验证器（数学：对比标准答案；代码：运行测试）"""
        if problem.type == "math":
            return 1.0 if answer.final == problem.ground_truth else 0.0
        elif problem.type == "code":
            return run_tests(answer.code, problem.test_cases)

import OpenAI from 'openai';

interface RefinementConfig {
  maxIterations: number;
  qualityThreshold: number;
  model: string;
  critiqueModel: string;
}

interface RefinementResult {
  finalOutput: string;
  iterations: number;
  scores: number[];
  totalTokens: number;
}

async function iterativeRefinement(
  task: string,
  config: RefinementConfig,
  client: OpenAI
): Promise<RefinementResult> {
  const { maxIterations, qualityThreshold, model, critiqueModel } = config;
  let currentOutput = '';
  const scores: number[] = [];
  let totalTokens = 0;

  // 初始生成
  const initial = await client.chat.completions.create({
    model,
    messages: [
      { role: 'system', content: '你是一个专业的问题解决者。' },
      { role: 'user', content: task }
    ],
    temperature: 0.7
  });
  currentOutput = initial.choices[0].message.content || '';
  totalTokens += initial.usage?.total_tokens || 0;

  for (let i = 0; i < maxIterations; i++) {
    // 评估当前输出质量
    const critique = await client.chat.completions.create({
      model: critiqueModel,
      messages: [
        {
          role: 'system',
          content: `评估以下输出的质量。返回 JSON 格式：
            {"score": 0.0-1.0, "issues": ["问题1", ...], "suggestions": ["建议1", ...]}`
        },
        {
          role: 'user',
          content: `任务：${task}\n\n输出：${currentOutput}`
        }
      ],
      temperature: 0.0,
      response_format: { type: 'json_object' }
    });
    totalTokens += critique.usage?.total_tokens || 0;

    const evaluation = JSON.parse(critique.choices[0].message.content || '{}');
    scores.push(evaluation.score || 0);

    // 达到质量阈值则提前退出
    if (evaluation.score >= qualityThreshold) {
      return { finalOutput: currentOutput, iterations: i + 1, scores, totalTokens };
    }

    // 基于反馈进行优化
    const refinement = await client.chat.completions.create({
      model,
      messages: [
        {
          role: 'system',
          content: '根据以下反馈改进你的输出。保留好的部分，修正问题。'
        },
        { role: 'user', content: `原始任务：${task}` },
        { role: 'assistant', content: currentOutput },
        {
          role: 'user',
          content: `改进反馈：\n问题：${evaluation.issues?.join(', ')}\n建议：${evaluation.suggestions?.join(', ')}\n\n请输出改进后的完整版本。`
        }
      ],
      temperature: 0.5
    });
    totalTokens += refinement.usage?.total_tokens || 0;
    currentOutput = refinement.choices[0].message.content || currentOutput;
  }

  return { finalOutput: currentOutput, iterations: maxIterations, scores, totalTokens };
}

// 使用示例
const result = await iterativeRefinement(
  '用 Python 实现一个线程安全的 LRU Cache，支持 TTL 过期',
  {
    maxIterations: 3,
    qualityThreshold: 0.85,
    model: 'gpt-4o',
    critiqueModel: 'gpt-4o-mini'
  },
  new OpenAI()
);

console.log(`优化轮次: ${result.iterations}, 最终评分: ${result.scores.at(-1)}`);

import math
import random
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class MCTSNode:
    """MCTS 推理节点"""
    state: str  # 当前推理状态
    parent: Optional['MCTSNode'] = None
    children: List['MCTSNode'] = field(default_factory=list)
    visits: int = 0
    value: float = 0.0
    reasoning_step: str = ""
    
    @property
    def ucb1(self) -> float:
        if self.visits == 0:
            return float('inf')
        exploitation = self.value / self.visits
        exploration = math.sqrt(2 * math.log(self.parent.visits) / self.visits)
        return exploitation + exploration


class ReasoningMCTS:
    """基于 MCTS 的推理搜索引擎"""
    
    def __init__(
        self,
        client: openai.OpenAI,
        num_simulations: int = 50,
        max_depth: int = 5,
        expansion_width: int = 3
    ):
        self.client = client
        self.num_simulations = num_simulations
        self.max_depth = max_depth
        self.expansion_width = expansion_width
    
    def search(self, problem: str) -> str:
        """执行 MCTS 搜索，返回最优推理路径"""
        root = MCTSNode(state=problem)
        
        for _ in range(self.num_simulations):
            # 1. Selection: 沿 UCB1 选路
            leaf = self._select(root)
            
            # 2. Expansion: 生成新推理步骤
            if leaf.visits > 0 and leaf.children == []:
                self._expand(leaf, problem)
                if leaf.children:
                    leaf = random.choice(leaf.children)
            
            # 3. Simulation: 快速评估
            value = self._simulate(leaf, problem)
            
            # 4. Backpropagation: 反向更新
            self._backpropagate(leaf, value)
        
        # 返回最佳路径
        return self._extract_best_path(root)
    
    def _select(self, node: MCTSNode) -> MCTSNode:
        """UCB1 选择策略"""
        current = node
        while current.children:
            current = max(current.children, key=lambda c: c.ucb1)
        return current
    
    def _expand(self, node: MCTSNode, problem: str):
        """展开节点：生成多个候选下一步"""
        path = self._reconstruct_path(node)
        
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": (
                    f"问题：{problem}\n"
                    f"当前推理：{path}\n\n"
                    f"生成 {self.expansion_width} 个不同的下一步推理方向。"
                    f"每个方向以 --- 分隔。"
                )
            }],
            temperature=0.9
        )
        
        steps = response.choices[0].message.content.split('---')
        for step in steps[:self.expansion_width]:
            child = MCTSNode(
                state=node.state + "\n" + step.strip(),
                parent=node,
                reasoning_step=step.strip()
            )
            node.children.append(child)
    
    def _simulate(self, node: MCTSNode, problem: str) -> float:
        """使用 LLM 快速评估当前路径的终局价值"""
        path = self._reconstruct_path(node)
        
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",  # 用小模型做快速评估
            messages=[{
                "role": "user",
                "content": (
                    f"评估以下推理路径最终得出正确答案的可能性 (0-1)：\n"
                    f"问题：{problem}\n"
                    f"当前推理：{path}\n"
                    f"只输出一个数字。"
                )
            }],
            temperature=0.0
        )
        
        try:
            return float(response.choices[0].message.content.strip())
        except ValueError:
            return 0.5
    
    def _backpropagate(self, node: MCTSNode, value: float):
        """反向传播更新"""
        current = node
        while current:
            current.visits += 1
            current.value += value
            current = current.parent
    
    def _reconstruct_path(self, node: MCTSNode) -> str:
        """重建从根到当前节点的推理路径"""
        steps = []
        current = node
        while current.parent:
            steps.append(current.reasoning_step)
            current = current.parent
        return " → ".join(reversed(steps)) if steps else "(起始)"
    
    def _extract_best_path(self, root: MCTSNode) -> str:
        """提取访问次数最多的路径（最可靠）"""
        path_steps = []
        current = root
        while current.children:
            current = max(current.children, key=lambda c: c.visits)
            path_steps.append(current.reasoning_step)
        return "\n".join(path_steps)

import OpenAI from 'openai';

// OpenAI o1 系列：通过 max_completion_tokens 控制推理预算
async function o1ReasoningWithBudget(
  problem: string,
  thinkingBudget: 'low' | 'medium' | 'high',
  client: OpenAI
) {
  const budgetMap = {
    low: 4096,     // 快速回答，最少推理
    medium: 16384, // 平衡模式
    high: 65536    // 深度推理，不计成本
  };

  const response = await client.chat.completions.create({
    model: 'o1',
    messages: [{ role: 'user', content: problem }],
    max_completion_tokens: budgetMap[thinkingBudget]
  });

  return {
    answer: response.choices[0].message.content,
    reasoningTokens: response.usage?.completion_tokens_details?.reasoning_tokens,
    outputTokens: response.usage?.completion_tokens_details?.accepted_prediction_tokens,
    totalCost: calculateCost(response.usage)
  };
}

// DeepSeek R1：推理过程可见
async function deepseekR1Reasoning(problem: string) {
  const client = new OpenAI({
    baseURL: 'https://api.deepseek.com/v1',
    apiKey: process.env.DEEPSEEK_API_KEY
  });

  const response = await client.chat.completions.create({
    model: 'deepseek-reasoner',
    messages: [{ role: 'user', content: problem }]
  });

  // DeepSeek R1 返回 reasoning_content（思考过程）和 content（最终答案）
  const message = response.choices[0].message as any;
  return {
    thinking: message.reasoning_content,  // 完整思考过程
    answer: message.content               // 最终答案
  };
}

function calculateCost(usage: any): number {
  // o1 的推理 Token 单价低于输出 Token
  const reasoningCost = (usage?.completion_tokens_details?.reasoning_tokens || 0) * 0.015 / 1000;
  const outputCost = (usage?.completion_tokens || 0) * 0.06 / 1000;
  const inputCost = (usage?.prompt_tokens || 0) * 0.015 / 1000;
  return reasoningCost + outputCost + inputCost;
}

import subprocess
import tempfile
from typing import Tuple

class CodeGenerationWithVerification:
    """带验证循环的代码生成（TTC 在代码领域的典型应用）"""
    
    def __init__(self, client: openai.OpenAI, max_attempts: int = 5):
        self.client = client
        self.max_attempts = max_attempts
    
    def generate_and_verify(
        self, 
        task: str, 
        test_code: str
    ) -> Tuple[str, int]:
        """生成代码并通过测试验证，失败则迭代修复"""
        
        code = ""
        for attempt in range(self.max_attempts):
            if attempt == 0:
                code = self._generate_initial(task)
            else:
                code = self._fix_code(task, code, error_output)
            
            success, error_output = self._run_tests(code, test_code)
            
            if success:
                return code, attempt + 1
        
        return code, self.max_attempts  # 返回最后一次尝试
    
    def _generate_initial(self, task: str) -> str:
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "编写 Python 代码。只输出代码，不要解释。"},
                {"role": "user", "content": task}
            ],
            temperature=0.3
        )
        return self._extract_code(response.choices[0].message.content)
    
    def _fix_code(self, task: str, code: str, error: str) -> str:
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "修复代码错误。只输出修正后的完整代码。"},
                {"role": "user", "content": f"任务：{task}\n\n代码：\n{code}\n\n错误：\n{error}"}
            ],
            temperature=0.2
        )
        return self._extract_code(response.choices[0].message.content)
    
    def _run_tests(self, code: str, test_code: str) -> Tuple[bool, str]:
        """在沙箱中运行代码和测试"""
        full_code = f"{code}\n\n{test_code}"
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(full_code)
            f.flush()
            result = subprocess.run(
                ['python', f.name],
                capture_output=True, text=True, timeout=10
            )
        
        if result.returncode == 0:
            return True, ""
        return False, result.stderr
    
    def _extract_code(self, text: str) -> str:
        if '```python' in text:
            return text.split('```python')[1].split('```')[0].strip()
        return text.strip()

from enum import Enum

class DifficultyLevel(Enum):
    TRIVIAL = "trivial"
    EASY = "easy"
    MEDIUM = "medium"
    HARD = "hard"
    EXPERT = "expert"

class AdaptiveComputeAllocator:
    """自适应推理计算分配器"""
    
    STRATEGIES = {
        DifficultyLevel.TRIVIAL: {
            "method": "direct",
            "samples": 1,
            "max_tokens": 256,
            "model": "gpt-4o-mini"
        },
        DifficultyLevel.EASY: {
            "method": "cot",
            "samples": 1,
            "max_tokens": 1024,
            "model": "gpt-4o-mini"
        },
        DifficultyLevel.MEDIUM: {
            "method": "self_consistency",
            "samples": 3,
            "max_tokens": 2048,
            "model": "gpt-4o"
        },
        DifficultyLevel.HARD: {
            "method": "self_consistency",
            "samples": 7,
            "max_tokens": 4096,
            "model": "gpt-4o"
        },
        DifficultyLevel.EXPERT: {
            "method": "mcts",
            "simulations": 30,
            "max_tokens": 8192,
            "model": "o1"
        }
    }
    
    def __init__(self, client: openai.OpenAI):
        self.client = client
    
    async def solve(self, problem: str) -> dict:
        difficulty = await self._classify_difficulty(problem)
        strategy = self.STRATEGIES[difficulty]
        
        if strategy["method"] == "direct":
            return await self._direct_solve(problem, strategy)
        elif strategy["method"] == "cot":
            return await self._cot_solve(problem, strategy)
        elif strategy["method"] == "self_consistency":
            return await self._sc_solve(problem, strategy)
        elif strategy["method"] == "mcts":
            return await self._mcts_solve(problem, strategy)
    
    async def _classify_difficulty(self, problem: str) -> DifficultyLevel:
        """用小模型快速分类问题难度"""
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": (
                    f"将以下问题分类为 trivial/easy/medium/hard/expert：\n"
                    f"{problem}\n只输出一个单词。"
                )
            }],
            temperature=0.0,
            max_tokens=10
        )
        level_str = response.choices[0].message.content.strip().lower()
        return DifficultyLevel(level_str) if level_str in [e.value for e in DifficultyLevel] else DifficultyLevel.MEDIUM

interface CascadeConfig {
  stages: Array<{
    model: string;
    maxTokens: number;
    confidenceThreshold: number;
  }>;
}

async function cascadeReasoning(
  problem: string,
  config: CascadeConfig,
  client: OpenAI
): Promise<{ answer: string; stage: number; totalCost: number }> {
  let totalCost = 0;

  for (let i = 0; i < config.stages.length; i++) {
    const stage = config.stages[i];
    
    const response = await client.chat.completions.create({
      model: stage.model,
      messages: [
        { role: 'system', content: '解决问题并评估你的置信度 (0-1)。以 JSON 返回 {"answer": "...", "confidence": 0.X}' },
        { role: 'user', content: problem }
      ],
      max_tokens: stage.maxTokens,
      response_format: { type: 'json_object' }
    });

    totalCost += estimateCost(response.usage, stage.model);
    const result = JSON.parse(response.choices[0].message.content || '{}');

    if (result.confidence >= stage.confidenceThreshold) {
      return { answer: result.answer, stage: i + 1, totalCost };
    }
  }

  // 最后阶段无论如何返回结果
  return { answer: 'Fallback to last stage', stage: config.stages.length, totalCost };
}

// 使用：简单问题在第一阶段解决，复杂问题逐步升级
const cascade = await cascadeReasoning(problem, {
  stages: [
    { model: 'gpt-4o-mini', maxTokens: 512, confidenceThreshold: 0.9 },
    { model: 'gpt-4o', maxTokens: 2048, confidenceThreshold: 0.8 },
    { model: 'o1', maxTokens: 16384, confidenceThreshold: 0.0 }
  ]
}, client);

async def early_stopping_consistency(
    problem: str,
    client: openai.AsyncOpenAI,
    max_samples: int = 10,
    consensus_threshold: int = 3
) -> dict:
    """带早停的 Self-Consistency：连续 N 次相同答案则停止"""
    
    answers = []
    consecutive_same = 0
    last_answer = None
    
    for i in range(max_samples):
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "逐步推理，最终答案用 \\boxed{} 包裹。"},
                {"role": "user", "content": problem}
            ],
            temperature=0.7
        )
        
        answer = extract_boxed_answer(response.choices[0].message.content)
        answers.append(answer)
        
        # 早停：连续 consensus_threshold 次相同答案
        if answer == last_answer:
            consecutive_same += 1
            if consecutive_same >= consensus_threshold:
                break
        else:
            consecutive_same = 1
            last_answer = answer
    
    vote_counts = Counter(answers)
    best = vote_counts.most_common(1)[0]
    
    return {
        "answer": best[0],
        "confidence": best[1] / len(answers),
        "samples_used": len(answers),
        "samples_saved": max_samples - len(answers),
        "early_stopped": len(answers) < max_samples
    }