HumanOmniV2: 強制的なコンテキスト理解によるマルチモーダル推論の基盤化

ビジョン言語モデルは複雑なマルチモーダル理解タスクに苦労しています。画像内のオブジェクトを特定することはできますが、コンテキスト、感情、意図に関する推論に失敗します。これらは、表面的な特徴のパターンマッチングではなく、実際に見たものを理解することが必要とされる微妙な部分です。一般的な失敗パターンは「ショートカッティング」です。モデルが重要な視覚情報を無視して、一般的な事前知識に基づいて回答する現象です。

HumanOmniV2は、モデルが回答を生成する前に、マルチモーダルコンテキストの理解を明示的に述べることを要求することで、この問題に対処します。質問に直接回答する代わりに、モデルは最初に画像/動画/音声で観察した内容をまとめ、真の場面理解を強制します。これはショートカッティングを防ぎ、人間の意図と感情に関するより優れた下流推論を可能にします。

コアコンセプト

重要な洞察は、明示的なコンテキストの言語化がショートカッティングを防ぐということです。コンテキスト理解をスキップして直接回答に飛び込むことができるモデルは、もっともらしく聞こえるが不正確な応答を生成することがよくあります。次の要素を要求することで：

1. <context> タグ：モデルが観察したことの明示的な要約
2. <think> タグ：コンテキストを回答に接続する推論プロセス
3. <answer> タグ：推論に基づいて根拠のある最終応答

モデルは以下を強制されます：

• すべての関連する視覚情報を処理する
• 観察を推論ステップに接続する
• 根拠のあるロジックで回答を正当化する

この3段階の出力形式は、各段階を評価する専用の報酬モデルと組み合わせて、すべてのステップで説明責任を作成します。最終的な回答が妥当であっても、不正確なコンテキストはコンテキスト報酬によってペナルティを受けます。

アーキテクチャの概要

HumanOmniV2は、マルチモーダル大規模言語モデルをベースに構築されており、以下を含みます：

• ビジョンと音声エンコーダ：画像、動画、音声を同時に処理
• 3段階出力形式：コンテキスト要約 → 推論チェーン → 回答
• 複数報酬システム：コンテキスト精度、推論品質、フォーマット準拠、回答正確性のための個別の報酬
• GRPO トレーニング：各出力段階の専用報酬マスキングを伴うグループ相対ポリシー最適化
• IntentBench ベンチマーク：人間の意図と感情理解を強調する新しいデータセット（24K トレーニング、633 動画）

実装

ステップ1：3段階出力フォーマットフレームワークの作成

モデル生成を修正して、コンテキスト・推論・回答の構造を強制します。

import torch
import torch.nn as nn
from typing import Tuple

class ThreeStageOutputFormatter:
    """
    Enforces and parses three-stage output format:
    <context>...context...</context>
    <think>...reasoning...</think>
    <answer>...final answer...</answer>
    """

    def __init__(self, tokenizer):
        self.tokenizer = tokenizer

        # Add special tokens if not present
        special_tokens = ['<context>', '</context>', '<think>', '</think>',
                         '<answer>', '</answer>']
        self.tokenizer.add_special_tokens({'additional_special_tokens': special_tokens})

        self.context_start_id = self.tokenizer.encode('<context>')[1]
        self.context_end_id = self.tokenizer.encode('</context>')[1]
        self.think_start_id = self.tokenizer.encode('<think>')[1]
        self.think_end_id = self.tokenizer.encode('</think>')[1]
        self.answer_start_id = self.tokenizer.encode('<answer>')[1]
        self.answer_end_id = self.tokenizer.encode('</answer>')[1]

    def create_format_mask(self, seq_length, stage='context'):
        """
        Create a mask that enforces proper stage ordering.
        At step t, which tokens are valid?
        """
        mask = torch.zeros(seq_length, self.tokenizer.vocab_size, dtype=torch.bool)

        # Logic: enforce ordering of stages
        # This is a simplified version; real implementation uses constraints during generation
        return mask

    def parse_output(self, generated_text):
        """
        Extract context, reasoning, and answer from generated output.
        """
        try:
            # Extract context
            context_start = generated_text.find('<context>') + len('<context>')
            context_end = generated_text.find('</context>')
            context = generated_text[context_start:context_end].strip()

            # Extract thinking
            think_start = generated_text.find('<think>') + len('<think>')
            think_end = generated_text.find('</think>')
            thinking = generated_text[think_start:think_end].strip()

            # Extract answer
            answer_start = generated_text.find('<answer>') + len('<answer>')
            answer_end = generated_text.find('</answer>')
            answer = generated_text[answer_start:answer_end].strip()

            return {
                'context': context,
                'thinking': thinking,
                'answer': answer,
                'is_valid': all([context, thinking, answer])
            }
        except:
            return {
                'context': '',
                'thinking': '',
                'answer': '',
                'is_valid': False
            }

    def enforce_format_during_generation(self, model, prompt, max_length=512):
        """
        Generate with format enforcement: force proper stage transitions.
        """
        input_ids = self.tokenizer.encode(prompt, return_tensors='pt')
        current_stage = 'context'
        generated = []

        for step in range(max_length):
            # Get model prediction
            with torch.no_grad():
                outputs = model(input_ids)
                logits = outputs.logits[:, -1, :]

            # Apply format constraints
            # If in context stage, forbid think/answer tokens
            if current_stage == 'context':
                logits[0, self.think_start_id] = float('-inf')
                logits[0, self.answer_start_id] = float('-inf')

                # If we see </context>, transition
                if logits[0].argmax() == self.context_end_id:
                    current_stage = 'think'

            elif current_stage == 'think':
                logits[0, self.context_start_id] = float('-inf')
                logits[0, self.answer_start_id] = float('-inf')

                if logits[0].argmax() == self.think_end_id:
                    current_stage = 'answer'

            elif current_stage == 'answer':
                logits[0, self.context_start_id] = float('-inf')
                logits[0, self.think_start_id] = float('-inf')

            # Sample from constrained distribution
            probs = torch.softmax(logits, dim=-1)
            next_token = torch.multinomial(probs, num_samples=1)

            generated.append(next_token.item())
            input_ids = torch.cat([input_ids, next_token], dim=1)

            # Stop at </answer>
            if next_token.item() == self.answer_end_id:
                break

        return self.tokenizer.decode(generated)

ステップ2：各出力段階のための複数報酬システムの構築

コンテキスト、推論、正確性を個別に評価する個別の報酬モデルを作成します。

class MultiStageRewardSystem:
    """
    Evaluates each stage of the three-part output separately.
    Prevents any one stage from being ignored.
    """

    def __init__(self, device='cuda'):
        self.device = device

        # Separate reward models for each stage
        self.context_reward_model = self._build_context_reward_model()
        self.reasoning_reward_model = self._build_reasoning_reward_model()
        self.answer_reward_model = self._build_answer_reward_model()
        self.format_reward_model = self._build_format_reward_model()

    def _build_context_reward_model(self):
        """
        Reward model that scores how well the context captures visual information.
        Trained on examples where models correctly vs incorrectly summarize images.
        """
        # In practice, this would be a fine-tuned RoBERTa or similar
        # For demonstration, we show the structure
        return ContextRewardClassifier()

    def _build_reasoning_reward_model(self):
        """Scores whether reasoning logically follows from context."""
        return ReasoningRewardClassifier()

    def _build_answer_reward_model(self):
        """Scores answer correctness against ground truth."""
        return AnswerRewardClassifier()

    def _build_format_reward_model(self):
        """Scores whether output follows proper format."""
        return FormatRewardClassifier()

    def compute_stage_rewards(self, parsed_output, ground_truth, image):
        """
        Compute rewards for each stage.
        Returns: {'context': score, 'reasoning': score, 'answer': score, 'format': score}
        """
        rewards = {}

        # Context reward: does the context accurately describe the image?
        # Compare against other valid context descriptions
        context_score = self.context_reward_model.score(
            parsed_output['context'],
            image,
            ground_truth.get('reference_context', '')
        )
        rewards['context'] = context_score

        # Reasoning reward: does reasoning logically follow?
        # Score based on consistency and completeness
        reasoning_score = self.reasoning_reward_model.score(
            parsed_output['context'],
            parsed_output['thinking'],
            ground_truth.get('reference_reasoning', '')
        )
        rewards['reasoning'] = reasoning_score

        # Answer reward: is the answer correct?
        answer_score = self.answer_reward_model.score(
            parsed_output['answer'],
            ground_truth['answer']
        )
        rewards['answer'] = answer_score

        # Format reward: does output follow format?
        format_score = float(parsed_output['is_valid'])
        rewards['format'] = format_score

        return rewards

    def compute_combined_reward(self, stage_rewards, weights=None):
        """
        Combine stage rewards with learnable weights.
        Default: equal weight to all stages.
        """
        if weights is None:
            weights = {'context': 0.25, 'reasoning': 0.25, 'answer': 0.35, 'format': 0.15}

        combined = sum(stage_rewards[stage] * weights[stage]
                      for stage in stage_rewards)
        return combined

ステップ3：段階固有の学習のための因果マスキングを伴う GRPO の実装

GRPO を使用してトレーニングしますが、各段階が対応する報酬に対して最適化されるようにグラデーションをマスクします。

def grpo_training_with_stage_masking(model, tokenizer, training_data,
                                    reward_system, learning_rate=1e-5,
                                    num_steps=5000):
    """
    Train with GRPO where each output stage gets its own reward signal.
    Causal masking ensures context doesn't receive answer reward,
    and thinking doesn't receive format reward.
    """
    optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)
    formatter = ThreeStageOutputFormatter(tokenizer)

    for step in range(num_steps):
        # Sample batch
        batch = random.sample(training_data, batch_size=16)

        stage_losses = {'context': [], 'reasoning': [], 'answer': []}

        for example in batch:
            image = example['image']
            prompt = example['question']
            ground_truth = example['answer_data']  # context, reasoning, answer

            # Generate response
            input_ids = tokenizer.encode(prompt, return_tensors='pt')

            # Generate with format enforcement
            generated_text = formatter.enforce_format_during_generation(
                model, prompt, max_length=512
            )

            # Parse output
            parsed = formatter.parse_output(generated_text)

            # Skip if format is invalid
            if not parsed['is_valid']:
                continue

            # Compute rewards for each stage
            stage_rewards = reward_system.compute_stage_rewards(
                parsed, ground_truth, image
            )

            # Convert to advantage: reward - mean baseline
            baseline_reward = sum(stage_rewards.values()) / len(stage_rewards)

            # Compute log probability for each stage separately
            for stage_name in ['context', 'reasoning', 'answer']:
                # Extract the specific stage tokens
                start_token = formatter.__dict__[f'{stage_name}_start_id']
                end_token = formatter.__dict__[f'{stage_name}_end_id']

                # Find token positions for this stage
                start_pos = (input_ids == start_token).nonzero()[0, 0].item()
                end_pos = (input_ids == end_token).nonzero()[0, 0].item()

                # Compute log prob only for this stage
                # (don't apply reward from other stages to this stage)
                stage_ids = input_ids[:, start_pos:end_pos + 1]

                with torch.no_grad():
                    outputs = model(stage_ids)
                    logits = outputs.logits

                log_probs = F.log_softmax(logits, dim=-1)
                # Extract log probs for actual tokens
                actual_log_probs = log_probs[0, range(stage_ids.shape[1]), stage_ids[0]]

                # GRPO: advantage-weighted log probability
                advantage = stage_rewards[stage_name] - baseline_reward
                stage_loss = -(actual_log_probs.sum() * advantage)

                stage_losses[stage_name].append(stage_loss)

        # Combine losses
        total_loss = (sum(stage_losses['context']) +
                     sum(stage_losses['reasoning']) +
                     sum(stage_losses['answer'])) / (3 * len(batch))

        optimizer.zero_grad()
        total_loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
        optimizer.step()

        if step % 100 == 0:
            print(f"Step {step}: Loss {total_loss:.4f}")

    return model

ステップ4：IntentBench と汎用マルチモーダルベンチマークで評価

IntentBench データセット（人間の意図/感情理解に焦点）と従来のベンチマークでテストします。

def evaluate_omni_modal_reasoning(model, tokenizer, test_set_intentbench,
                                 test_set_general):
    """
    Evaluate on both IntentBench (human intention focus) and general benchmarks.
    Metrics: accuracy, context quality, reasoning quality.
    """
    formatter = ThreeStageOutputFormatter(tokenizer)

    results = {
        'intentbench': evaluate_on_benchmark(model, formatter, tokenizer,
                                            test_set_intentbench),
        'general': evaluate_on_benchmark(model, formatter, tokenizer,
                                        test_set_general)
    }

    return results

def evaluate_on_benchmark(model, formatter, tokenizer, test_set):
    """
    Evaluate on a specific benchmark.
    Metrics: answer accuracy, context quality, reasoning coherence.
    """
    correct_answers = 0
    context_quality_scores = []
    reasoning_quality_scores = []
    total = 0

    model.eval()

    for example in test_set:
        with torch.no_grad():
            prompt = example['prompt']
            image = example['image']
            ground_truth_answer = example['answer']

            # Generate
            generated = formatter.enforce_format_during_generation(
                model, prompt, max_length=512
            )

            parsed = formatter.parse_output(generated)

            # Evaluate answer correctness
            is_correct = (parsed['answer'].strip().lower() ==
                         ground_truth_answer.strip().lower())
            correct_answers += int(is_correct)

            # Evaluate context (does it capture key visual elements?)
            context_quality = evaluate_context_against_image(
                parsed['context'], image
            )
            context_quality_scores.append(context_quality)

            # Evaluate reasoning (coherent and grounded?)
            reasoning_quality = evaluate_reasoning_coherence(
                parsed['context'], parsed['thinking']
            )
            reasoning_quality_scores.append(reasoning_quality)

            total += 1

    return {
        'accuracy': correct_answers / total,
        'context_quality': np.mean(context_quality_scores),
        'reasoning_quality': np.mean(reasoning_quality_scores)
    }

実践的なガイダンス

ハイパーパラメータ	推奨値	注記
コンテキスト報酬ウェイト	0.25	推論と同等。合わせて総合の50%
回答報酬ウェイト	0.35	最も重要。正確な回答が重要
フォーマット報酬ウェイト	0.15	構造を強制。ショートカッティングを防止
ステージマスキング	厳密	回答報酬をコンテキストトークンに適用しない
トレーニングエポック	3-5	GRPO は通常 SFT より高速に収束
RL バッチサイズ	16-32	より大きいサイズは優位性推定の安定性を向上

HumanOmniV2 トレーニングを使用すべき場合：

• 視覚シーンを深く理解する必要があるモデルが必要な場合（表面的なレベルではなく）
• ショートカッティング動作を防ぎたい場合
• 人間の意図、感情、またはコンテキストに関する推論が必要なタスクの場合
• マルチモーダルデータ（画像、動画、音声を含む）がある場合

HumanOmniV2 を使用すべきでない場合：

• タスクが単純なオブジェクト認識である場合（過剰）
• トレーニングするための明示的なコンテキスト/推論のグラウンドトゥルースがない場合
• 推論レイテンシが重要である場合（3段階生成は遅い）
• 最終的な回答の正確性だけが重要である場合（コンテキスト/推論は不要）

一般的な落とし穴：

• ステージ順序の崩壊：モデルが何も言う前に </context> を生成する場合、フォーマットマスクが十分に強くありません。確率調整だけでなく、トークンレベルの厳密な制約を使用してください。
• コンテキスト報酬が寛容すぎる：モデルは妥当だが不正確なコンテキストを書くことを学びます。人間注釈からの参考コンテキストを使用し、主要な視覚要素の欠落にペナルティを課してください。
• 推論と回答の不整合：推論が X と言っているが回答が Y と言っている場合、両方のステージ報酬が失敗する必要があります。reasoning_reward_model が一貫性をチェックすることを確認してください。
• GRPO を使用した学習の不安定性：優位性推定が大きく変動する場合、バッチサイズを増やすか、ステージ固有の報酬スケーリングを減らしてください。

参考文献

HumanOmniV2: From Understanding to Omni-Modal Reasoning with Context https://arxiv.org/abs/2506.21277