html Каузальні залежності у World Models: від кореляції до причинності| SKP-Degree
Технології AI Written by practicing developers

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

Updated: 12min of reading 50views

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

Don't want to figure it out yourself?We will order turnkey work— payment after demonstration. Order on the website

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

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

Для 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.

About the author

Команда SKP-Degree

Verified author

Розробники та дослідники AI · Python, TensorFlow, PyTorch · Досвід у промисловій розробці

SKP-Degree team are professional developers with experience7+ yearsin industrial development. Executed1000+ projectsfor students from Ukraine, Poland and the Baltic countries.

Python Django Java ML/AI React C# / .NET JavaScript

Order programming work

Let's write a turnkey - payment after the demonstration of the finished code:

Need help with work?

Order a term paper or thesis on programming. Payment after demonstration!

No prepayment Video demonstration Autonomous work 24/7

SKP-Degree— course, diploma and laboratory courses on turnkey programming