Starphix
Agentic Engineering / Sources
The Lab / Sources / Agentic Engineering / ModelArchitectures/Reasoning-Models.md

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)


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.).

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown