# 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)
```

```python
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)
```python
# 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
```python
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)
```python
# 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
```python
# 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
```python
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)
```python
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
```python
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
```python
# 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)
```python
# 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)
```python
# 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
```python
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
```python
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

- [[ModelArchitectures/Mixture-of-Experts.md]] — MoE for reasoning
- [[ModelArchitectures/Model-Routing.md]] — Routing to reasoning models
- [[ModelArchitectures/Inference-Optimization.md]] — Reasoning inference
- [[Architectures/Single-Agent-Patterns.md]] — ToT, GoT as agent patterns
- [[Architectures/Cognitive-Architectures.md]] — Neural-symbolic reasoning
- [[Domains/Coding-Agents.md]] — Code reasoning
- [[Domains/Scientific-Agents.md]] — Math/science reasoning
- [[Evaluation/Agent-Benchmarks.md]] — Reasoning benchmarks
- [[Frontiers/Self-Improving-Agents.md]] — Self-improving reasoning
- [[Frontiers/Formal-Verification.md]] — 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.).*