async def parallel_reason(
    query: str,
    config: HeavySkillConfig,
    agent: OpenAICompatibleAgent = None,
) -> List[str]:
    """生成 K 条独立的推理轨迹"""
    trajectories = await agent.generate(
        prompt=query,
        n=config.reason_k,  # K 值：并行轨迹数量
        temperature=config.reason_temperature,
        max_tokens=config.reason_max_tokens,
        top_p=config.reason_top_p,
    )
    # 过滤掉重复/退化的回答
    valid = [t for t in trajectories if not duplication(t, think_end_tag)]
    return valid

async def deliberate(
    query: str,
    cache: MemoryCache,
    config: HeavySkillConfig,
    agent: Optional[OpenAICompatibleAgent] = None,
    trajectory_selection: str = "max_diversity",
) -> List[str]:
    """对推理轨迹进行批判性分析"""
    # 从缓存中选择轨迹（支持多种选择策略）
    selected = cache.select(k=config.reason_k, strategy=trajectory_selection)

    # 如果超出 token 预算，进行截断
    selected = clip_trajectories(selected, config.max_trajectory_tokens)

    # 构建深思提示词
    prompt = build_summary_prompt(
        query=query,
        trajectories=selected,
        prompt_type=config.prompt_type,
        language=config.language,
    )

    # 生成 summary_k 个深思输出
    summaries = await agent.generate(
        prompt=prompt,
        n=config.summary_k,
        temperature=config.summary_temperature,
        max_tokens=config.summary_max_tokens,
    )
    return summaries

def split_response(
    response: str,
    think_start_tag: str = "<think>",
    think_end_tag: str = "\n</think>\n\n"
) -> tuple:
    """Split a response into thinking content and answer content."""
    if think_end_tag in response:
        parts = response.split(think_end_tag)
        think_content = parts[0].replace(think_start_tag, "").strip()
        answer_content = parts[-1].strip()
    else:
        think_content = response.replace(think_start_tag, "").strip()
        answer_content = ""
    return think_content, answer_content

def clip_trajectories(
    trajectories: List[str],
    max_total_tokens: int,
    estimate_fn=None,
) -> List[str]:
    """Clip trajectories from the beginning if total tokens exceed the budget."""
    if estimate_fn is None:
        estimate_fn = estimate_token_count

    token_counts = [estimate_fn(t) for t in trajectories]
    total = sum(token_counts)

    if total <= max_total_tokens:
        return trajectories

    excess = total - max_total_tokens
    cut_per_trajectory = excess // len(trajectories) + 1

    clipped = []
    for text, count in zip(trajectories, token_counts):
        if count <= cut_per_trajectory:
            clipped.append(text)
            continue
        ratio = cut_per_trajectory / count
        char_cut = int(len(text) * ratio)
        clipped.append(text[char_cut:])

    return clipped

def duplication(response, think_end_tag):
    # 如果模型出现复读导致截断的问题
    if response[-512:] == response[-1024:-512]:
        return True
    # 如果输出存在超长截断的问题，也忽略
    if think_end_tag not in response:
        return True
    return False

git clone https://github.com/wjn1996/HeavySkill.git
cd HeavySkill
pip install -e .

python scripts/run_heavyskill.py \
    --query "Find the number of paths of length 16 on an 8x8 grid that change direction exactly four times." \
    --model "deepseek-r1" \
    --api_base "http://localhost:8080" \
    --reason_k 8 \
    --summary_k 4 \
    --prompt_type "stem" \
    --output "outputs/result.json" \
    --verbose

python scripts/run_heavyskill.py \
    --query "Your problem here" \
    --model "r1-distill-qwen-7b" \
    --api_base "http://localhost:8080" \
    --summary_model "qwen3-32b" \
    --summary_api_base "http://localhost:8081" \
    --reason_k 16 \
    --summary_k 4

python scripts/run_heavyskill.py \
    --input_file "examples/example_math.json" \
    --model "deepseek-r1" \
    --api_base "http://localhost:8080" \
    --output "outputs/batch_result.json"

cp skill/heavyskill.md ~/.claude/skills/heavyskill.md

config = HeavySkillConfig(
    model_name="deepseek-reasoner",
    api_base="https://api.deepseek.com",
    api_key="your-api-key",
    reason_k=8,
    summary_k=4,
)

def _max_answer_num_select(self, k: int) -> List[str]:
    """选择答案出现次数最多的轨迹"""
    from collections import Counter
    answers = [extract_boxed_answer(t) or t[-200:] for t in self.trajectories]
    counter = Counter(answers)
    # 返回答案出现次数最多的轨迹
    answer_counts = [(t, counter[extract_boxed_answer(t) or t[-200:]]) for t in self.trajectories]
    answer_counts.sort(key=lambda x: x[1], reverse=True)
    return [t for t, _ in answer_counts[:k]]

PROMPT_REGISTRY = {
    "general": {"en": SUMMARY_PROMPT_GENERAL_EN, "cn": SUMMARY_PROMPT_GENERAL_CN},
    "stem": {"en": SUMMARY_PROMPT_STEM, "cn": SUMMARY_PROMPT_STEM},
}

def build_custom_prompt(
    query: str,
    trajectories: List[str],
    prompt_type: str = "general",
    language: str = "en",
) -> str:
    if prompt_type == "custom":
        return CUSTOM_TEMPLATE.format(
            problem=query,
            response_prompt=format_thinkers_prompt(trajectories)
        )
    return build_summary_prompt(query, trajectories, prompt_type, language)

python scripts/evaluate.py \
    --result_file "outputs/result.json" \
    --target_file "path/to/targets.json"

@article{wang2026heavyskill,
  title={HeavySkill: Heavy Thinking as the Inner Skill in Agentic Harness},
  author={Wang, Jianing and Guo, Linsen and Chen, Zhengyu and Guo, Qi and Zang, Hongyu and Shi, Wenjie and Ma, Haoxiang and Xi, Xiangyu and Li, Xiaoyu and Wang, Wei and Cai, Xunliang},
  journal={arXiv preprint arXiv:2605.02396},
  year={2026},
  url={https://arxiv.org/abs/2605.02396}
}