CausalFlow — LLM Agent 因果调试：失败轨迹 → 最小反事实修复

Skill-CausalFlow-Agent-Failure-Repair · 16-智能体工程

causalexperimentforecastingoptimizationmulti_agent广告与投放供应链与补货MAS与智能体工程风控与合规WF-A 智能补货WF-B 广告优化WF-D 选品扫描

实现难度⭐⭐⭐☆☆

业务优先级⭐⭐⭐⭐⭐

业务视角

适用角色CTO / 技术负责人 · 产品经理 · 数据工程师

适用平台跨境运营 AI Agent 工程落地 · Amazon SP API + LLM 集成 · 多平台数据采集 Agent

什么情况下用想把 AI 集成到业务系统，但 LLM 稳定性差、幻觉问题、成本控制都是挑战；Agent 任务失败了不知道哪步出了问题

成功是什么样的AI Agent 在生产环境稳定运行，失败可追踪，成本可控，复杂任务完成率 >85%

业务痛点

LLM 返回结果不稳定不可靠AI 幻觉导致业务决策错误Agent 任务失败了不知道哪步出问题AI 调用成本控制不住

1. 解决的问题

CausalFlow 将 Agent 的一次执行视为有序步骤序列

2. 核心算法逻辑

执行轨迹建模为步骤依赖链

3. 业务应用场景

业务背景：补货 Agent 执行 3 步流程：① 预测需求 → ② 计算安全库存 → ③ 生成 PO。某次执行中，最终 PO 数量异常（比正常大 3 倍），触发审核警告。

修复生成： - 定位 step_2 为根因（安全库存计算公式错误：将安全系数 3 误用了 3 倍） - 生成修复版：`safety_stock = forecast * safety_factor`（safety_factor=0.3，而非 3） - 验证：修复后重跑，po_qty = 500 + 150 = 650（合理）✅

业务背景：合规检查 Agent 完成 4 步检查流程后报告"合规通过"，但人工复核发现漏掉了 CPSC 强制认证要求。

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

未自动抽取；请查看原始 Skill 卡片。

7. 代码模板

代码块数量：6 · 路径：未检测到

"""
CausalFlow — LLM Agent 因果调试与反事实修复
论文：CausalFlow: Causal Attribution and Counterfactual Repair for LLM Agent Failures
arXiv：2605.25338 | 2026年5月
"""

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
from collections import defaultdict


# ──────────────────────────────────────────────
# 数据类
# ──────────────────────────────────────────────

class StepOutcome(Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    UNKNOWN = "unknown"


@dataclass
class AgentStep:
    """Agent 执行轨迹中的单个步骤"""
    step_id: str
    action: str              # 该步骤执行的动作/推理
    observation: str         # 执行结果/观察
    dependencies: list[str]  # 依赖的前置步骤 ID 列表
    outcome: StepOutcome = StepOutcome.UNKNOWN
    corrected_action: Optional[str] = None  # 修复后的动作（反事实）

    def __repr__(self) -> str:
        return f"Step({self.step_id}: {self.action[:40]}... [{self.outcome.value}])"


@dataclass
class ExecutionTrace:
    """完整的 Agent 执行轨迹"""
    trace_id: str
    steps: list[AgentStep]
    final_outcome: StepOutcome
    task_description: str = ""

    def get_step(self, step_id: str) -> Optional[AgentStep]:
        return next((s for s in self.steps if s.step_id == step_id), None)

    def is_failed(self) -> bool:
        return self.final_outcome == StepOutcome.FAILURE

    def step_ids(self) -> list[str]:
        return [s.step_id for s in self.steps]


@dataclass
class RepairedTrace:
    """修复后的执行轨迹"""
    original_trace_id: str
    repaired_steps: list[AgentStep]

8. 论文来源

2605.25338