Каузальні залежності у World Models: від кореляції до причинності

«Півень кукурікає — сонце встає». Кореляція? Так. Причинність? Ні. Півень не змушує сонце вставати.

Сучасні нейронні мережі — майстри кореляцій. Вони знаходять патерни в даних, які людина ніколи б не помітила. Але вони не розуміють причинності. Вони не знають, що сонце встало б і без півня. Вони не можуть відповісти на питання: «А що буде, якщо півень не закукурікає?»

Для world models це критично. Бо якщо модель не розуміє причинність — вона не може відповісти на питання «Що буде, ЯКЩО я зроблю X?». А саме це питання — суть автономних систем, робототехніки та прийняття рішень.

Чому кореляція ≠ причинність: реальні приклади

Приклад 1: Робототехніка

Робот спостерігає: коли людина махає рукою, двері відчиняються.

Спостереження:
  wave_hand → doors_open (кореляція = 0.95)

Робот робить висновок:
  махни рукою → двері відкриються

Робот пробує: махає рукою. Двері не відчиняються.

Справжня причина:
  людина натискала кнопку (яку робот не бачив)
  рух руки — побічний ефект руху до кнопки

Модель, яка вивчила кореляцію — провалилась. Модель, яка зрозуміла б причинність — шукала б кнопку.

Приклад 2: Автопілот

Дані: коли попереду є тінь, машина гальмує
Кореляція: shadow → brake

Реальність: тінь від пішохідного переходу
Причина: не тінь, а перехід (і потенційні пішоходи)

Проблема: в новому місті тінь від дерева
Модель гальмує без причини

Приклад 3: Медичний AI

Дані: пацієнти з кисневою маскою частіше помирають
Кореляція: oxygen_mask → death (позитивна!)

Реальність: кисневу маску дають важким пацієнтам
Причина: не маска вбиває, а важкий стан

Висновок моделі: "не давати кисень" — небезпечний!

Три рівні каузального reasoning (Judea Pearl)

Pearl's Ladder of Causation:

Рівень 3: COUNTERFACTUAL (Уявляти)
         "Що БУЛО Б, якби...?"
         P(Y_x' | X=x, Y=y)
              ▲
              │
Рівень 2: INTERVENTION (Робити)
         "Що СТАНЕТЬСЯ, якщо я зроблю...?"
         P(Y | do(X=x))
              ▲
              │
Рівень 1: ASSOCIATION (Бачити)
         "Що я можу спостерігати?"
         P(Y | X)

Рівень 1 — Association:

# Більшість ML працює тут
P(Y | X) = P(X, Y) / P(X)

# Питання: "Яка ймовірність, що пацієнт одужає,
#           якщо ми БАЧИМО, що він приймає ліки?"

# Проблема: selection bias
# Ті, хто приймають ліки — можуть бути здоровішими спочатку

Рівень 2 — Intervention:

# do-calculus
P(Y | do(X=x)) ≠ P(Y | X=x)

# Питання: "Яка ймовірність, що пацієнт одужає,
#           якщо ми ПРИЗНАЧИМО йому ліки?"

# Це interventional — ми активно змінюємо X
# Не просто спостерігаємо

Рівень 3 — Counterfactual:

# Питання: "Чи одужав би цей конкретний пацієнт,
#           якби він НЕ приймав ліки?"

# Потребує:
# 1. Abduction: визначити latent factors для цього пацієнта
# 2. Action: уявити альтернативний сценарій
# 3. Prediction: обчислити результат

# Найскладніший рівень — потребує causal model

Structural Causal Models (SCM)

Формальне визначення:

SCM M = (U, V, F)

U = {U₁, U₂, ...} — exogenous variables (зовнішні, незалежні)
V = {V₁, V₂, ...} — endogenous variables (визначаються моделлю)
F = {f₁, f₂, ...} — structural equations

Кожна fᵢ визначає Vᵢ як функцію від Pa(Vᵢ) та Uᵢ:
  Vᵢ = fᵢ(Pa(Vᵢ), Uᵢ)

де Pa(Vᵢ) — батьки Vᵢ у каузальному графі

Приклад: Простий медичний SCM

import numpy as np
from dataclasses import dataclass

@dataclass
class MedicalSCM:
    """Structural Causal Model для медичного сценарію."""

    def __init__(self):
        # Exogenous variables (noise)
        self.U_health = lambda: np.random.normal(0, 1)
        self.U_treatment = lambda: np.random.normal(0, 1)
        self.U_outcome = lambda: np.random.normal(0, 1)

    def generate(self, n_samples=1000, do_treatment=None):
        """Генерує дані з SCM.

        Args:
            do_treatment: якщо не None, intervention do(Treatment=value)
        """
        samples = []

        for _ in range(n_samples):
            # Structural equations
            # Health (confunder)
            health = 50 + 10 * self.U_health()

            # Treatment (залежить від health, якщо не intervention)
            if do_treatment is not None:
                treatment = do_treatment  # INTERVENTION
            else:
                # Лікарі частіше призначають treatment хворим
                prob_treat = 1 / (1 + np.exp(-(60 - health) / 10))
                treatment = 1 if np.random.random() < prob_treat else 0

            # Outcome (залежить від health та treatment)
            outcome = (
                0.5 * health +          # здоров'я важливе
                15 * treatment +         # treatment допомагає
                5 * self.U_outcome()     # noise
            )

            samples.append({
                'health': health,
                'treatment': treatment,
                'outcome': outcome
            })

        return samples

    def observational_effect(self, samples):
        """P(Outcome | Treatment) — association."""
        treated = [s['outcome'] for s in samples if s['treatment'] == 1]
        untreated = [s['outcome'] for s in samples if s['treatment'] == 0]
        return np.mean(treated) - np.mean(untreated)

    def interventional_effect(self, n_samples=1000):
        """P(Outcome | do(Treatment=1)) - P(Outcome | do(Treatment=0))."""
        treated = self.generate(n_samples, do_treatment=1)
        untreated = self.generate(n_samples, do_treatment=0)
        return np.mean([s['outcome'] for s in treated]) - \
               np.mean([s['outcome'] for s in untreated])

# Демонстрація різниці
scm = MedicalSCM()
obs_data = scm.generate(10000)

print(f"Observational effect: {scm.observational_effect(obs_data):.2f}")
# Може бути НЕГАТИВНИМ! Бо хворі частіше отримують treatment

print(f"Interventional (true) effect: {scm.interventional_effect():.2f}")
# Завжди позитивний — treatment справді допомагає

Do-Calculus: три правила Pearl

Правила для обчислення interventional queries з observational data:

# Notation:
# P(Y | do(X), Z) — distribution of Y given intervention do(X) and observation Z
# G_X — граф з видаленими стрілками, що входять у X
# G_\bar{X} — граф з видаленими стрілками, що виходять з X

# Rule 1: Insertion/deletion of observations
# P(Y | do(X), Z, W) = P(Y | do(X), Z)
# якщо (Y ⊥ W | X, Z) у G_\bar{X}

# Rule 2: Action/observation exchange
# P(Y | do(X), do(Z), W) = P(Y | do(X), Z, W)
# якщо (Y ⊥ Z | X, W) у G_\bar{X}Z

# Rule 3: Insertion/deletion of actions
# P(Y | do(X), do(Z), W) = P(Y | do(X), W)
# якщо (Y ⊥ Z | X, W) у G_\bar{X}\bar{Z(W)}

Backdoor Criterion:

def is_valid_adjustment_set(graph, X, Y, Z):
    """Перевіряє, чи Z блокує всі backdoor paths від X до Y.

    Backdoor path: path від X до Y, що починається зі стрілки -> X

    Умови:
    1. Z не містить descendants of X
    2. Z блокує всі backdoor paths
    """
    # Знаходимо всі backdoor paths
    backdoor_paths = find_paths_into_x(graph, X, Y)

    for path in backdoor_paths:
        if not is_blocked_by(path, Z):
            return False

    if any(is_descendant(z, X, graph) for z in Z):
        return False

    return True

def causal_effect_backdoor(data, X, Y, Z):
    """Обчислює P(Y | do(X)) через backdoor adjustment.

    P(Y | do(X)) = Σ_z P(Y | X, Z=z) P(Z=z)
    """
    effect = 0
    for z in data[Z].unique():
        # P(Y | X, Z=z)
        conditional = data[data[Z] == z].groupby(X)[Y].mean()
        # P(Z=z)
        p_z = (data[Z] == z).mean()
        effect += conditional * p_z
    return effect

Neural Causal Models

Causal VAE — Kocaoglu et al., 2018

import torch
import torch.nn as nn

class CausalVAE(nn.Module):
    """VAE з каузальною структурою в latent space."""

    def __init__(self, input_dim, latent_dim, causal_graph):
        super().__init__()
        self.latent_dim = latent_dim
        self.causal_graph = causal_graph  # Adjacency matrix

        # Encoder
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 256),
            nn.ReLU(),
        )
        self.fc_mu = nn.Linear(256, latent_dim)
        self.fc_var = nn.Linear(256, latent_dim)

        # Causal mechanisms for each latent variable
        self.causal_mechanisms = nn.ModuleList()
        for i in range(latent_dim):
            parents = self._get_parents(i)
            if len(parents) == 0:
                # Exogenous — just noise
                mechanism = nn.Identity()
            else:
                # Endogenous — function of parents
                mechanism = nn.Sequential(
                    nn.Linear(len(parents), 64),
                    nn.ReLU(),
                    nn.Linear(64, 1)
                )
            self.causal_mechanisms.append(mechanism)

        # Decoder
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 256),
            nn.ReLU(),
            nn.Linear(256, input_dim),
            nn.Sigmoid()
        )

    def _get_parents(self, node_idx):
        """Повертає індекси батьків вузла."""
        return torch.where(self.causal_graph[:, node_idx] == 1)[0].tolist()

    def encode(self, x):
        h = self.encoder(x)
        return self.fc_mu(h), self.fc_var(h)

    def reparameterize(self, mu, logvar):
        std = torch.exp(0.5 * logvar)
        eps = torch.randn_like(std)
        return mu + eps * std

    def apply_causal_structure(self, z_exogenous):
        """Застосовує каузальну структуру до exogenous noise."""
        z = torch.zeros_like(z_exogenous)

        # Topological order traversal
        for i in self._topological_order():
            parents = self._get_parents(i)
            if len(parents) == 0:
                z[:, i] = z_exogenous[:, i]
            else:
                parent_values = z[:, parents]
                z[:, i] = self.causal_mechanisms[i](parent_values).squeeze(-1)
                z[:, i] += z_exogenous[:, i]  # Add noise

        return z

    def decode(self, z):
        return self.decoder(z)

    def forward(self, x):
        mu, logvar = self.encode(x)
        z_exogenous = self.reparameterize(mu, logvar)
        z = self.apply_causal_structure(z_exogenous)
        return self.decode(z), mu, logvar, z

    def intervene(self, x, intervention_idx, intervention_value):
        """Виконує intervention на latent variable."""
        mu, logvar = self.encode(x)
        z_exogenous = self.reparameterize(mu, logvar)

        # Apply intervention: fix z[intervention_idx]
        z = torch.zeros_like(z_exogenous)
        for i in self._topological_order():
            if i == intervention_idx:
                z[:, i] = intervention_value  # INTERVENTION
            else:
                parents = self._get_parents(i)
                if len(parents) == 0:
                    z[:, i] = z_exogenous[:, i]
                else:
                    parent_values = z[:, parents]
                    z[:, i] = self.causal_mechanisms[i](parent_values).squeeze(-1)
                    z[:, i] += z_exogenous[:, i]

        return self.decode(z)

    def _topological_order(self):
        """Повертає вузли в топологічному порядку."""
        # Простий алгоритм для DAG
        visited = set()
        order = []

        def dfs(node):
            if node in visited:
                return
            visited.add(node)
            for parent in self._get_parents(node):
                dfs(parent)
            order.append(node)

        for i in range(self.latent_dim):
            dfs(i)

        return order

Counterfactual Reasoning у World Models

class CounterfactualWorldModel:
    """World Model з counterfactual reasoning."""

    def __init__(self, dynamics_model, causal_graph):
        self.dynamics = dynamics_model
        self.causal_graph = causal_graph

    def predict(self, state, action):
        """Звичайне передбачення наступного стану."""
        return self.dynamics(state, action)

    def counterfactual(self, observed_trajectory, intervention_time,
                       counterfactual_action):
        """
        Відповідає на питання:
        'Що було б, якби в момент t агент зробив іншу дію?'

        Args:
            observed_trajectory: реальна траєкторія {states, actions}
            intervention_time: момент часу для counterfactual
            counterfactual_action: альтернативна дія

        Returns:
            counterfactual_trajectory: траєкторія 'що було б, якби'
        """
        # Step 1: ABDUCTION
        # Визначаємо latent factors (U) з спостережень
        latent_factors = self._abduct(observed_trajectory)

        # Step 2: ACTION
        # Замінюємо дію в момент t
        cf_actions = observed_trajectory['actions'].copy()
        cf_actions[intervention_time] = counterfactual_action

        # Step 3: PREDICTION
        # Прогоняємо модель з новою дією, але тими самими U
        cf_states = [observed_trajectory['states'][0]]

        for t in range(len(cf_actions)):
            state = cf_states[-1]
            action = cf_actions[t]
            noise = latent_factors[t]  # Ті самі зовнішні фактори!

            next_state = self.dynamics.predict_with_noise(state, action, noise)
            cf_states.append(next_state)

        return {'states': cf_states, 'actions': cf_actions}

    def _abduct(self, trajectory):
        """Визначає latent factors з траєкторії."""
        latent_factors = []

        for t in range(len(trajectory['actions'])):
            s_t = trajectory['states'][t]
            a_t = trajectory['actions'][t]
            s_next = trajectory['states'][t + 1]

            # Знаходимо U таке, що dynamics(s, a, U) = s_next
            # Це inverse problem — може бути складно
            U = self.dynamics.infer_noise(s_t, a_t, s_next)
            latent_factors.append(U)

        return latent_factors

    def explain_decision(self, state, action, outcome):
        """Пояснює: 'Чому ця дія призвела до цього outcome?'"""
        # Тестуємо counterfactuals для різних альтернативних дій
        explanations = []

        for alt_action in self._get_alternative_actions(action):
            cf_outcome = self.counterfactual_single_step(state, alt_action)

            if cf_outcome != outcome:
                explanations.append({
                    'counterfactual_action': alt_action,
                    'counterfactual_outcome': cf_outcome,
                    'explanation': f"Якби агент зробив {alt_action}, "
                                   f"результат був би {cf_outcome}"
                })

        return explanations

Causal Discovery: навчитися каузальній структурі

from sklearn.linear_model import LassoCV
import networkx as nx

class CausalDiscovery:
    """Методи для визначення каузального графа з даних."""

    def pc_algorithm(self, data, alpha=0.05):
        """PC Algorithm для causal discovery.

        Починаємо з повного графа, видаляємо ребра на основі
        conditional independence tests.
        """
        n_vars = data.shape[1]
        graph = nx.complete_graph(n_vars).to_directed()

        # Phase 1: видаляємо ребра на основі unconditional independence
        for i, j in list(graph.edges()):
            if self._independent(data, i, j, []):
                graph.remove_edge(i, j)
                if graph.has_edge(j, i):
                    graph.remove_edge(j, i)

        # Phase 2: умовна незалежність зі зростаючим conditioning set
        for size in range(1, n_vars):
            for i, j in list(graph.edges()):
                # Шукаємо conditioning set
                neighbors = set(graph.predecessors(i)) | set(graph.successors(i))
                neighbors.discard(j)

                for cond_set in self._subsets(neighbors, size):
                    if self._independent(data, i, j, list(cond_set), alpha):
                        if graph.has_edge(i, j):
                            graph.remove_edge(i, j)
                        if graph.has_edge(j, i):
                            graph.remove_edge(j, i)
                        break

        # Phase 3: Orient edges (v-structures)
        graph = self._orient_edges(graph, data)

        return graph

    def _independent(self, data, i, j, cond_set, alpha=0.05):
        """Тест умовної незалежності."""
        from scipy import stats

        if len(cond_set) == 0:
            # Unconditional: Pearson correlation
            corr, p_value = stats.pearsonr(data[:, i], data[:, j])
        else:
            # Conditional: partial correlation
            corr = self._partial_correlation(data, i, j, cond_set)
            # Fisher z-transformation for p-value
            n = len(data)
            z = 0.5 * np.log((1 + corr) / (1 - corr + 1e-10))
            se = 1 / np.sqrt(n - len(cond_set) - 3)
            p_value = 2 * (1 - stats.norm.cdf(abs(z) / se))

        return p_value > alpha

    def notears(self, data, lambda_reg=0.1):
        """NOTEARS: Neural network-based causal discovery.

        Оптимізує:
        min ||X - XW||² + λ||W||₁
        s.t. trace(e^W) - d = 0  (acyclicity constraint)
        """
        n_vars = data.shape[1]
        W = torch.zeros(n_vars, n_vars, requires_grad=True)

        optimizer = torch.optim.Adam([W], lr=0.01)
        rho = 1.0  # Augmented Lagrangian parameter
        alpha = 0.0  # Lagrange multiplier

        for iteration in range(1000):
            optimizer.zero_grad()

            # Reconstruction loss
            X_pred = data @ W
            loss_recon = torch.mean((data - X_pred) ** 2)

            # Sparsity
            loss_sparse = lambda_reg * torch.sum(torch.abs(W))

            # Acyclicity constraint: h(W) = trace(e^W) - d
            M = torch.matrix_exp(W * W)  # element-wise square for positive
            h = torch.trace(M) - n_vars

            # Augmented Lagrangian
            loss = loss_recon + loss_sparse + alpha * h + 0.5 * rho * h * h

            loss.backward()
            optimizer.step()

            # Update Lagrange multiplier
            with torch.no_grad():
                alpha += rho * h.item()
                if h.item() > 0.25 * h.item():  # Slow progress
                    rho *= 2

            if h.abs().item() < 1e-8 and iteration > 100:
                break

        # Threshold small values
        W_final = W.detach().numpy()
        W_final[np.abs(W_final) < 0.3] = 0

        return W_final

Benchmarks для каузального reasoning

Causal Discovery:

• Sachs: protein signaling network (biology)
• DREAM: gene regulatory networks challenge
• Tuebingen Cause-Effect Pairs: 100+ bivariate datasets

Causal Inference:

• IHDP: infant health study
• Jobs: job training program
• Twins: twin birth outcomes

Counterfactual Reasoning:

• CoPhy: physical counterfactuals (video)
• CLEVRER: compositional language and video reasoning
• CausalWorld: robotic manipulation benchmark

Ідеї для дослідження

Для бакалавра:

• Реалізація простого SCM та порівняння observational vs interventional ефектів
• Тестування PC algorithm на синтетичних даних
• Огляд методів causal discovery

Для магістра:

• Інтеграція SCM у Dreamer-style world model
• Causal attention для video prediction
• Порівняння методів causal discovery для робототехніки
• Counterfactual data augmentation для imitation learning

Для PhD:

• Теоретичні межі causal learning з observational data
• Counterfactual world models для safe RL
• Causal transfer learning між доменами
• Unified framework для causal discovery + causal inference

Каузальне reasoning — це frontier AI research, яке об'єднує машинне навчання з філософією науки. Якщо вас цікавить ця тема для дисертації чи дипломної роботи, спеціалісти skp-degree.com.ua допоможуть з формулюванням дослідницьких питань та технічною реалізацією. Звертайтесь у Telegram: @kursovi_diplomy.

Де брати матеріал

Книги:

• "The Book of Why" (Judea Pearl) — доступний вступ для всіх
• "Causality: Models, Reasoning, and Inference" (Pearl) — формальна теорія
• "Elements of Causal Inference" (Peters et al.) — ML-focused
• "Causal Inference in Statistics: A Primer" (Pearl et al.)

Ключові статті:

• "Causal Confusion in Imitation Learning" (de Haan et al., 2019)
• "CausalWorld: A Robotic Manipulation Benchmark" (Ahmed et al., 2020)
• "Towards Causal Representation Learning" (Schölkopf et al., 2021)
• "A Meta-Transfer Objective for Learning to Disentangle Causal Mechanisms" (Bengio et al., 2019)

Бібліотеки:

• DoWhy (Microsoft): causal inference
• CausalNex (QuantumBlack/McKinsey): causal discovery
• pgmpy: probabilistic graphical models
• causal-learn: Python package for causal discovery

Курси:

• "Introduction to Causal Inference" (Brady Neal) — YouTube
• "Causal Inference" (Stanford CS 295)

Складність: ? PhD-рівень (для повного дослідження), ? Магістр (для applied)

Чому складно:

• Каузальна теорія — окрема математична область зі своєю термінологією
• Багато фундаментально нерозв'язаних проблем
• Потребує міждисциплінарних знань (ML + statistics + philosophy)
• Ідентифікація каузальних ефектів часто неможлива без експериментів

Для магістра: можна взяти конкретну підзадачу:

• Імплементація одного методу (NOTEARS або PC algorithm)
• Застосування до конкретного домену (робототехніка, медицина)
• Causal attention в трансформерах

Чому каузальність — це майбутнє AI

Каузальність — це те, що відрізняє справжнє розуміння від pattern matching. Система, яка розуміє причинність, може:

1. Узагальнювати на нові ситуації — бо розуміє механізми, а не поверхневі ознаки
2. Відповідати на "що якщо" — планувати та приймати рішення
3. Пояснювати свої рішення — через каузальні ланцюжки
4. Діяти безпечно — передбачаючи наслідки дій
5. Вчитися ефективніше — використовуючи структуру світу

Без каузальності немає AGI. Без каузальності робот не зможе адаптуватися до нового середовища. Без каузальності автопілот не зрозуміє, чому гальмувати перед тінню від дерева — погана ідея.

Це фундаментальна проблема. І вона відкрита для дослідників.

Ключові слова: causality, causal inference, causal discovery, counterfactual reasoning, Judea Pearl, SCM, structural causal models, do-calculus, world models, робототехніка, machine learning, PhD, дипломна робота, магістерська, дослідження AI.