Reduce redundant tokens in parallel reasoning by 80% while maintaining accuracy via dynamic pruning of equivalent reasoning paths. Trigger: improve efficiency of consensus-based reasoning (multiple CoT generation).
Reduce redundant tokens in parallel reasoning by 80% while maintaining accuracy via dynamic pruning of equivalent reasoning paths. Trigger: improve efficiency of consensus-based reasoning (multiple CoT generation).
DeepPrune: Eliminating Redundant Reasoning Through Dynamic Pruning
Core Concept
When generating multiple reasoning chains (consensus sampling), over 80% of tokens are wasted on duplicate reasoning paths that reach the same conclusion. DeepPrune introduces a judge model that predicts answer equivalence from incomplete traces, allowing dynamic pruning of redundant paths before completion. The approach achieves 80%+ token reduction while maintaining accuracy within 3 percentage points.
The key insight: A lightweight classifier can identify when parallel reasoning traces will converge to the same answer before they finish generating.
Architecture Overview
Judge Model: Trained classifier predicting answer equivalence from partial reasoning
Dynamic Pruning: On-the-fly elimination of redundant traces during generation
Online Clustering: Group equivalent reasoning paths in real-time
Focal Loss Training: Handle class imbalance (most traces are equivalent)
Zero-Shot Transfer: Judge generalizes across different problem types
Implementation Steps
1. Understand Reasoning Trace Equivalence
Define what makes two reasoning paths equivalent.
classReasoningEquivalenceAnalyzer:
"""
Determine when different reasoning traces reach the same conclusion.
""" @staticmethoddefextract_answer(trace):
"""Extract final answer from reasoning trace."""# Look for common answer markersif"Answer:"in trace:
return trace.split("Answer:")[-1].strip().split('\n')[0]
elif"Therefore,"in trace:
return trace.split("Therefore,")[-1].strip().split('\n')[]
:
lines = [l.strip() l trace.split() l.strip()]
lines[-] lines
():
answer1.lower() == answer2.lower():
:
num1 = (answer1.replace(, ))
num2 = (answer2.replace(, ))
(num1 - num2) <
:
tokens1 = (answer1.lower().split())
tokens2 = (answer2.lower().split())
(tokens1 | tokens2) == :
jaccard = (tokens1 & tokens2) / (tokens1 | tokens2)
jaccard >
0
else
# Use last meaningful line
for
in
'\n'
if
return
1
if
else
""
@staticmethod
def
are_answers_equivalent
answer1, answer2
"""
Check if two answers are semantically equivalent.
"""
# Exact match
if
return
True
# Numeric equivalence
try
float
','
''
float
','
''
return
abs
1e-6
except
pass
# Jaccard similarity for text answers
set
set
if
len
0
return
False
len
len
return
0.8
2. Train Judge Model
Build a classifier that predicts answer equivalence from partial traces.
classJudgeModel:
"""
Predict if incomplete reasoning traces will reach equivalent answers.
"""def__init__(self, model, hidden_size=256):
self.model = model
# Simple MLP judge: embedding → classificationself.embedding_layer = torch.nn.Linear(768, hidden_size) # BERT-likeself.hidden_layer = torch.nn.Linear(hidden_size, 128)
self.output_layer = torch.nn.Linear(128, 1) # Binary classificationdefembed_trace(self, partial_trace):
"""
Embed a partial reasoning trace.
"""# Use pretrained encoder
tokens = tokenize(partial_trace)
embeddings = self.model.encode(tokens)
# Mean pooling over sequence
trace_embedding = torch.mean(embeddings, dim=0)
return trace_embedding
defpredict_equivalence(self, trace1, trace2):
"""
Predict: will these traces reach the same answer?
Args:
trace1, trace2: Partial reasoning traces
Returns:
Probability [0, 1] that answers will be equivalent
"""# Embed both traces
emb1 = self.embed_trace(trace1)
emb2 = self.embed_trace(trace2)
# Concatenate embeddings
combined = torch.cat([emb1, emb2], dim=-1)
# Classify
hidden = torch.relu(self.embedding_layer(combined))
hidden = torch.relu(self.hidden_layer(hidden))
logit = self.output_layer(hidden)
prob = torch.sigmoid(logit)
return prob.item()
deftrain_on_dataset(self, reasoning_pairs, num_epochs=10):
"""
Train judge on pairs of complete reasoning traces.
Args:
reasoning_pairs: List of (trace1, trace2, are_equivalent)
"""
optimizer = torch.optim.Adam(self.parameters(), lr=1e-4)
# Compute class weights for focal loss (handle imbalance)
equiv_count = sum(1for _, _, equiv in reasoning_pairs if equiv)
total_count = len(reasoning_pairs)
pos_weight = (total_count - equiv_count) / (equiv_count + 1)
for epoch inrange(num_epochs):
epoch_loss = 0for trace1, trace2, are_equiv in reasoning_pairs:
# Embed traces
emb1 = self.embed_trace(trace1)
emb2 = self.embed_trace(trace2)
combined = torch.cat([emb1, emb2], dim=-1)
# Forward pass
hidden = torch.relu(self.embedding_layer(combined))
hidden = torch.relu(self.hidden_layer(hidden))
logit = self.output_layer(hidden)
prob = torch.sigmoid(logit)
# Focal loss: emphasize hard examples
target = float(are_equiv)
bce_loss = torch.nn.functional.binary_cross_entropy(
prob,
torch.tensor(target)
)
# Focal term
pt = prob if target == 1else (1 - prob)
focal_loss = -(1 - pt) ** 2 * bce_loss
# Backprop
optimizer.zero_grad()
focal_loss.backward()
optimizer.step()
epoch_loss += focal_loss.item()
print(f"Epoch {epoch}: loss={epoch_loss / len(reasoning_pairs):.4f}")
3. Implement Dynamic Pruning
Prune redundant traces during generation.
classDynamicPruner:
"""
Prune redundant reasoning traces on-the-fly.
"""def__init__(self, judge_model, redundancy_threshold=0.85):
self.judge = judge_model
self.threshold = redundancy_threshold
defshould_prune_trace(self, current_traces, new_trace):
"""
Decide if new_trace is redundant with existing ones.
Args:
current_traces: List of active reasoning traces
new_trace: New trace to evaluate
Returns:
Boolean: should prune this trace?
"""for existing_trace in current_traces:
# Compare at partial level
equivalence_prob = self.judge.predict_equivalence(
existing_trace,
new_trace
)
if equivalence_prob > self.threshold:
# New trace is redundant with existing onereturnTrue# Not redundant with any existing tracereturnFalsedefprune_and_cluster(self, all_traces):
"""
Group traces into equivalence clusters.
Args:
all_traces: All completed reasoning traces
Returns:
List of equivalence clusters (representative + size)
"""
clusters = []
for trace in all_traces:
assigned = Falsefor cluster in clusters:
# Check equivalence with cluster representative
equiv_prob = self.judge.predict_equivalence(
cluster["representative"],
trace
)
if equiv_prob > self.threshold:
cluster["size"] += 1
assigned = Truebreakifnot assigned:
# New cluster
clusters.append({
"representative": trace,
"size": 1
})
return clusters
4. Orchestrate Parallel Generation with Pruning
Generate multiple traces in parallel, pruning redundant ones.
classPrunedParallelReasoner:
"""
Generate multiple reasoning traces with dynamic pruning.
"""def__init__(self, model, judge_model, num_parallel=4):
self.model = model
self.pruner = DynamicPruner(judge_model)
self.num_parallel = num_parallel
defgenerate_with_pruning(self, problem, max_tokens=500):
"""
Generate multiple reasoning traces, pruning redundant ones.
Args:
problem: Problem to reason about
max_tokens: Max tokens per trace
Returns:
Dictionary with unique traces and redundancy stats
"""
active_traces = [""] * self.num_parallel
token_counts = [0] * self.num_parallel
completed_traces = []
pruned_count = 0# Generate tokens in roundsfor token_round inrange(max_tokens):
for trace_idx inrange(len(active_traces)):
if active_traces[trace_idx] isNone:
continue# Already pruned# Generate next token
next_token = self.model.generate_one_token(
problem + active_traces[trace_idx],
temperature=0.8
)
active_traces[trace_idx] += next_token
token_counts[trace_idx] += 1# Check for completionif is_complete_solution(active_traces[trace_idx]):
completed_traces.append(active_traces[trace_idx])
active_traces[trace_idx] = None# Pruning step: every 50 tokens, check for redundancyif token_round % 50 == 0:
for idx inrange(len(active_traces)):
if active_traces[idx] isNone:
continue# Check if redundant with completed tracesfor completed in completed_traces:
equiv_prob = self.pruner.judge.predict_equivalence(
active_traces[idx],
completed
)
if equiv_prob > self.pruner.threshold:
# Prune this trace
active_traces[idx] = None
pruned_count += 1break# Collect remaining active tracesfor trace in active_traces:
if trace isnotNone:
completed_traces.append(trace)
# Cluster identical answers
clusters = self.pruner.prune_and_cluster(completed_traces)
# Calculate efficiency
total_tokens_generated = sum(token_counts)
unique_traces = len(clusters)
savings_ratio = (total_tokens_generated - sum(
len(c["representative"].split()) * c["size"]
for c in clusters
)) / total_tokens_generated
return {
"traces": [c["representative"] for c in clusters],
"cluster_sizes": [c["size"] for c in clusters],
"total_tokens": total_tokens_generated,
"pruned_traces": pruned_count,
"savings_ratio": savings_ratio
}
