AI 数据分析：大模型驱动的智能归因分析，从相关性到因果推断

一、"数据会说话"的困境：相关性不等于因果性

数据分析中最常见的陷阱，是把相关性当作因果性。电商平台的运营报告显示"用户浏览时长与购买转化率正相关"，于是产品团队投入大量资源优化页面停留时长。然而，真正的因果关系可能是"有明确购买意图的用户本身就会浏览更久"——延长浏览时间并不能提升转化率。

传统归因分析依赖人工假设和 A/B 测试验证，周期长、成本高，且只能验证预设的假设，无法发现未知的因果路径。大模型的出现为归因分析提供了新的可能：通过自然语言交互，分析师可以快速生成假设、自动匹配数据、识别混淆变量，将归因分析从"人工假设驱动"推进到"AI 辅助发现驱动"。

但大模型本身并不理解因果性——它只是模式匹配器。如何将因果推断的数学框架与大模型的语义理解能力结合，是当前 AI 数据分析的前沿课题。

二、因果推断的数学基础与 AI 辅助框架

2.1 从相关性到因果性：潜在结果框架

因果推断的核心问题是反事实推断——"如果用户没有收到优惠券，转化率会是多少？"潜在结果框架（Potential Outcome Framework）为这个问题提供了数学表达：

$$Y_i(1) - Y_i(0) = \text{个体因果效应}$$

由于同一时刻只能观测到一个处理状态的结果，个体因果效应不可观测。我们只能估计平均处理效应（ATE）：

$$\text{ATE} = E[Y(1) - Y(0)]$$

2.2 混淆变量与倾向得分

混淆变量同时影响处理分配和结果变量，导致观察到的相关性不等于因果性。倾向得分匹配（PSM）通过估计每个个体接受处理的概率来平衡混淆变量：

flowchart TD
    A[原始数据：含混淆变量] --> B[估计倾向得分 P X]
    B --> C[倾向得分匹配：平衡处理组和对照组]
    C --> D[检验平衡性：SMD < 0.1]
    D --> E[估计因果效应：ATT / ATE]
    E --> F[敏感性分析：检验隐藏混淆变量]

    subgraph AI辅助环节
        G[LLM 识别潜在混淆变量]
        H[LLM 生成因果图 DAG]
        I[LLM 解释分析结果]
    end

    A -.-> G
    G -.-> B
    E -.-> I
    H -.-> B

2.3 AI 辅助因果推断的架构

大模型在因果推断中的角色不是替代数学框架，而是辅助三个关键环节：混淆变量识别、因果图构建、结果解释。

三、智能归因分析的工程实现

3.1 因果推断引擎

import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler
from typing import List, Tuple, Optional
from dataclasses import dataclass


@dataclass
class CausalEffect:
    """因果效应估计结果"""
    ate: float              # 平均处理效应
    att: float              # 处理组平均处理效应
    ci_lower: float         # 95% 置信区间下界
    ci_upper: float         # 95% 置信区间上界
    p_value: float          # 显著性 p 值
    balanced: bool          # 匹配后混淆变量是否平衡


class CausalInferenceEngine:
    """因果推断引擎：倾向得分匹配 + 效应估计"""

    def __init__(self, data: pd.DataFrame):
        self.data = data
        self.scaler = StandardScaler()

    def estimate_propensity_score(
        self,
        treatment: str,
        confounders: List[str],
    ) -> pd.Series:
        """估计倾向得分：每个样本接受处理的概率"""
        X = self.data[confounders].values
        y = self.data[treatment].values

        X_scaled = self.scaler.fit_transform(X)
        ps_model = LogisticRegression(max_iter=1000, C=1.0)
        ps_model.fit(X_scaled, y)

        propensity = ps_model.predict_proba(X_scaled)[:, 1]
        return pd.Series(propensity, index=self.data.index, name='propensity_score')

    def match_by_propensity(
        self,
        treatment: str,
        outcome: str,
        confounders: List[str],
        caliper: float = 0.2,
    ) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """基于倾向得分匹配处理组和对照组"""
        ps = self.estimate_propensity_score(treatment, confounders)
        self.data = self.data.copy()
        self.data['ps'] = ps

        treated = self.data[self.data[treatment] == 1].copy()
        control = self.data[self.data[treatment] == 0].copy()

        # 计算倾向得分的标准差，用于确定匹配容差
        ps_std = ps.std()
        max_dist = caliper * ps_std

        matched_treated = []
        matched_control = []

        for _, t_row in treated.iterrows():
            distances = (control['ps'] - t_row['ps']).abs()
            min_idx = distances.idxmin()

            if distances[min_idx] <= max_dist:
                matched_treated.append(t_row)
                matched_control.append(control.loc[min_idx])
                # 移除已匹配的对照样本，避免重复匹配
                control = control.drop(min_idx)

        return pd.DataFrame(matched_treated), pd.DataFrame(matched_control)

    def estimate_effect(
        self,
        treatment: str,
        outcome: str,
        confounders: List[str],
        caliper: float = 0.2,
    ) -> CausalEffect:
        """估计因果效应：ATE 和 ATT"""
        treated, control = self.match_by_propensity(
            treatment, outcome, confounders, caliper
        )

        if len(treated) == 0 or len(control) == 0:
            raise ValueError("匹配后样本为空，请检查混淆变量和匹配容差")

        # ATT：处理组的平均处理效应
        att = treated[outcome].mean() - control[outcome].mean()

        # ATE：整体平均处理效应（简化估计）
        ate = att  # 严格来说需要更复杂的估计方法

        # Bootstrap 置信区间
        n_bootstrap = 1000
        bootstrap_ates = []
        combined = pd.concat([treated, control]).reset_index(drop=True)
        for _ in range(n_bootstrap):
            sample = combined.sample(n=len(combined), replace=True)
            t_sample = sample[sample[treatment] == 1]
            c_sample = sample[sample[treatment] == 0]
            if len(t_sample) > 0 and len(c_sample) > 0:
                bootstrap_ates.append(
                    t_sample[outcome].mean() - c_sample[outcome].mean()
                )

        ci_lower = np.percentile(bootstrap_ates, 2.5)
        ci_upper = np.percentile(bootstrap_ates, 97.5)

        # 检验匹配后混淆变量的平衡性
        max_smd = 0.0
        for conf in confounders:
            t_mean, c_mean = treated[conf].mean(), control[conf].mean()
            t_std, c_std = treated[conf].std(), control[conf].std()
            pooled_std = np.sqrt((t_std**2 + c_std**2) / 2)
            if pooled_std > 0:
                smd = abs(t_mean - c_mean) / pooled_std
                max_smd = max(max_smd, smd)

        balanced = max_smd < 0.1

        # 简化的 p 值估计
        p_value = np.mean([abs(b) >= abs(ate) for b in bootstrap_ates])

        return CausalEffect(
            ate=ate,
            att=att,
            ci_lower=ci_lower,
            ci_upper=ci_upper,
            p_value=p_value,
            balanced=balanced,
        )

3.2 大模型辅助的混淆变量识别

import json
from openai import OpenAI

def identify_confounders_with_llm(
    client: OpenAI,
    treatment: str,
    outcome: str,
    available_columns: List[str],
    domain_context: str = "",
) -> List[str]:
    """使用大模型识别潜在混淆变量"""
    prompt = f"""你是一名专业的数据分析师，正在执行因果推断分析。

处理变量：{treatment}
结果变量：{outcome}
可用数据列：{json.dumps(available_columns, ensure_ascii=False)}
业务背景：{domain_context}

请从可用数据列中识别可能同时影响处理变量和结果变量的混淆变量。
只返回 JSON 数组格式，不要其他内容。例如：["age", "income", "city_tier"]"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
    )

    content = response.choices[0].message.content.strip()
    confounders = json.loads(content)
    # 验证返回的变量名是否在可用列中
    valid_confounders = [c for c in confounders if c in available_columns]
    return valid_confounders

flowchart TD
    A[分析师输入：处理变量 + 结果变量] --> B[LLM 识别混淆变量]
    B --> C[LLM 生成因果图 DAG]
    C --> D[倾向得分匹配]
    D --> E[因果效应估计]
    E --> F[LLM 解释结果]
    F --> G{效应显著?}
    G -- 是 --> H[输出归因结论]
    G -- 否 --> I[LLM 建议新假设]
    I --> B

四、AI 辅助因果推断的边界与权衡

4.1 大模型不保证因果正确性

大模型基于训练数据的统计模式识别混淆变量，可能遗漏领域专家才知道的关键变量，也可能引入伪混淆变量。因果图 DAG 的方向性判断（A→B 还是 B→A）本质上需要领域知识，大模型的判断不可靠。建议将 LLM 识别的混淆变量作为候选列表，由分析师审核确认。

4.2 倾向得分匹配的局限

PSM 假设所有混淆变量都可观测（无隐藏混淆变量假设），这在现实中几乎不可能完全满足。敏感性分析（如 Rosenbaum Bounds）可以量化隐藏混淆变量对结论的影响程度，但无法消除不确定性。

4.3 样本量与匹配质量的矛盾

严格的匹配容差（小 caliper）保证匹配质量，但可能导致大量样本被丢弃，降低统计功效。宽松的容差保留更多样本，但匹配质量下降，引入偏差。在样本量有限时，这个矛盾尤为突出。

4.4 大模型的成本与延迟

每次因果分析调用 LLM 识别混淆变量和解释结果，增加了 API 成本和等待时间。对于高频归因分析场景（如实时监控看板），需要缓存 LLM 结果或使用本地小模型替代。

五、总结

AI 辅助的智能归因分析将大模型的语义理解能力与因果推断的数学框架结合，在混淆变量识别、因果图构建和结果解释三个环节提升分析效率。但大模型不能替代因果推断的数学严谨性——它只是辅助工具，最终结论仍需通过统计检验和领域验证。

工程落地的关键决策：混淆变量识别采用"LLM 候选 + 人工审核"模式，避免遗漏关键变量；倾向得分匹配的 caliper 设为 0.2 倍标准差作为起步值，根据匹配后 SMD 调整；Bootstrap 置信区间至少 1000 次重采样，保证估计稳定性；敏感性分析是必做步骤，不检验隐藏混淆变量就下因果结论是不负责任的。

因果推断的价值不在于给出确定答案，而在于量化不确定性。AI 辅助让这个过程更快、更全面，但"相关性不等于因果性"这条红线，任何工具都不能跨越。