RAPTOR - 递归抽象树型分层检索

Skill-RAPTOR-Hierarchical-RAG · 08-知识图谱

causalexperimentrecommendationragknowledge_graph广告与投放客服与VOC推荐与搜索知识图谱与RAG风控与合规WF-B 广告优化WF-C 客服分诊

业务视角

适用角色选品负责人 / 运营负责人 · 数据分析师 · 供应链负责人

适用平台Amazon 品类体系 · 竞品 ASIN 网络分析

什么情况下用品类很多，不清楚品类间的关联，没法做系统性类目扩张规划；竞品矩阵太复杂，品牌/SKU/渠道理不清

成功是什么样的建立品类知识图谱，清晰看到哪些是入口品/引流品/利润品，指导下一步选品扩张方向

业务痛点

品类太多不知道先做哪个竞品关系理不清楚不知道用户买了奶瓶还会买什么类目扩张没有逻辑

1. 解决的问题

RAPTOR（Recursive Abstractive Processing for Tree-Organized Retrieval）将长文档转化为一棵"抽象树"

2. 核心算法逻辑

RAPTOR（Recursive Abstractive Processing for TreeOrganized Retrieval）将长文档转化为一棵"抽象树"：

3. 业务应用场景

业务背景：亚马逊卖家政策（ToS）文档长达 200+ 页，覆盖商品安全法规、禁售品类、广告规则、FBA 操作手册。母婴类目尤其复杂（儿童安全法规 CPSC、FDA 婴儿食品标准）。合规团队每天需要回答"这款奶瓶清洗液能在亚马逊卖吗？"类问题，人工核查耗时 2-4 小时/问题。

RAPTOR 方案： 1. 将亚马逊 ToS + CPSC 法规 + FDA 标准文档构建 RAPTOR 树 2. 摘要层覆盖"禁售化学品"、"婴儿食品标准"等主题（全局） 3. 叶节点保留具体条款号和数值标准（细节） 4. 用户提问时，Collapsed Tree 同时命中合规总结 + 具体条款

量化 ROI： | 指标 | Before | After | 提升 | |---|---|---|---| | 合规问题响应时间 | 2-4 小时 | 3-5 分钟 | 95% 降低 | | 条款引用准确率 | 70%（人工记忆） | 91% | +30% | | 合规团队人力 | 4人 | 1.5人 | 节省 $180K/年 | | ToS 违规处罚风险 | 基准 | 降低 60% | |

4. 输入数据要求

推荐块大小：150-300 tokens（语义完整段落）
避免在句子中间截断（配合 `Skill-Semantic-Chunking-Strategy`）
标题/章节信息保留在块内（如 `[3.2 清洗规范] 奶瓶每次使用后...`）

5. 输出结果

推荐块大小：150-300 tokens（语义完整段落）
避免在句子中间截断（配合 `Skill-Semantic-Chunking-Strategy`）
标题/章节信息保留在块内（如 `[3.2 清洗规范] 奶瓶每次使用后...`）

6. 业务价值 / ROI

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

7. 代码模板

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

"""
RAPTOR - 递归抽象树型分层检索系统
arXiv: 2401.18059 (RAPTOR, Stanford, ICLR 2024)

实现要点：
1. 文档分块 -> embedding
2. UMAP 降维 + GMM 软聚类
3. LLM 生成摘要（mock）
4. 递归构建抽象树
5. Collapsed Tree 向量检索

运行环境：Python 3.9+，无需外部 API（全 mock）
"""

import ast
import math
import random
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field


# ─────────────────────────────────────────────
# 数据结构
# ─────────────────────────────────────────────

@dataclass
class TreeNode:
    """RAPTOR 树节点"""
    node_id: str
    text: str
    embedding: List[float]
    level: int                          # 0=叶节点，1,2,...=摘要层
    children: List[str] = field(default_factory=list)   # 子节点 ID
    parent: Optional[str] = None
    cluster_id: Optional[int] = None


@dataclass
class RAPTORTree:
    """RAPTOR 树结构"""
    nodes: Dict[str, TreeNode] = field(default_factory=dict)
    root_id: Optional[str] = None
    max_level: int = 0

    def get_all_nodes(self) -> List[TreeNode]:
        return list(self.nodes.values())

    def get_nodes_by_level(self, level: int) -> List[TreeNode]:
        return [n for n in self.nodes.values() if n.level == level]


# ─────────────────────────────────────────────
# Mock 工具函数（生产环境替换为真实实现）
# ─────────────────────────────────────────────

def mock_embed(text: str, dim: int = 16) -> List[float]:
    """Mock embedding：用文本 hash 生成确定性向量"""
    random.seed(hash(text) % (2 ** 31))
    vec = [random.gauss(0, 1) for _ in range(dim)]
    norm = math.sqrt(sum(v * v for v in vec)) + 1e-9

8. 论文来源

2401.18059