SkillRL: 再帰的スキル拡張強化学習

ポリシーが過去の知見を繰り返し再発見しなければならない場合、エージェントのパフォーマンスは停滞します。SkillRLはこの問題に対処するため、相互作用履歴から行動パターンを自動抽出し、コンパクトで再利用可能なスキルに蒸留して、将来の意思決定を導きます。生トラジェクトリを保存する代わりに、このシステムは戦略的パターンと失敗の教訓を蒸留し、エージェントとともに成長するスキルライブラリを作成します。

コアコンセプト

SkillRLはトラジェクトリを差別的に処理します：

• 成功エピソード → 戦略的パターンを抽出（生トラジェクトリ比で10～20倍の圧縮）
• 失敗エピソード → 何が問題だったかを捉える失敗の教訓を抽出

スキルは階層的に組織されます：汎用スキル（探索、状態管理）とタスク固有スキル。意思決定時に、エージェントはセマンティック類似度を介して関連するスキルを取得し、コンテキストのオーバーヘッドを削減しながら推論品質を維持します。

スキルライブラリは再帰的に進化します：検証エポック後、失敗モードから新しいスキルが生成されたり既存スキルが改善されたりして、改善されたポリシーが新しい課題に遭遇する好循環が生まれます。

アーキテクチャ概要

• エクスペリエンス処理：成功トラジェクトリ（パターン抽出）と失敗トラジェクトリ（教訓抽出）を分離
• スキルライブラリ（SkillBank）：2段階組織—汎用スキル（汎用）とタスク固有スキル
• セマンティック検索：埋め込み類似度を用いて現在の状態に関連するスキルを取得
• 動的進化：失敗モードを分析して新しいスキルを生成または既存スキルを更新
• 統合：ポリシーロールアウト時にスキルをコンテキストの先頭に追加

実装

トラジェクトリを高レベルパターンを抽出してスキルに処理します：

def extract_skills_from_trajectory(trajectory, success=True):
    """Extract skills from a trajectory (success or failure)."""
    if success:
        # For successful trajectories: extract strategic patterns
        # E.g., "When encountering X, use strategy Y"
        skill = {
            'type': 'strategic',
            'condition': identify_key_decision_points(trajectory),
            'action_pattern': abstract_action_sequence(trajectory),
            'outcome': 'success'
        }
    else:
        # For failures: extract lessons
        # E.g., "Avoid X because it leads to Y failure"
        failure_point = identify_failure_point(trajectory)
        skill = {
            'type': 'lesson',
            'condition': failure_point['state'],
            'anti_action': failure_point['action_taken'],
            'failure_mode': failure_point['reason'],
            'outcome': 'failure'
        }
    return skill

def compress_skill_text(skill, max_tokens=100):
    """Compress skill into concise text representation."""
    if skill['type'] == 'strategic':
        return f"When {skill['condition']}, {skill['action_pattern']} → success"
    else:
        return f"Avoid {skill['anti_action']} in {skill['condition']} (→ {skill['failure_mode']})"

# Build skill library from trajectory batch
skill_library = {'general': [], 'task_specific': {}}
for traj in trajectories:
    if traj['success']:
        skill = extract_skills_from_trajectory(traj, success=True)
        skill_library['general'].append(compress_skill_text(skill))
    else:
        skill = extract_skills_from_trajectory(traj, success=False)
        task = traj['task_id']
        if task not in skill_library['task_specific']:
            skill_library['task_specific'][task] = []
        skill_library['task_specific'][task].append(compress_skill_text(skill))

埋め込み類似度を介して関連するスキルを取得します：

import torch
import torch.nn.functional as F

def retrieve_relevant_skills(current_state, skill_library, embedding_model, top_k=5):
    """Retrieve top-k skills relevant to current state."""
    # Embed current state
    state_embedding = embedding_model.encode(current_state)

    # Embed all skills
    all_skills = skill_library['general'] + sum(
        skill_library['task_specific'].values(), []
    )
    skill_embeddings = embedding_model.encode(all_skills)

    # Compute similarities
    similarities = F.cosine_similarity(
        state_embedding.unsqueeze(0),
        skill_embeddings,
        dim=-1
    )

    # Get top-k
    top_indices = torch.topk(similarities, min(top_k, len(all_skills)))[1]
    relevant_skills = [all_skills[i] for i in top_indices]

    return relevant_skills

# During rollout
def policy_forward_with_skills(state, policy, skill_library, embedding_model):
    """Generate action using policy augmented with relevant skills."""
    # Retrieve skills
    skills = retrieve_relevant_skills(state, skill_library, embedding_model, top_k=5)
    skill_context = "\n".join([f"- {s}" for s in skills])

    # Augment prompt with skills
    augmented_prompt = f"Skills:\n{skill_context}\n\nState: {state}\nAction:"

    # Generate action
    action = policy.generate(augmented_prompt, max_tokens=50)
    return action

失敗を分析してスキルライブラリを進化させます：

def analyze_failures_and_evolve_skills(trajectories, skill_library, embedding_model):
    """Analyze failures to generate new skills."""
    failed_trajectories = [t for t in trajectories if not t['success']]

    for failed_traj in failed_trajectories:
        # Extract failure lesson
        failure_skill = extract_skills_from_trajectory(failed_traj, success=False)
        skill_text = compress_skill_text(failure_skill)

        # Check if similar skill already exists
        existing_skills = skill_library['general'] + \
            sum(skill_library['task_specific'].values(), [])
        existing_embeddings = embedding_model.encode(existing_skills)
        new_embedding = embedding_model.encode(skill_text)

        similarities = F.cosine_similarity(
            new_embedding.unsqueeze(0),
            existing_embeddings,
            dim=-1
        )

        # If no similar skill exists, add it
        if similarities.max() < 0.8:
            task = failed_traj['task_id']
            if task not in skill_library['task_specific']:
                skill_library['task_specific'][task] = []
            skill_library['task_specific'][task].append(skill_text)

    return skill_library

実践的ガイダンス

コンポーネント	推奨事項	注釈
スキル抽出	成功と失敗の両方	どちらも価値があります。失敗は反パターンを防ぎます。
スキル検索 top-k	3～7	コンテキスト長とカバレッジのバランスを取ります。
類似度閾値	0.75～0.85	近似重複スキルの保存を避けます。
進化頻度	各検証エポック後	タイムリーなスキル更新がポリシー適応を有効にします。

使用すべき場合

• エージェントがパターンを記憶する必要がある長期的タスク（Webナビゲーション、ダイアログ）
• 汎用スキルがドメイン全体に適用される多タスク学習
• 成功/失敗の区別が明確なドメイン

使用しない場合

• トラジェクトリの再利用がない単一エピソードタスク
• すべてのトラジェクトリが一意なドメイン（低パターン再利用性）

参考文献

スキルライブラリ管理、埋め込みモデル、ALFWorld家事タスクの検証を含む完全な実装については、https://arxiv.org/abs/2602.08234 を参照してください。