CAR — Agent步骤因果Shapley归因：多步交互效应定量拆解
Skill-CAR-Agent-Causal-Shapley · 16-智能体工程
causalexperimentforecastingmulti_agent广告与投放供应链与补货MAS与智能体工程WF-A 智能补货WF-B 广告优化WF-F 动态定价WF-I 智能体工程
实现难度⭐⭐⭐☆☆
业务视角
适用角色CTO / 技术负责人 · 产品经理 · 数据工程师
适用平台跨境运营 AI Agent 工程落地 · Amazon SP API + LLM 集成 · 多平台数据采集 Agent
什么情况下用想把 AI 集成到业务系统，但 LLM 稳定性差、幻觉问题、成本控制都是挑战；Agent 任务失败了不知道哪步出了问题
成功是什么样的AI Agent 在生产环境稳定运行，失败可追踪，成本可控，复杂任务完成率 >85%
业务痛点
1. 解决的问题
运营技术团队面临"补货/广告Agent决策失效但不知哪步出错"——结构因果模型+蒙特卡洛Shapley将多步交互责任定量拆解，MTTR从3天降至0.5天，年化断货损失预防50-150万元
2. 核心算法逻辑
当 LLM Agent 失败时（多退款、错误工具调用、数据泄露），现有工具只能回答"发生了什么"（可观测性）或"是否通过"（评估），但无法回答"哪一步决定导致了失败"。
3. 业务应用场景
业务问题：母婴跨境卖家的补货工作流由4个Agent组成（需求预测 → 库存评估 → 采购决策 → PO生成），大促前某批PO生成了错误采购量（比需求预测低40%），导致大促期间断货损失¥380,000。事后需要确定"哪个Agent的哪步决定是根本原因"，以便修复workflow逻辑。
现有工具的局限：AgentTrace（图遍历）定位到"采购决策Agent出错"，但无法区分是"库存评估Agent给的输入有偏差"还是"采购决策Agent的安全库存计算逻辑本身有问题"——两者都参与了最终错误，但贡献比例未知。
CAR处理： - 将整次补货执行轨迹建模为SCM，4个步骤互为依赖节点 - 分别对"库存评估步骤"和"采购决策步骤"执行反事实干预（do-operator） - 重新跑轨迹，测量各步干预后PO准确率变化 - 蒙特卡洛Shapley分配：库存评估步骤贡献 φ=0.62，采购决策步骤贡献 φ=0.31
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI预估：
- 断货损失预防：¥50,000-380,000/次（母婴大促场景）
- 工程排查时间：从3天→0.5天/次，节省工程师成本 ¥15,000+/次
- ROAS损失预防（广告Agent）：¥80,000-150,000/季度
- 年化综合ROI：¥500,000-1,500,000（视Agent规模）
- 实施难度：⭐⭐⭐☆☆
7. 代码模板
代码块数量：1 · 路径：未检测到
"""
CAR - Causal Agent Replay: Agent步骤Shapley因果归因
基于结构因果模型和蒙特卡洛Shapley估计
依赖: numpy, itertools
"""
import numpy as np
from itertools import combinations
from typing import Callable, List, Dict, Tuple
import random
# ─────────────────────────────────────────────
# 数据结构定义
# ─────────────────────────────────────────────
class AgentStep:
"""Agent执行轨迹中的单个步骤"""
def __init__(self, step_id: str, action: dict, observation: dict, outcome: float = None):
self.step_id = step_id
self.action = action # 该步执行的动作
self.observation = observation # 该步的输入上下文
self.outcome = outcome # 0=失败, 1=成功, 或连续值
class AgentTrajectory:
"""完整的Agent执行轨迹"""
def __init__(self, steps: List[AgentStep], final_outcome: float, trajectory_id: str = ""):
self.steps = steps
self.final_outcome = final_outcome # 最终结果(0=失败/1=成功)
self.trajectory_id = trajectory_id
# ─────────────────────────────────────────────
# CAR核心实现
# ─────────────────────────────────────────────
class CausalAgentReplay:
"""
CAR: 结构因果模型 + 蒙特卡洛Shapley
用法：
1. 提供历史失败轨迹
2. 提供每步的"反事实正确动作"
3. 提供轨迹重执行函数
4. 调用 compute_shapley() 获取每步的因果贡献
"""
def __init__(self, replay_fn: Callable, n_monte_carlo: int = 100, seed: int = 42):
"""
replay_fn: 给定步骤干预集合，重执行轨迹并返回结果
签名: (trajectory, interventions: dict{step_id: counterfactual_action}) -> float
n_monte_carlo: Shapley蒙特卡洛采样次数（100次约达到±0.05精度）
"""
self.replay_fn = replay_fn
self.n_monte_carlo = n_monte_carlo
np.random.seed(seed)
random.seed(seed)
def counterfactual_effect(
self,
trajectory: AgentTrajectory,8. 论文来源
- 2606.08275