观察性数据里的“相关性”并不等于“因果性”。真正的因果分析通常需要明确分析计划、选择合适的统计方法、检查假设、解释结果，并把不确定性和后续行动写进报告。Netflix 开源的 Agentic Workflow for Observational Causal Inference（OCI）试图把这条链路组织成一个智能工作流：用户提供观察性数据和分析计划，Agent 通过 actor-critic 循环估计因果关系、生成报告，并提出下一步建议。
它的价值不只是“让模型替人写分析结论”，而是把因果分析中的重复性工作拆成可以反复评估和修正的步骤。
从一次性回答变成迭代式分析
传统的数据分析脚本往往是线性的：读取数据、运行模型、输出结果。因果推断却很少能靠一次运行完成，因为每个结果都依赖一组前提：处理变量是否定义清楚，结果变量是否合适，混杂因素是否被考虑，模型假设是否足够可信。
OCI 使用 actor-critic loop，可以把它理解为两个互补角色：
- Actor 负责执行分析动作，例如选择估计方法、运行检验、生成报告草稿或提出诊断问题。
- Critic 负责检查当前分析是否符合计划、结果是否存在矛盾、假设是否被满足，以及是否需要重做某一步。
这种循环让工作流具备“分析、审查、修正”的节奏。Agent 的输出不应被看成脱离上下文的最终真理，而应被看成一份带有证据、假设和后续问题的分析结果。
人类分析计划仍然是边界条件
摘要中一个重要设计是：用户需要提供 observational data 和 human user's analysis plan。也就是说，Agent 并不是在没有目标的情况下自由探索数据，而是在人的研究问题和分析计划约束下工作。
这一区分很关键。因果推断首先是研究设计问题，其次才是模型选择问题。一个自动化系统可以帮助完成数据检查、估计和文档整理，却不能凭空决定“什么干预值得研究”或“哪些未观测变量足以解释结果”。
实践中，分析计划至少应明确以下内容：
- 处理变量：哪些样本接受了什么处理或干预。
- 结果变量：要衡量的业务、用户或实验结果是什么。
- 目标人群：结论适用于哪些样本。
- 因果 estimand：例如平均处理效应，还是特定人群的处理效应。
- 已知混杂因素：哪些变量可能同时影响处理和结果。
- 结果解释边界：哪些结论可以用于决策，哪些只能作为探索性发现。
计划写得越清楚，Agent 越容易在正确的问题空间内工作；计划写得越模糊，自动化越可能把统计上的便利误当成研究上的合理。
一个可改造的最小工作流示例
下面的代码是一个独立的最小示例，用于演示如何把 actor-critic 思路接入观察性因果分析。它不是 Netflix 工作流的实现，而是一个可以改造成内部分析流水线的示意版本。示例使用 pandas 和 statsmodels，假设数据文件包含结果列 outcome、处理列 treatment，以及若干混杂因素。
运行前安装依赖，并将 data.csv 换成自己的数据文件：
python -m pip install pandas statsmodels
python causal_workflow.py
将下面内容保存为 causal_workflow.py：
from pathlib import Path
import pandas as pd
import statsmodels.formula.api as smf
ANALYSIS_PLAN = {
"treatment": "treatment",
"outcome": "outcome",
"confounders": ["age", "region_score"],
"estimand": "adjusted average treatment effect",
}
def actor_estimate(data: pd.DataFrame, plan: dict) -> dict:
"""执行一个可解释的回归调整估计步骤。"""
columns = [plan["outcome"], plan["treatment"], *plan["confounders"]]
frame = data[columns].dropna()
rhs = " + ".join([plan["treatment"], *plan["confounders"]])
formula = f'{plan["outcome"]} ~ {rhs}'
model = smf.ols(formula, data=frame).fit(cov_type="HC3")
return {
"formula": formula,
"estimate": float(model.params[plan["treatment"]]),
"p_value": float(model.pvalues[plan["treatment"]]),
"n": int(model.nobs),
"warnings": [],
}
def critic_review(result: dict, plan: dict) -> dict:
"""检查结果是否有明显问题，并决定是否需要补充分析。"""
warnings = list(result["warnings"])
if result["n"] < 100:
warnings.append("有效样本量较小，结果可能不稳定")
if result["p_value"] >= 0.05:
warnings.append("处理变量系数未达到预设显著性阈值")
return {
"approved": not warnings,
"warnings": warnings,
"next_steps": [
"检查处理组和对照组的协变量平衡",
"进行替代模型或敏感性分析",
"确认未观测混杂因素对结论的潜在影响",
],
}
def main() -> None:
data = pd.read_csv(Path("data.csv"))
result = actor_estimate(data, ANALYSIS_PLAN)
review = critic_review(result, ANALYSIS_PLAN)
print("估计公式:", result["formula"])
print("处理效应估计:", round(result["estimate"], 4))
print("p 值:", round(result["p_value"], 4))
print("有效样本量:", result["n"])
print("审查通过:", review["approved"])
print("警告:")
for warning in review["warnings"]:
print("-", warning)
print("建议的下一步:")
for step in review["next_steps"]:
print("-", step)
if __name__ == "__main__":
main()
这个示例只覆盖了“执行一个估计步骤”和“生成审查意见”。更完整的 Agentic Workflow 可以把 actor_estimate 扩展为工具调用层，把 critic_review 扩展为基于分析计划、诊断结果和报告证据的审查器。当审查失败时，工作流可以要求 Actor 补充协变量、运行替代估计方法，或把结果标记为需要人工确认。
自动化最适合处理哪些工作
这类工作流最适合减少分析过程中的机械劳动，例如：
- 自动检查输入数据中的缺失值、类型和变量定义。
- 根据分析计划生成候选模型和运行记录。
- 把估计结果、诊断信息和假设整理成结构化报告。
- 发现样本量不足、显著性不足或变量不平衡等信号。
- 根据当前结果生成下一步敏感性分析建议。
报告生成尤其有价值。因果分析的可复现性不仅依赖一个效应值，还依赖数据版本、变量处理、模型公式、样本筛选、置信区间、诊断结果和解释限制。Agent 可以帮助把这些信息集中起来，降低“代码运行过了，但没人知道为什么这样运行”的风险。
不能交给 Agent 自动决定的事情
因果推断的核心限制不会因为引入 Agent 而消失。观察性数据仍然可能受到未观测混杂、选择偏差、测量误差和反向因果的影响。一个模型给出统计显著的处理效应，并不自动证明处理导致了结果变化。
因此，生产环境应至少保留以下控制点：
- 人工批准分析计划和因果 estimand。
- 对关键变量的定义、时间窗口和样本筛选进行审计。
- 要求报告同时展示效应、不确定性、假设和失败的诊断。
- 对 Agent 选择的模型和生成的下一步建议保留完整日志。
- 将探索性结果与可以直接支持业务决策的结论分开。
落地时的一份检查清单
采用这类工作流时，可以从一个窄问题开始，而不是立即自动化所有因果分析：
- 选择一个已有人工分析流程、数据边界稳定的 OCI 场景。
- 把分析计划写成结构化输入，而不是只放在自然语言备注里。
- 为每个 Agent 工具定义输入、输出和失败条件。
- 让 Critic 检查证据链，而不只是检查文本是否通顺。
- 保存每次迭代的模型、参数、数据版本和人工决策。
- 用历史分析结果评估 Agent 是否减少了耗时，以及是否引入了新的误导性结论。
Netflix 开源这一工作流的启发在于：因果分析自动化的重点不是让模型替人“猜因果”，而是把研究问题、统计估计、质量审查和报告沟通连接起来。只要人类仍然负责定义问题和判断证据边界，Agent 才有机会成为可靠的分析协作者，而不是一个会生成漂亮结论的黑盒。