cat_v3/fusion.py

"""Concept Fusion Layer module for CAT V3."""

from __future__ import annotations

from typing import Any, Dict, List
import torch
import torch.nn as nn


class ConceptFusionLayer(nn.Module):
    """Fuses path predictions and embeddings from multiple active GAT experts."""

    def __init__(self, num_concepts: int, pad_id: int, eos_id: int, top_m: int = 8) -> None:
        super().__init__()
        self.num_concepts = num_concepts
        self.pad_id = pad_id
        self.eos_id = eos_id
        self.top_m = top_m

    def forward(
        self,
        expert_reports: Dict[str, Dict[str, torch.Tensor]],
        router_probs: torch.Tensor,
        router_mask: torch.Tensor,
        global_embeddings: torch.Tensor,
        domain_names: List[str],
    ) -> Dict[str, torch.Tensor]:
        """Fuses expert reports into a unified tensor representation.
        
        Args:
            expert_reports: Dict mapping domain -> GATExpert output dict
            router_probs: [batch_size, num_experts] tensor of router probabilities
            router_mask: [batch_size, num_experts] boolean mask of active experts
            global_embeddings: [num_concepts, concept_dim] concept embedding table
            domain_names: List of domain names mapping to indices in router outputs
            
        Returns:
            Dict containing:
                fused_concept_ids: [batch_size, top_m]
                fused_embeddings: [batch_size, top_m, concept_dim]
                fused_scores: [batch_size, num_concepts]
                top_scores: [batch_size, top_m]
        """
        batch_size = router_probs.size(0)
        device = router_probs.device
        path_len = next(iter(expert_reports.values()))["predicted_path"].size(1)

        # Accumulator for scores of each concept: [batch_size, num_concepts]
        fused_scores = torch.zeros(batch_size, self.num_concepts, device=device)

        for idx, domain in enumerate(domain_names):
            report = expert_reports[domain]
            pred_path = report["predicted_path"]  # [batch_size, path_length]
            path_scores = report["path_scores"]   # [batch_size, path_length] (log probabilities)
            path_probs = torch.exp(path_scores)   # [batch_size, path_length] (probabilities)

            # Mask and weight based on router probability for this domain
            # active_weight: [batch_size]
            active_weight = router_probs[:, idx] * router_mask[:, idx].float()

            # Scatter probabilities into the concept accumulator for each step
            for step in range(path_len):
                step_concepts = pred_path[:, step]  # [batch_size]
                step_probs = path_probs[:, step]    # [batch_size]

                # Weight step probability by expert activation weight
                weighted_probs = step_probs * active_weight  # [batch_size]
                fused_scores.scatter_add_(1, step_concepts.unsqueeze(1), weighted_probs.unsqueeze(1))

        # Clear scores for special pad and eos tokens to avoid selecting them
        fused_scores[:, self.pad_id] = 0.0
        fused_scores[:, self.eos_id] = 0.0

        # Retrieve the top-M most activated concepts
        scores, fused_concept_ids = torch.topk(fused_scores, k=self.top_m, dim=-1)

        # Retrieve unified concept embeddings: [batch_size, top_m, concept_dim]
        fused_embeddings = global_embeddings[fused_concept_ids]

        return {
            "fused_concept_ids": fused_concept_ids,
            "fused_embeddings": fused_embeddings,
            "fused_scores": fused_scores,
            "top_scores": scores
        }

    def get_symbolic_report(
        self,
        vocab: Any,
        expert_reports: Dict[str, Dict[str, torch.Tensor]],
        router_mask: torch.Tensor,
        domain_names: List[str],
    ) -> List[Dict[str, Any]]:
        """Converts raw tensor results into human-readable concept graphs and paths.
        
        Args:
            vocab: ConceptVocabulary instance
            expert_reports: Dict of expert outputs
            router_mask: [batch_size, num_experts]
            domain_names: List of domains
            
        Returns:
            List of dicts, one per batch item:
                {
                    "concepts": List[str],
                    "reasoning_paths": List[List[str]],
                    "confidence": List[float]
                }
        """
        batch_size = router_mask.size(0)
        symbolic_reports = []

        for b in range(batch_size):
            concepts_set = set()
            reasoning_paths = []
            confidences = []

            for idx, domain in enumerate(domain_names):
                if router_mask[b, idx].item():
                    report = expert_reports[domain]
                    path_ids = report["predicted_path"][b].tolist()
                    path = vocab.decode_path(path_ids)
                    if path:
                        reasoning_paths.append(path)
                        for c in path:
                            concepts_set.add(c)
                        
                        # Mean path confidence calculation
                        path_scores = report["path_scores"][b].tolist()
                        path_probs = [torch.exp(torch.tensor(s)).item() for s in path_scores[:len(path)]]
                        mean_prob = sum(path_probs) / max(len(path_probs), 1)
                        confidences.append(mean_prob)

            symbolic_reports.append({
                "concepts": sorted(list(concepts_set)),
                "reasoning_paths": reasoning_paths,
                "confidence": confidences
            })

        return symbolic_reports