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).
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
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
}
