GraphRAG 生产部署指南：知识图谱增强的 RAG 系统

┌─────────────────────────────────────────────────────┐
│                   GraphRAG 架构                      │
├─────────────────────────────────────────────────────┤
│                                                     │
│  离线构建 Pipeline                                   │
│  ┌──────────┐  ┌───────────┐  ┌──────────────┐     │
│  │ 文档解析  │→│ 实体抽取   │→│ 关系抽取      │     │
│  └──────────┘  └───────────┘  └──────────────┘     │
│                     ↓                               │
│  ┌───────────┐  ┌───────────┐  ┌──────────────┐    │
│  │ 图谱构建   │←│ 社区检测   │←│ Embedding     │    │
│  └───────────┘  └───────────┘  └──────────────┘    │
│                     ↓                               │
│              ┌─────────────┐                        │
│              │ 索引持久化   │                        │
│              └─────────────┘                        │
│                                                     │
│  在线查询 Engine                                     │
│  ┌──────────┐  ┌───────────┐  ┌──────────────┐     │
│  │ Query     │→│ 路由决策   │→│ 双路检索      │     │
│  │ 理解      │  │           │  │ 向量+图谱    │     │
│  └──────────┘  └───────────┘  └──────────────┘     │
│                                    ↓                │
│  ┌──────────┐  ┌───────────┐  ┌──────────────┐    │
│  │ 生成+引用 │←│ 信息整合   │←│ 重排序        │    │
│  └──────────┘  └───────────┘  └──────────────┘    │
│                                                     │
└─────────────────────────────────────────────────────┘

from pydantic import BaseModel
from typing import List

class Entity(BaseModel):
    name: str
    type: str  # Person, Organization, Concept, Event, etc.
    description: str
    source_chunk_id: str

class Relation(BaseModel):
    source_entity: str
    target_entity: str
    relation_type: str
    description: str
    confidence: float
    source_chunk_id: str

ENTITY_EXTRACTION_PROMPT = """
你是一个信息抽取专家。从以下文本中抽取实体和关系。

实体类型：Person, Organization, Concept, Technology, Event, Location
关系类型：works_at, created, related_to, part_of, located_in, depends_on, competes_with

文本：
{text}

请以 JSON 格式输出：
{{
  "entities": [
    {{"name": "...", "type": "...", "description": "..."}}
  ],
  "relations": [
    {{"source": "...", "target": "...", "type": "...", "description": "...", "confidence": 0.0-1.0}}
  ]
}}
"""

class EntityExtractor:
    def __init__(self, llm):
        self.llm = llm
    
    def extract(self, text: str, chunk_id: str):
        prompt = ENTITY_EXTRACTION_PROMPT.format(text=text)
        result = self.llm.generate(prompt, response_format="json")
        
        entities = [
            Entity(**e, source_chunk_id=chunk_id) 
            for e in result["entities"]
        ]
        relations = [
            Relation(**r, source_chunk_id=chunk_id) 
            for r in result["relations"]
        ]
        
        return entities, relations

import networkx as nx
from community_detection import LeidenAlgorithm

class KnowledgeGraph:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.entity_index = {}  # name -> node_id
    
    def add_entities(self, entities: List[Entity]):
        for entity in entities:
            if entity.name not in self.entity_index:
                node_id = len(self.entity_index)
                self.entity_index[entity.name] = node_id
                self.graph.add_node(
                    node_id,
                    name=entity.name,
                    type=entity.type,
                    description=entity.description,
                    source_chunks=[entity.source_chunk_id]
                )
            else:
                # 合并描述
                node_id = self.entity_index[entity.name]
                self.graph.nodes[node_id]["source_chunks"].append(
                    entity.source_chunk_id
                )
    
    def add_relations(self, relations: List[Relation]):
        for rel in relations:
            if rel.source_entity in self.entity_index and rel.target_entity in self.entity_index:
                src = self.entity_index[rel.source_entity]
                tgt = self.entity_index[rel.target_entity]
                
                if self.graph.has_edge(src, tgt):
                    # 合并关系
                    existing = self.graph[src][tgt]
                    existing["relations"].append({
                        "type": rel.relation_type,
                        "description": rel.description,
                        "confidence": rel.confidence
                    })
                else:
                    self.graph.add_edge(
                        src, tgt,
                        relations=[{
                            "type": rel.relation_type,
                            "description": rel.description,
                            "confidence": rel.confidence
                        }]
                    )
    
    def detect_communities(self):
        """使用 Leiden 算法进行社区检测"""
        undirected = self.graph.to_undirected()
        communities = LeidenAlgorithm().fit(undirected)
        
        # 为每个社区生成摘要
        for comm_id, nodes in communities.items():
            subgraph = self.graph.subgraph(nodes)
            summary = self._summarize_community(subgraph)
            
            for node in nodes:
                self.graph.nodes[node]["community_id"] = comm_id
            
            self.graph.graph.setdefault("community_summaries", {})[comm_id] = summary
        
        return communities
    
    def _summarize_community(self, subgraph):
        entities_info = []
        for node_id in subgraph.nodes():
            node = self.graph.nodes[node_id]
            entities_info.append(f"{node['name']} ({node['type']}): {node['description']}")
        
        relations_info = []
        for u, v, data in subgraph.edges(data=True):
            for rel in data["relations"]:
                relations_info.append(
                    f"{self.graph.nodes[u]['name']} --{rel['type']}--> {self.graph.nodes[v]['name']}"
                )
        
        prompt = f"""
        请总结以下知识图谱社区的关键信息：
        
        实体：
        {chr(10).join(entities_info)}
        
        关系：
        {chr(10).join(relations_info)}
        
        请生成一段简洁的摘要，涵盖主要实体和它们之间的关系。
        """
        
        return self.llm.generate(prompt)

class GraphRAGIndex:
    def __init__(self):
        self.graph = KnowledgeGraph()
        self.vector_store = MilvusIndex(dim=1024)
        self.community_store = CommunityStore()
    
    def build(self, documents: List[Document]):
        # 1. 文本分块与向量化
        chunks = document_aware_chunk(documents)
        for chunk in chunks:
            embedding = embed_model.encode(chunk.text)
            self.vector_store.add(id=chunk.id, embedding=embedding, 
                                  metadata={"text": chunk.text})
        
        # 2. 实体关系抽取
        all_entities = []
        all_relations = []
        for chunk in chunks:
            entities, relations = extractor.extract(chunk.text, chunk.id)
            all_entities.extend(entities)
            all_relations.extend(relations)
        
        # 3. 构建知识图谱
        self.graph.add_entities(all_entities)
        self.graph.add_relations(all_relations)
        
        # 4. 社区检测与摘要
        communities = self.graph.detect_communities()
        
        # 5. 持久化
        self._persist()
    
    def _persist(self):
        # 图谱存 Neo4j 或 NetworkX pickle
        nx.write_gpickle(self.graph.graph, "graph.gpickle")
        
        # 向量索引已在 Milvus 中持久化
        # 社区摘要存数据库
        self.community_store.save(self.graph.graph.graph.get("community_summaries", {}))

class GraphRAGQueryEngine:
    def __init__(self, index: GraphRAGIndex, llm):
        self.index = index
        self.llm = llm
    
    def query(self, question: str) -> str:
        # 1. 判断查询类型
        query_type = self._classify_query(question)
        
        if query_type == "global":
            # 全局型问题 → 社区检索
            context = self._global_search(question)
        elif query_type == "specific":
            # 具体型问题 → 向量+图遍历
            context = self._local_search(question)
        else:
            # 混合型 → 双路检索
            context = self._hybrid_search(question)
        
        # 2. 生成答案
        answer = self.llm.generate(
            prompt=ANSWER_PROMPT.format(question=question, context=context),
            citations=True
        )
        
        return answer
    
    def _local_search(self, question: str):
        # 向量检索找到相关文本块
        query_emb = embed_model.encode(question)
        vector_hits = self.index.vector_store.search(query_emb, top_k=10)
        
        # 从命中块中提取实体
        entities = set()
        for hit in vector_hits:
            entities.update(self._extract_entities_from_chunk(hit))
        
        # 图遍历扩展上下文
        graph_context = []
        for entity_name in entities:
            if entity_name in self.index.graph.entity_index:
                node_id = self.index.graph.entity_index[entity_name]
                # 1-hop 和 2-hop 邻居
                neighbors = nx.single_source_shortest_path_length(
                    self.index.graph.graph, node_id, cutoff=2
                )
                for neighbor_id, hops in neighbors.items():
                    if hops > 0:
                        node = self.index.graph.graph.nodes[neighbor_id]
                        graph_context.append({
                            "entity": node["name"],
                            "type": node["type"],
                            "hops": hops,
                            "description": node["description"]
                        })
        
        return {
            "vector_context": vector_hits,
            "graph_context": graph_context
        }
    
    def _global_search(self, question: str):
        # 从社区摘要中检索
        community_summaries = self.index.community_store.get_all()
        
        # 用 LLM 判断哪些社区相关
        relevant = self.llm.generate(
            prompt=f"""
            以下问题与哪些社区摘要相关？
            问题：{question}
            
            社区摘要：
            {json.dumps(community_summaries, ensure_ascii=False)}
            
            返回最相关的 3 个社区 ID。
            """
        )
        
        return {"community_context": relevant}

@dataclass
class GraphRAGMetrics:
    # 构建阶段
    entity_extraction_rate: float    # 每分钟抽取实体数
    relation_extraction_rate: float  # 每分钟抽取关系数
    community_detection_time: float  # 社区检测耗时
    
    # 查询阶段
    query_latency_p50: float
    query_latency_p99: float
    vector_recall: float             # 向量检索召回率
    graph_coverage: float            # 图遍历覆盖率
    community_hit_rate: float        # 社区命中率
    
    # 质量指标
    answer_accuracy: float           # 答案准确率
    citation_rate: float             # 引用覆盖率
    hallucination_rate: float        # 幻觉率

class IncrementalUpdater:
    """GraphRAG 增量更新：只处理新增/修改的文档"""
    
    def update(self, new_documents: List[Document], deleted_ids: List[str]):
        # 1. 删除过期数据
        for doc_id in deleted_ids:
            self._remove_document(doc_id)
        
        # 2. 处理新文档
        for doc in new_documents:
            entities, relations = extractor.extract(doc)
            self.graph.add_entities(entities)
            self.graph.add_relations(relations)
        
        # 3. 局部社区检测（只重新检测受影响的社区）
        affected_communities = self._find_affected_communities(new_documents, deleted_ids)
        self._redetect_communities(affected_communities)
        
        # 4. 更新向量索引
        self.vector_store.upsert(new_documents)