DLM Integration Quick Reference

# Location: packages/dlm/response/embedding_provider.py
# Method: _apply_ircp_weighting()

def _apply_ircp_weighting(
    self, similarity: float, emb1: np.ndarray, emb2: np.ndarray
) -> float:
    """Apply I-RCP coordinate weighting to similarity score."""
    # Check if embedder supports coordinate prediction
    if not hasattr(self, 'ircp_embedder') or self.ircp_embedder is None:
        return similarity

    try:
        # Predict IRCP coordinates for both embeddings
        coords1 = self.ircp_embedder.predict_coordinates(emb1)
        coords2 = self.ircp_embedder.predict_coordinates(emb2)

        # Calculate coordinate distance
        coord_distance = np.linalg.norm(coords1 - coords2)

        # Apply weighting: closer coordinates = higher similarity boost
        # Formula: weight = 1 / (1 + distance)
        ircp_weight = 1.0 / (1.0 + coord_distance)

        # Blend base similarity with IRCP weighting
        # 70% base similarity + 30% IRCP boost
        weighted_similarity = similarity * (0.7 + 0.3 * ircp_weight)

        # Ensure result is in [0, 1]
        return float(np.clip(weighted_similarity, 0.0, 1.0))

    except Exception as e:
        logger.warning(f"IRCP weighting failed: {e}, using base similarity")
        return similarity

# Location: packages/dlm/response/embedding_provider.py
# Method: _apply_temporal_weighting()

def _apply_temporal_weighting(
    self, similarity: float, timestamp1: Optional[float] = None,
    timestamp2: Optional[float] = None
) -> float:
    """Apply temporal weighting to similarity score."""
    if timestamp1 is None or timestamp2 is None:
        return similarity

    try:
        # Calculate temporal distance
        time_diff = abs(timestamp1 - timestamp2)

        # Apply exponential decay
        # Formula: decay = exp(-λ * time_diff)
        # Where λ is temporal decay factor (default: 0.1)
        decay_factor = getattr(self, 'temporal_decay_factor', 0.1)
        temporal_decay = np.exp(-decay_factor * time_diff)

        # Apply temporal weighting
        # More recent messages get higher similarity
        weighted_similarity = similarity * temporal_decay

        return float(np.clip(weighted_similarity, 0.0, 1.0))

    except Exception as e:
        logger.warning(f"Temporal weighting failed: {e}, using base similarity")
        return similarity

# Location: packages/dlm/core/rcp/context_propagator.py (TO BE CREATED)

from typing import Dict, List, Optional, Any
import numpy as np
import torch
from dlm.core.coordinates import DLMCoordinate
from dlm.core.ircp import IRCP_AVAILABLE

if IRCP_AVAILABLE:
    from ircp.core.ring_topology import RingTopology
    from ircp.core.attention_mechanism import ContextualAttention
    from ircp.core.flow_dynamics import FlowDynamics

class RCPContextPropagator:
    """
    RCP Context Propagator for DLM

    Integrates RCP forward propagation (dC/dt = A(C)C) into DLM pipeline.
    """

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}

        if not IRCP_AVAILABLE:
            raise ImportError("IRCP package required for RCP integration")

        # Initialize RCP components
        self.ring_topology = RingTopology(
            ring_construction_method=self.config.get("ring_method", "hierarchical_preserving"),
            max_ring_size=self.config.get("max_ring_size", 50)
        )

        self.attention_mechanism = ContextualAttention(
            coordinate_weights=(
                self.config.get("alpha", 1.0),
                self.config.get("beta", 1.0),
                self.config.get("gamma", 1.0)
            )
        )

        self.flow_dynamics = FlowDynamics(
            context_dim=self.config.get("context_dim", 768),
            flow_method=self.config.get("flow_method", "euler"),
            dt=self.config.get("dt", 0.1)
        )

    def build_ring_topology(
        self,
        messages: List[Dict[str, Any]],
        coordinates: Dict[str, DLMCoordinate]
    ) -> Any:
        """Build ring topology from messages and coordinates."""
        # Convert DLM coordinates to RCP spatial coordinates
        spatial_coords = {}
        for msg_id, dlm_coord in coordinates.items():
            spatial_coords[msg_id] = SpatialCoordinates(
                x=dlm_coord.x,
                y=dlm_coord.y,
                z=dlm_coord.z
            )

        # Build ring topology
        rings = self.ring_topology.build_ring_from_coordinates(
            coordinates=spatial_coords,
            messages={msg["id"]: msg for msg in messages}
        )

        return rings

    def compute_attention(
        self,
        current_message: Dict[str, Any],
        ring_topology: Any,
        coordinates: DLMCoordinate
    ) -> np.ndarray:
        """Compute forward attention weights."""
        # Extract coordinate tensor
        coord_tensor = torch.tensor(
            [[coordinates.x, coordinates.y, coordinates.z]],
            dtype=torch.float32
        )

        # Compute attention (simplified - full implementation would use ring topology)
        attention_weights = self.attention_mechanism(
            coordinates=coord_tensor,
            embeddings=None  # Would use actual embeddings
        )

        return attention_weights.weights.numpy()

    def propagate_context(
        self,
        initial_context: np.ndarray,
        attention_matrix: np.ndarray,
        coordinates: DLMCoordinate
    ) -> np.ndarray:
        """Propagate context using dC/dt = A(C)C."""
        # Convert to tensors
        context_tensor = torch.tensor(
            initial_context.reshape(1, 1, -1),
            dtype=torch.float32
        )

        coord_tensor = torch.tensor(
            [[coordinates.x, coordinates.y, coordinates.z]],
            dtype=torch.float32
        )

        # Solve differential equation
        with torch.no_grad():
            flow_result = self.flow_dynamics(
                initial_context=context_tensor,
                coordinates=coord_tensor.unsqueeze(0)
            )

        # Extract final context
        final_context = flow_result.context_vectors.squeeze().numpy()

        return final_context

# Location: packages/dlm/core/tpo/path_analyzer.py (TO BE CREATED)

from typing import Dict, List, Optional, Any
import numpy as np
from dlm.core.coordinates import DLMCoordinate

class TPOPathAnalyzer:
    """
    TPO Path Analyzer for DLM

    Analyzes conversation paths and calculates quality scores Q(P).
    """

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}

        # Quality weights
        self.weights = {
            "linearity": self.config.get("linearity_weight", 0.4),
            "terminal": self.config.get("terminal_weight", 0.3),
            "semantic": self.config.get("semantic_weight", 0.2),
            "completion": self.config.get("completion_weight", 0.1)
        }

    def extract_paths(
        self,
        conversation_graph: Any,  # Would use actual graph structure
        current_message: Dict[str, Any]
    ) -> List[Dict[str, Any]]:
        """Extract all paths from conversation graph."""
        # Implementation would extract root-to-leaf paths
        # For now, return simplified structure
        paths = []
        # TODO: Implement actual path extraction
        return paths

    def calculate_quality(
        self,
        path: Dict[str, Any],
        coordinates: Dict[str, DLMCoordinate],
        rcp_context_flow: Optional[np.ndarray] = None,
        ircp_patterns: Optional[np.ndarray] = None
    ) -> float:
        """
        Calculate path quality: Q(P) = α·L(P) + β·T(P) + γ·S(P) + δ·C(P)
        """
        # Linearity score
        linearity = self._calculate_linearity(path)

        # Terminal quality
        terminal = self._calculate_terminal_quality(path, coordinates)

        # Semantic coherence
        semantic = self._calculate_semantic_coherence(path, coordinates)

        # Completion quality
        completion = self._calculate_completion_quality(path)

        # Calculate weighted sum
        quality = (
            self.weights["linearity"] * linearity +
            self.weights["terminal"] * terminal +
            self.weights["semantic"] * semantic +
            self.weights["completion"] * completion
        )

        # Optional: Enhance with RCP and IRCP
        if rcp_context_flow is not None:
            # Add RCP context flow quality
            quality += 0.1 * self._calculate_rcp_quality(rcp_context_flow)

        if ircp_patterns is not None:
            # Add IRCP pattern compliance
            quality += 0.1 * self._calculate_ircp_compliance(ircp_patterns)

        return float(np.clip(quality, 0.0, 1.0))

    def _calculate_linearity(self, path: Dict[str, Any]) -> float:
        """Calculate linearity score L(P)."""
        # Implementation: exp(-λ * Σ max(0, |children(vᵢ)| - 1))
        # Simplified for now
        return 1.0

    def _calculate_terminal_quality(
        self, path: Dict[str, Any], coordinates: Dict[str, DLMCoordinate]
    ) -> float:
        """Calculate terminal quality T(P)."""
        # Implementation: (1/4)(D(vₖ) + Z(vₖ) + N(vₖ) + τ(vₖ))
        # Simplified for now
        return 0.5

    def _calculate_semantic_coherence(
        self, path: Dict[str, Any], coordinates: Dict[str, DLMCoordinate]
    ) -> float:
        """Calculate semantic coherence S(P)."""
        # Implementation: (1/|P|-1) * Σ coherence(vᵢ, vᵢ₊₁)
        # Simplified for now
        return 0.5

    def _calculate_completion_quality(self, path: Dict[str, Any]) -> float:
        """Calculate completion quality C(P)."""
        # Implementation: |P| / (|P| + B(P))
        # Simplified for now
        return 0.5

    def _calculate_rcp_quality(self, context_flow: np.ndarray) -> float:
        """Calculate RCP context flow quality."""
        # Measure how well context flows
        flow_magnitude = np.linalg.norm(context_flow)
        return float(np.clip(flow_magnitude / 10.0, 0.0, 1.0))

    def _calculate_ircp_compliance(self, patterns: np.ndarray) -> float:
        """Calculate IRCP pattern compliance."""
        # Measure how well patterns match
        pattern_strength = np.linalg.norm(patterns)
        return float(np.clip(pattern_strength / 10.0, 0.0, 1.0))

1. Message Input
   ↓
2. IRCP Embedding + Coordinates
   ↓
3. DLM Coordinate (Enhanced)
   ↓
4. RCP Context Propagation
   ↓
5. TPO Path Quality
   ↓
6. Enhanced Similarity
   ↓
7. Context Retrieval
   ↓
8. Response Generation