Reasoning Models
Chain-of-Thought, Tree-of-Thoughts, self-consistency, process reward models, and test-time compute scaling.
Reasoning Paradigms
Chain-of-Thought (CoT)
Question: What is 15 * 24?
Reasoning: 15 * 24 = 15 * (20 + 4) = 300 + 60 = 360
Answer: 360
Variants: - Zero-shot: "Let's think step by step" - Few-shot: Exemplars with reasoning - Auto-CoT: Cluster questions → generate reasoning - Structured CoT: Enforce format (steps, verification)
Tree-of-Thoughts (ToT)
Root: Query
│
┌───────────┼───────────┐
▼ ▼ ▼
Thought 1 Thought 2 Thought 3
(value=0.7) (value=0.3) (value=0.8)
│ │ │
▼ ▼ ▼
Branch 1 Branch 2 Branch 3
(pruned) (explore) (best)
class TreeOfThoughts:
def __init__(self, llm, breadth=3, depth=3, evaluator=None):
self.llm = llm
self.breadth = breadth
self.depth = depth
self.evaluator = evaluator or self.default_evaluator
async def search(self, problem: str) -> str:
root = Node(state=problem, value=0)
for level in range(self.depth):
candidates = []
for node in current_level:
# Generate k thoughts
thoughts = await self.generate_thoughts(node.state, self.breadth)
for thought in thoughts:
child = Node(state=thought, parent=node)
child.value = await self.evaluator(thought)
candidates.append(child)
# Prune: keep top-breadth
current_level = sorted(candidates, key=lambda n: n.value, reverse=True)[:self.breadth]
# Return best path
best = max(current_level, key=lambda n: n.value)
return self.extract_solution(best)
Graph-of-Thoughts (GoT)
# DAG instead of tree - allows merging, recombination
class GraphOfThoughts:
def __init__(self, llm):
self.llm = llm
self.graph = nx.DiGraph()
async def reason(self, problem: str) -> str:
# 1. Decompose into sub-thoughts
thoughts = await self.decompose(problem)
# 2. Build dependency graph
for thought in thoughts:
self.graph.add_node(thought.id, content=thought.content)
for dep in thought.dependencies:
self.graph.add_edge(dep, thought.id)
# 3. Topological execution
for node_id in nx.topological_sort(self.graph):
thought = self.graph.nodes[node_id]['content']
result = await self.execute_thought(thought)
self.graph.nodes[node_id]['result'] = result
# 4. Aggregate
return await self.aggregate(problem, self.graph)
Self-Consistency
async def self_consistency(llm, query: str, n_samples=8) -> str:
# Sample multiple CoT trajectories
responses = await asyncio.gather(*[
llm.generate(query + "\nLet's think step by step:")
for _ in range(n_samples)
])
# Extract answers
answers = [extract_answer(r) for r in responses]
# Majority vote
return majority_vote(answers)
Program-of-Thoughts (PoT)
# Reasoning as executable code
query = "What is 15% of 240?"
pot_prompt = f"""
Write a Python program to solve:
{query}
```python
# Your code here
result = ...
print(result)
"""
Execute in sandbox
code = llm.generate(pot_prompt) result = execute_python(code)
---
## Process Reward Models (PRM)
### Step-Level Rewards
```python
class ProcessRewardModel:
def __init__(self, model):
self.model = model # Trained to score reasoning steps
def score_step(self, question: str, step: str, context: str) -> float:
"""Score a single reasoning step: 0-1"""
prompt = f"Question: {question}\nContext: {context}\nStep: {step}\nScore (0-1):"
return float(self.model.generate(prompt))
def score_trajectory(self, question: str, steps: list[str]) -> list[float]:
scores = []
context = ""
for step in steps:
score = self.score_step(question, step, context)
scores.append(score)
context += f"\n{step}"
return scores
PRM Training
# Annotate steps: +1 (correct), -1 (incorrect), 0 (neutral)
# Train: step + context → reward
# Loss: MSE or ranking loss
class PRMTrainer:
def train(self, trajectories: list[Trajectory]):
# Each trajectory: question, steps[], final_answer, correctness
# Annotate each step
for traj in trajectories:
for i, step in enumerate(traj.steps):
label = self.annotate_step(traj, i)
yield (traj.question, step, traj.context[:i], label)
Test-Time Compute Scaling (o1 Paradigm)
Scaling Laws for Reasoning
Performance = f(Test-Time Compute)
o1-Style Scaling
class O1StyleReasoner:
def __init__(self, model, prm, budget=1000):
self.model = model
self.prm = prm
self.budget = budget # Token budget for reasoning
async def reason(self, query: str) -> str:
best_answer = None
best_score = -1
for attempt in range(self.max_attempts):
# Generate with reasoning tokens
response = await self.model.generate(
query,
max_tokens=self.budget,
stop=["Final Answer:"]
)
# Score with PRM
score = await self.prm.score_full(response)
if score > best_score:
best_score = score
best_answer = response
return best_answer
MCTS + PRM (AlphaGo-Style)
class MCTSReasoner:
def __init__(self, model, prm, simulations=50):
self.model = model
self.prm = prm
self.simulations = simulations
async def search(self, question: str) -> str:
root = MCTSNode(state=question)
for _ in range(self.simulations):
# Selection
node = self.select(root)
# Expansion
if not node.is_terminal:
children = await self.expand(node)
node.add_children(children)
# Simulation (rollout)
reward = await self.simulate(node)
# Backpropagation
self.backpropagate(node, reward)
# Best child
best_child = max(root.children, key=lambda c: c.visits)
return best_child.state
async def expand(self, node):
# Generate k next steps
steps = await self.model.generate_next_steps(node.state, k=4)
return [MCTSNode(state=s, parent=node) for s in steps]
async def simulate(self, node):
# Quick rollout to terminal
state = node.state
for _ in range(10):
if node.is_terminal: break
step = await self.model.generate_step(state)
state += step
return await self.prm.score(state)
Reasoning Model Architectures
1. Outcome Reward Models (ORM)
- Score final answer only
- Simpler, less supervision needed
- Used in early RLHF
2. Process Reward Models (PRM)
- Score each reasoning step
- More granular, better credit assignment
- Used in o1, Math-Shepherd
3. Self-Rewarding Models
class SelfRewardingModel:
async def improve(self, query: str, attempts=3):
for i in range(attempts):
response = await self.generate(query)
reward = await self.self_evaluate(query, response)
if reward > self.threshold:
# Add to training data
self.add_to_buffer(query, response)
break
Reasoning Benchmarks
| Benchmark | Domain | Metric |
|---|---|---|
| GSM8K | Grade school math | Accuracy |
| MATH | Competition math | Accuracy |
| BBH | Big-Bench Hard | Accuracy |
| GPQA | Graduate-level QA | Accuracy |
| TheoremQA | Theorem proving | Accuracy |
| ProofNet | Formal proofs | Pass rate |
| HumanEval | Code generation | Pass@1 |
| MBPP | Basic programming | Pass@1 |
Training Reasoning Models
SFT on Reasoning Traces
# Collect high-quality CoT traces
reasoning_data = [
{"question": q, "response": cot + answer}
for q, cot, answer in high_quality_traces
]
# Standard SFT
trainer = Trainer(
model=model,
train_dataset=reasoning_data,
args=TrainingArguments(
learning_rate=1e-5,
num_train_epochs=2,
packing=True
)
)
RL with PRM (o1-style)
# PPO with step-level rewards
class ReasoningPPOTrainer:
def step(self, batch):
# Generate trajectories
queries = batch["question"]
responses = self.model.generate(queries, max_tokens=2048)
# Score with PRM
rewards = self.prm.score_trajectories(queries, responses)
# PPO update
self.ppo_update(queries, responses, rewards)
RLAIF (AI Feedback)
# Constitutional AI: model critiques itself
async def constitutional_training(model, constitution):
for batch in dataloader:
# Generate
responses = model.generate(batch.questions)
# Critique
critiques = await ai_critique(model, batch.questions, responses, constitution)
# Revise
revised = await model.revise(batch.questions, responses, critiques)
# Train on (question, revised) pairs
sft_train(model, batch.questions, revised)
Inference Optimization for Reasoning
Token Budget Management
class ReasoningBudgetManager:
def __init__(self, max_tokens=4096):
self.max = max_tokens
def allocate(self, problem_complexity: float) -> dict:
# Complex problems get more tokens
reasoning = int(self.max * (0.3 + 0.5 * problem_complexity))
return {
"reasoning": reasoning,
"answer": self.max - reasoning
}
Early Exit for Simple Problems
def adaptive_reasoning(model, query, prm):
# Quick classification
complexity = classify_complexity(query)
if complexity < 0.3:
# Direct answer
return model.generate(query, max_tokens=256)
else:
# Full reasoning
return reasoned_generation(model, query, prm)
Related Notes
- Mixture of Experts — MoE for reasoning
- Model Routing — Routing to reasoning models
- Inference Optimization — Reasoning inference
- Single Agent Patterns — ToT, GoT as agent patterns
- Cognitive Architectures — Neural-symbolic reasoning
- Coding Agents — Code reasoning
- Scientific Agents — Math/science reasoning
- Agent Benchmarks — Reasoning benchmarks
- Self Improving Agents — Self-improving reasoning
- Formal Verification — Verified reasoning
Sources: "Chain-of-Thought" (Wei et al.), "Tree of Thoughts" (Yao et al.), "Graph of Thoughts" (Besta et al.), "Self-Consistency" (Wang et al.), "Program of Thoughts" (Chen et al.), "Math-Shepherd" (Wang et al.), "Process Reward Models" (Lightman et al.), "o1" (OpenAI), "AlphaGo" (Silver et al.), "RLAIF" (Bai et al.), "Constitutional AI" (Bai et al.), "Self-Rewarding" (Yuan et al.).