threadweaver-parallel-reasoning

name	threadweaver-parallel-reasoning
title	ThreadWeaver: Adaptive Threading for Efficient Parallel Reasoning in Language Models
version	0.0.2
engine	skillxiv-v0.0.2-claude-opus-4.6
license	MIT
url	https://arxiv.org/abs/2512.07843
keywords	["parallel reasoning","speculative decoding","inference optimization","chain-of-thought","adaptive parallelization"]
description	Enable parallel reasoning threads on standard autoregressive inference engines without modifications to position embeddings or KV cache. ThreadWeaver achieves 1.53× speedup while maintaining chain-of-thought quality—ideal when you need faster reasoning without special hardware.

Overview

ThreadWeaver implements adaptive parallel reasoning through a trie-based training-inference co-design that operates on standard autoregressive inference engines. The framework combines parallel trajectory generation, trie-structured reasoning paths, and parallelization-aware RL to balance reasoning accuracy with effective parallelization strategies.

When to Use

• Reasoning tasks where sequential chain-of-thought is bottleneck
• Scenarios requiring compatibility with standard LLM inference
• Need for 1.5× speedup without architectural changes
• Mathematical reasoning and problem-solving tasks
• Applications where multiple reasoning paths are beneficial
• Inference on existing autoregressive engines

When NOT to Use

• Tasks with strict sequential dependencies
• Simple one-step reasoning requirements
• Scenarios where parallel execution fails
• Applications already achieving acceptable latency
• Models without reasoning components

Core Technique

Trie-based parallel reasoning with training-inference co-design:

# Adaptive threading for parallel reasoning
class ThreadWeaverReasoner:
    def __init__(self, base_model):
        self.model = base_model
        self.trie_structure = ReasoningTrie()

    def two_stage_trajectory_generation(self, problems, num_trajectories=4):
        """
        Stage 1: Generate large-scale, high-quality chain-of-thought data
        with parallel annotations for supervised fine-tuning.
        """
        parallel_data = []

        for problem in problems:
            # Generate multiple reasoning trajectories
            trajectories = []
            for seed in range(num_trajectories):
                # Sequential CoT generation
                traj = self.model.generate_cot(problem, seed=seed)
                trajectories.append(traj)

            
            divergence_points = .find_divergence_points(trajectories)

            
            parallel_structure = {
                : problem,
                : trajectories,
                : divergence_points,
                : .extract_parallel_paths(trajectories)
            }

            parallel_data.append(parallel_structure)

         parallel_data

     ():
        
        divergence_points = []

        
        max_len = ((t)  t  trajectories)

         step_idx  (max_len):
            
            steps_at_idx = [
                t[step_idx]  step_idx < (t)  
                 t  trajectories
            ]

            
             .can_branch_here(steps_at_idx):
                divergence_points.append(step_idx)

         divergence_points

     ():
        
        
        trie = ReasoningTrie()

         item  parallel_data:
            problem = item[]
             traj  item[]:
                
                trie.insert(problem, traj)

        
         batch  .create_trie_batches(trie):
            problem, shared_prefix, parallel_branches = batch

            
            
            shared_hidden = .model.encode(problem + shared_prefix)

            
            branch_losses = []
             branch  parallel_branches:
                
                
                branch_loss = .compute_branch_loss(
                    shared_hidden,
                    branch
                )
                branch_losses.append(branch_loss)

            
            total_loss = torch.stack(branch_losses).mean()
            total_loss.backward()

            .optimizer.step()

        .trie_structure = trie
         trie

     ():
        
        
        env = ReasoningEnvironment()

         episode  (num_episodes):
            problem = env.sample_problem()
            state = env.reset(problem)

            total_reward = 
            trajectory = []

             step  (max_steps):
                
                parallelize_action = .policy.sample_action(state)

                 parallelize_action:
                    
                    candidates = .generate_parallel_candidates(
                        state,
                        num_candidates=
                    )
                    next_step = .select_best_candidate(candidates)
                :
                    
                    next_step = .model.generate_step(state)

                
                accuracy_reward = env.step(next_step)

                
                efficiency_reward = .compute_efficiency_reward(
                    parallelize_action
                )

                
                reward = accuracy_reward +  * efficiency_reward

                state = env.get_next_state(next_step)
                trajectory.append((state, parallelize_action, reward))
                total_reward += reward

            
            .update_policy(trajectory, total_reward)

         .policy

     ():
        
        candidates = []

         candidate_idx  (num_candidates):
            
            
            candidate = .model.generate_step(
                state,
                temperature=,
                seed=candidate_idx
            )
            candidates.append(candidate)

         candidates

     ():
        
        scores = []
         candidate  candidates:
            
            score = .score_candidate(candidate)
            scores.append(score)

        best_idx = torch.argmax(torch.tensor(scores))
         candidates[best_idx]

     ():
        
        
        initial_state = .model.encode(problem)

        
        current_node = .trie_structure.root

        reasoning_trace = []
        num_steps = 

          current_node.is_leaf()  num_steps < max_steps:
            
             (current_node.children) > :
                
                
                parallel_results = []

                 child_node  current_node.children:
                    
                    branch_output = .execute_branch(
                        initial_state,
                        reasoning_trace,
                        child_node
                    )
                    parallel_results.append(branch_output)

                
                merged = .merge_parallel_results(parallel_results)
                current_node = merged[]
                reasoning_trace.append(merged[])
            :
                
                child = current_node.children[]
                step = .execute_branch(
                    initial_state,
                    reasoning_trace,
                    child
                )
                current_node = child
                reasoning_trace.append(step[])

            num_steps += 

         .join(reasoning_trace)

     ():
        
        
        full_context = history + [node.step_text]

        
        hidden = .model.compute_hidden(full_context)

        
        next_step = .model.generate_from_hidden(hidden)

         {
            : next_step,
            : node,
            : hidden
        }