CausalRAG — 因果图增强检索：语义相似 + 因果链路双轨 RAG

1. 解决的问题

传统 RAG 将文档切分为固定长度的 chunk，导致原本有完整因果逻辑的段落被割裂。例如"产品A未通过认证 → 被召回 → 触发贸易禁令"这一因果链横跨多个 chunk，单个 chunk 无法表达完整逻辑。

2. 核心算法逻辑

传统 RAG 的两个核心问题

3. 业务应用场景

业务问题： 运营团队需要查询"婴儿配方奶粉 FDA 21 CFR 107 是什么？违规会导致什么召回？召回会触发哪些业务影响？"这类需要完整因果链的问题。

传统 RAG 的缺陷： 语义检索会分别返回"21 CFR 107 法规文本"、"召回案例"、"业务损失报告"三段不相关的文本块，LLM 需要自行推理它们的因果连接，容易出错或遗漏。

CausalRAG 的优势： 构建因果图后，图中已存在显式路径： 检索时沿该路径抽取所有相关节点，LLM 获得完整因果上下文，生成连贯且准确的答案。

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

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

7. 代码模板

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

"""
CausalRAG — 因果图增强检索
论文：CausalRAG: Integrating Causal Graphs into Retrieval-Augmented Generation
arXiv：2503.19878 | ACL Findings 2025
"""

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
import re
import math
from collections import defaultdict, deque


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

@dataclass
class CausalTriple:
    """因果三元组"""
    cause: str
    relation: str  # leads_to / causes / triggers / results_in
    effect: str
    source_doc: str
    confidence: float = 1.0

    def __repr__(self) -> str:
        return f"[{self.cause}] --{self.relation}--> [{self.effect}] (doc={self.source_doc})"


@dataclass
class Document:
    """文档片段"""
    doc_id: str
    content: str
    metadata: dict = field(default_factory=dict)


# ──────────────────────────────────────────────
# 因果知识图谱
# ──────────────────────────────────────────────

class CausalKnowledgeGraph:
    """有向因果图：存储因果三元组，支持前向/后向链路遍历"""

    CAUSAL_PATTERNS = [
        (r"(.+?)\s+(?:leads? to|lead to)\s+(.+)", "leads_to"),
        (r"(.+?)\s+(?:causes?|cause)\s+(.+)", "causes"),
        (r"(.+?)\s+(?:triggers?|trigger)\s+(.+)", "triggers"),
        (r"(.+?)\s+(?:results? in|result in)\s+(.+)", "results_in"),
        (r"(?:due to|because of)\s+(.+?),\s+(.+)", "caused_by"),
        (r"(.+?)\s+(?:导致|引起|触发)\s+(.+)", "leads_to"),
        (r"(.+?)\s+(?:造成|引发)\s+(.+)", "causes"),
    ]

    def __init__(self):
        # 邻接表：cause -> [(relation, effect, triple)]
        self._forward: dict[str, list[tuple[str, str, CausalTriple]]] = defaultdict(list)
        # 反向邻接表：effect -> [(relation, cause, triple)]

8. 论文来源

• 2503.19878