GraphRAG - 知识图谱增强检索生成

1. 解决的问题

GraphRAG（Graph Retrieval-Augmented Generation） 将传统 RAG（检索增强生成）中的文本块检索升级为知识图谱结构化检索，通过图遍历获取与查询相关的实体、关系和子图，显著提升复杂推理场景的答案准确性和可解释性。

2. 核心算法逻辑

GraphRAG（Graph RetrievalAugmented Generation） 将传统 RAG（检索增强生成）中的文本块检索升级为知识图谱结构化检索，通过图遍历获取与查询相关的实体、关系和子图，显著提升复杂推理场景的答案准确性和可解释性。

3. 业务应用场景

业务问题： 母婴出海电商的客服团队每天处理大量用户咨询，如"吸奶器A和吸奶器B有什么区别？"、"我买了吸奶器，还需要买什么配件？"。传统FAQ系统只能匹配关键词，无法处理需要关联推理的问题。

数据要求： - 商品知识图谱： - 实体：商品（吸奶器、温奶器、储奶袋）、品牌、属性（价格、材质、功能） - 关系：配套购买、同类竞品、适用人群、使用场景 - 用户行为数据：购买记录、浏览记录、评价内容 - 客服历史记录：常见问题及标准答案

预期产出： - 精准商品对比：回答"吸奶器A和B的区别"时，自动提取两者在价格、功能、评价上的差异 - 智能配件推荐：回答"买了吸奶器还需要什么"时，基于图谱中的"配套购买"关系推荐 - 用户画像关联：结合用户购买历史，提供个性化回答

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

• 数据要求：需要构建完整的商品和用户知识图谱
• 技术门槛：较高，需要理解图算法和检索策略
• 工程复杂度：高，涉及知识图谱构建、检索优化、LLM集成
• 维护成本：中，知识图谱需要持续更新
• 业务价值高：客服和推荐是电商核心场景
• 技术前沿性：GraphRAG 是 RAG 2.0 的发展方向

7. 代码模板

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

"""
GraphRAG - 知识图谱增强检索生成系统
用于母婴出海电商的智能问答和推荐

功能：
1. 知识图谱构建（商品、用户、行为）
2. 结构化检索（实体匹配、邻居扩展、路径搜索）
3. 混合评分（语义相似度 + 图结构相关性）
4. 上下文组装（生成 LLM 可用的结构化上下文）

Author: paper2skills
Date: 2026-04-06
"""

import numpy as np
from typing import List, Dict, Tuple, Set, Optional
from collections import defaultdict, deque
import json
import re


class KnowledgeGraph:
    """知识图谱基础类"""
    
    def __init__(self):
        self.entities = {}  # entity_id -> {type, name, attributes}
        self.relations = []  # [(source, relation_type, target, weight)]
        self.entity_index = defaultdict(set)  # entity_type -> {entity_ids}
        self.adjacency = defaultdict(list)  # entity_id -> [(target, relation, weight)]
        
    def add_entity(self, entity_id: str, entity_type: str, name: str, 
                   attributes: Dict = None):
        """添加实体"""
        self.entities[entity_id] = {
            'id': entity_id,
            'type': entity_type,
            'name': name,
            'attributes': attributes or {}
        }
        self.entity_index[entity_type].add(entity_id)
        
    def add_relation(self, source: str, relation_type: str, target: str, 
                     weight: float = 1.0, attributes: Dict = None):
        """添加关系"""
        rel = {
            'source': source,
            'type': relation_type,
            'target': target,
            'weight': weight,
            'attributes': attributes or {}
        }
        self.relations.append(rel)
        self.adjacency[source].append((target, relation_type, weight))
        
    def get_neighbors(self, entity_id: str, relation_type: str = None,
                      max_depth: int = 1) -> Dict[str, List[Tuple]]:
        """
        获取邻居节点
        
        Returns:

8. 论文来源

• 2404.16130