菜单
Monitor parallel reasoning branches via 2D probing—periodically extracting intermediate answers to detect consensus and prune divergent branches. Reduces token cost by 25%+ while maintaining accuracy without requiring model retraining.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/ADu2021/skillXiv --skill parallel-probe-efficient-reasoning命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
基于 SOC 职业分类
| name | parallel-probe-efficient-reasoning |
| title | Parallel-Probe: Towards Efficient Parallel Thinking via 2D Probing |
| version | 0.0.2 |
| engine | skillxiv-v0.0.2-claude-opus-4.6 |
| license | MIT |
| url | https://arxiv.org/abs/2602.03845 |
| keywords | ["Parallel Reasoning","Early Stopping","Branch Pruning","Token Efficiency","Inference Optimization"] |
| description | Monitor parallel reasoning branches via 2D probing—periodically extracting intermediate answers to detect consensus and prune divergent branches. Reduces token cost by 25%+ while maintaining accuracy without requiring model retraining. |
When solving difficult problems, LLMs benefit from parallel reasoning paths that explore different approaches. However, many branches diverge from the consensus solution, creating computational waste. Parallel-Probe uses periodic probing to detect when the majority vote has stabilized and when branches have become unproductive, enabling dynamic early stopping and branch pruning without model retraining.
The key insight is that intermediate solutions reveal reasoning quality early: branches converging on the same answer should continue, while outliers should be pruned. This allows efficient parallel reasoning by coupling width (number of branches) and depth (reasoning length) dynamically.
Parallel-Probe operates on two complementary mechanisms:
Consensus-Based Early Stopping: At periodic intervals, extract the current best answer from each branch via majority voting. If consensus remains stable across consecutive probes, terminate generation for all branches.
Deviation-Based Branch Pruning: Identify branches that consistently diverge from consensus and remove them, redirecting their computational budget to productive branches.
This 2D monitoring (width × depth) allows the system to optimize both dimensions without explicit model modification.
Create N independent reasoning chains in parallel using a model API supporting batch processing.
# Parallel branch generation
class ParallelBranchGenerator:
def __init__(self, model: str, num_branches: int = ,
max_tokens: = ):
.model = model
.num_branches = num_branches
.max_tokens = max_tokens
.branches = {}
.branch_active = {}
():
branch_id (.num_branches):
.branches[branch_id] = prompt
.branch_active[branch_id] =
():
active_branches = {i: .branches[i]
i (.num_branches)
.branch_active[i]}
active_branches:
prompts = (active_branches.values())
outputs = .model.batch_generate(
prompts,
max_new_tokens=num_tokens,
temperature=
)
(branch_id, prompt), output (active_branches.items(), outputs):
.branches[branch_id] += output
(.branches[branch_id].split()) >= .max_tokens:
.branch_active[branch_id] =
() -> [, ]:
.branches.copy()
Extract intermediate answers from all branches at regular intervals.
# Periodic probing mechanism
class ProbeMonitor:
def __init__(self, model: str, probe_interval: int = 10):
"""
Extract intermediate answers periodically.
Args:
model: LLM for extracting answers
probe_interval: Extract answer every N tokens per branch
"""
self.model = model
self.probe_interval = probe_interval
self.total_tokens_generated = 0
def should_probe(self, tokens_generated_this_step: int) -> bool:
"""Check if it's time to probe based on token count."""
self.total_tokens_generated += tokens_generated_this_step
return self.total_tokens_generated % self.probe_interval == 0
def extract_intermediate_answers(self, branches: Dict[int, str]) -> Dict[int, str]:
"""Extract best answer from each branch via LLM summarization."""
answers = {}
for branch_id, content in branches.items():
# Use the model to extract the final answer so far
extraction_prompt = f"""
Given this reasoning so far:
{content[-500:]}
What is the most likely final answer based on current reasoning?
Respond with just the answer, nothing else."""
answer = self.model.generate(
extraction_prompt,
max_tokens=50,
temperature=0.0 # Deterministic extraction
)
answers[branch_id] = answer.strip()
return answers
def compute_consensus(self, answers: Dict[int, str],
threshold: float = 0.5) -> Optional[str]:
"""
Determine consensus answer via majority voting.
Args:
answers: Extracted answer from each branch
threshold: Fraction of branches that must agree
Returns:
Consensus answer if threshold met, else None
"""
from collections import Counter
if not answers:
return None
counts = Counter(answers.values())
most_common = counts.most_common(1)[0]
answer, count = most_common
agreement_ratio = count / len(answers)
if agreement_ratio >= threshold:
return answer
return None
Monitor consensus stability to trigger early termination.
# Consensus tracking and early stopping
class ConsensusTracker:
def __init__(self, stability_window: int = 3):
"""
Track consensus stability.
Args:
stability_window: Number of consecutive probes
showing same consensus needed to stop
"""
self.stability_window = stability_window
self.consensus_history = [] # List of consensus values
self.probe_count = 0
def update_consensus(self, consensus: Optional[str]):
"""Add new consensus observation."""
self.consensus_history.append(consensus)
self.probe_count += 1
def is_consensus_stable(self) -> bool:
"""
Check if consensus has stabilized across recent probes.
Returns:
True if last N probes show same consensus, False otherwise
"""
if len(self.consensus_history) < self.stability_window:
return False
recent = self.consensus_history[-self.stability_window:]
# Check if all recent values are same (ignoring None)
recent_valid = [c for c in recent if c is not None]
if len(recent_valid) < self.stability_window:
return False
return all(c == recent_valid[0] for c in recent_valid)
def should_stop_early(self) -> bool:
"""Decide whether to stop all generation."""
return self.is_consensus_stable()
Remove branches that consistently diverge from consensus.
# Branch pruning mechanism
class BranchPruner:
def __init__(self, divergence_threshold: float = 0.3,
window_size: int = 5):
"""
Identify and prune divergent branches.
Args:
divergence_threshold: Fraction of probes where branch disagrees
window_size: Number of recent probes to consider
"""
self.divergence_threshold = divergence_threshold
self.window_size = window_size
self.branch_history = {} # branch_id -> list of (answer, consensus)
def track_branch(self, branch_id: int, answer: str, consensus: str):
"""Record branch answer and consensus."""
if branch_id not in self.branch_history:
self.branch_history[branch_id] = []
matches_consensus = (answer == consensus)
self.branch_history[branch_id].append(matches_consensus)
def identify_divergent_branches(self) -> Set[int]:
"""Identify branches that should be pruned."""
divergent = set()
for branch_id, history in self.branch_history.items():
# Check recent window
recent = history[-self.window_size:]
if not recent:
continue
# Count divergences
divergence_count = sum(1 for match in recent if not match)
divergence_rate = divergence_count / len(recent)
if divergence_rate > self.divergence_threshold:
divergent.add(branch_id)
return divergent
Orchestrate the full reasoning process with probing and pruning.
# Main inference loop with Parallel-Probe
def parallel_probe_reasoning(
prompt: str,
model: str,
num_branches: int = 4,
probe_interval: int = 10,
max_tokens: int = 1024
) -> str:
"""
Execute parallel reasoning with probing and pruning.
Returns:
Final consensus answer
"""
# Initialize
generator = ParallelBranchGenerator(model, num_branches, max_tokens)
generator.initialize_branches(prompt)
probe_monitor = ProbeMonitor(model, probe_interval)
consensus_tracker = ConsensusTracker(stability_window=3)
pruner = BranchPruner(divergence_threshold=0.3)
total_tokens = 0
step = 0
# Main generation loop
while step < max_tokens // 4:
# Generate next tokens
has_active = generator.step_all_branches(num_tokens=4)
if not has_active:
break
total_tokens += 4 * generator.num_branches
step += 1
# Periodic probing
if probe_monitor.should_probe(4 * generator.num_branches):
# Extract answers from active branches
branches = generator.get_branch_contents()
answers = probe_monitor.extract_intermediate_answers(branches)
# Compute consensus
consensus = probe_monitor.compute_consensus(
answers,
threshold=0.5
)
consensus_tracker.update_consensus(consensus)
# Track for pruning
for branch_id, answer in answers.items():
pruner.track_branch(branch_id, answer, consensus or "")
# Prune divergent branches
divergent = pruner.identify_divergent_branches()
for branch_id in divergent:
generator.branch_active[branch_id] = False
# Check early stopping
if consensus_tracker.should_stop_early():
print(f"Early stop at step {step}: consensus achieved")
break
# Return final consensus
final_branches = generator.get_branch_contents()
final_answers = probe_monitor.extract_intermediate_answers(final_branches)
final_consensus = probe_monitor.compute_consensus(final_answers, threshold=0.5)
return final_consensus or list(final_answers.values())[0]
When to use Parallel-Probe:
When not to use:
Common Pitfalls:
Hyperparameter Guidelines:
| Parameter | Recommended | Tuning Strategy |
|---|---|---|
| num_branches | 4-8 | Higher = better exploration; diminishing returns above 8 |
| probe_interval | 10-20 tokens | Lower = earlier stopping; higher = less probing overhead |
| consensus_threshold | 0.5-0.7 | Higher = stricter consensus; 0.5 = simple majority |
| stability_window | 2-3 probes | Number of consecutive probes confirming consensus |
| divergence_threshold | 0.3-0.5 | Fraction of probes where branch must disagree to prune |
See the full paper at: https://arxiv.org/abs/2602.03845
Key results: Up to 35.8% sequential token reduction and 25.8% total token cost reduction while maintaining competitive accuracy. Training-free; works with any off-the-shelf LLM. SCOUT evaluation testbed released for prototyping similar strategies.