graphrag-patterns

GraphRAG 模式

将知识图谱与 RAG 结合，用于对关联数据进行复杂推理。

何时使用 GraphRAG 而非 Vector RAG

使用场景	Vector RAG	GraphRAG
简单问答	✅	杀鸡用牛刀
事实查找	✅	✅
多跳推理	❌	✅
“X 与 Y 有什么关系？”	❌	✅
实体关系	❌	✅
合规性／审计跟踪	❌	✅
主题摘要	❌	✅

核心架构

┌─────────────────────────────────────────────────────────────┐
│                      User Query                              │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    Query Analyzer                            │
│         (Determine: vector, graph, or hybrid?)               │
└─────────────────────────────────────────────────────────────┘
                    │                    │
          ┌────────┴────────┐  ┌────────┴────────┐
          ▼                 ▼  ▼                 ▼
┌─────────────────┐  ┌─────────────────┐
│  Vector Search  │  │  Graph Traverse │
│  (Semantic)     │  │  (Structured)   │
└─────────────────┘  └─────────────────┘
          │                    │
          └────────┬──────────┘
                   ▼
┌─────────────────────────────────────────────────────────────┐
│                  Context Fusion                              │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                    LLM Generation                            │
└─────────────────────────────────────────────────────────────┘

模式 1：实体提取 → 知识图谱

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from neo4j import GraphDatabase

# Step 1: Extract entities and relationships from documents
EXTRACTION_PROMPT = """Extract entities and relationships from this text.

Text: {text}

Return JSON format:
{{
  "entities": [
    {{"name": "...", "type": "Person|Organization|Concept|Event|Location"}}
  ],
  "relationships": [
    {{"source": "...", "target": "...", "type": "..."}}
  ]
}}
"""

async def extract_knowledge(text: str, llm: ChatOpenAI) -> dict:
    """Extract entities and relationships from text."""
    prompt = ChatPromptTemplate.from_template(EXTRACTION_PROMPT)
    chain = prompt | llm
    result = await chain.ainvoke({"text": text})
    return json.loads(result.content)


# Step 2: Store in Neo4j
class KnowledgeGraph:
    def __init__(self, uri: str, user: str, password: str):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def add_entity(self, name: str, entity_type: str, properties: dict = None):
        with self.driver.session() as session:
            session.run(
                f"""
                MERGE (e:{entity_type} {{name: $name}})
                SET e += $properties
                """,
                name=name,
                properties=properties or {}
            )

    def add_relationship(self, source: str, target: str, rel_type: str):
        with self.driver.session() as session:
            session.run(
                """
                MATCH (a {name: $source})
                MATCH (b {name: $target})
                MERGE (a)-[r:""" + rel_type + """]->(b)
                """,
                source=source,
                target=target
            )

模式 2：图增强检索

from llama_index.core import PropertyGraphIndex
from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore

def create_property_graph_index(documents):
    """Create a property graph index with LlamaIndex."""

    # Connect to Neo4j
    graph_store = Neo4jPropertyGraphStore(
        username="neo4j",
        password="password",
        url="bolt://localhost:7687",
    )

    # Build index - automatically extracts entities/relationships
    index = PropertyGraphIndex.from_documents(
        documents,
        property_graph_store=graph_store,
        show_progress=True,
    )

    return index


def query_with_graph(index, query: str):
    """Query using both vector and graph retrieval."""

    # Create retriever that uses both paths
    retriever = index.as_retriever(
        include_text=True,  # Include original text chunks
        similarity_top_k=5,
    )

    # Get results
    nodes = retriever.retrieve(query)
    return nodes

模式 3：将文本转换为 Cypher，直接查询图

from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain

def create_text_to_cypher_chain():
    """Create a chain that converts natural language to Cypher queries."""

    # Connect to Neo4j
    graph = Neo4jGraph(
        url="bolt://localhost:7687",
        username="neo4j",
        password="password"
    )

    # Print schema for debugging
    print(graph.schema)

    # Create chain
    chain = GraphCypherQAChain.from_llm(
        llm=ChatOpenAI(model="gpt-4", temperature=0),
        graph=graph,
        verbose=True,
        validate_cypher=True,  # Validate before executing
        return_intermediate_steps=True
    )

    return chain


# Usage
chain = create_text_to_cypher_chain()
result = chain.invoke({
    "query": "What companies has John Smith worked for?"
})
# Generated Cypher: MATCH (p:Person {name: 'John Smith'})-[:WORKED_AT]->(c:Company) RETURN c.name

模式 4：混合使用向量检索与图检索

class HybridGraphRAG:
    """Combine vector similarity with graph traversal."""

    def __init__(self, vector_store, graph_store):
        self.vector_store = vector_store
        self.graph_store = graph_store

    def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
        # 1. Vector search for relevant chunks
        vector_results = self.vector_store.similarity_search(query, k=top_k)

        # 2. Extract entities from query
        query_entities = self._extract_entities(query)

        # 3. Graph traversal from those entities
        graph_context = []
        for entity in query_entities:
            # Get 1-hop neighbors
            neighbors = self.graph_store.query(f"""
                MATCH (e {{name: '{entity}'}})-[r]-(n)
                RETURN e.name, type(r), n.name, n.description
                LIMIT 10
            """)
            graph_context.extend(neighbors)

        # 4. Combine results
        combined = {
            "vector_chunks": [r.page_content for r in vector_results],
            "graph_context": graph_context,
            "entities": query_entities
        }

        return combined

    def _extract_entities(self, text: str) -> list[str]:
        # Use NER or LLM to extract entities
        # Simplified version:
        prompt = f"Extract entity names from: {text}"
        # ... LLM call
        return entities

模式 5：Microsoft GraphRAG（社区检测）

# Microsoft's GraphRAG approach uses community detection
# for global summarization queries

from graphrag.index import run_indexing
from graphrag.query import LocalSearch, GlobalSearch

# Index documents (creates communities)
await run_indexing(
    input_dir="./documents",
    output_dir="./index",
    config={
        "llm": {"model": "gpt-4"},
        "embeddings": {"model": "text-embedding-3-small"},
        "chunks": {"size": 300, "overlap": 100},
        "community_detection": {
            "algorithm": "leiden",
            "resolution": 1.0
        }
    }
)

# Local search (specific entity questions)
local = LocalSearch(index_dir="./index")
result = local.search("What is Company X's main product?")

# Global search (summarization across communities)
global_search = GlobalSearch(index_dir="./index")
result = global_search.search("What are the main themes in these documents?")

各模式的适用场景

模式	适用场景
实体提取 → KG	从头构建、自定义架构
Property Graph Index	快速设置、LlamaIndex 生态系统
文本转 Cypher	已有图、复杂查询
混合向量检索与图检索	同时需要语义与结构信息
Microsoft GraphRAG	大型语料库、摘要查询

最佳实践

1. 定义架构 - 明确哪些实体和关系很重要
2. 从简单开始 - 先从 2-3 种实体类型开始，再按需扩展
3. 验证 Cypher - 执行前务必验证生成的查询
4. 缓存图查询 - 图遍历可能开销很大
5. 结合向量检索 - 纯图检索会遗漏语义相似性
6. 测试多跳查询 - 确保 2-3 跳查询的性能可以接受

常见问题

• 提取过度：实体过多会导致图谱充满噪声
• 缺少关系：没有连接的实体毫无用处
• 架构漂移：实体类型不一致会导致查询出错
• 没有回退方案：找不到实体时，纯图检索会失败

工具与资源

• Neo4j：用于生产环境的图数据库
• LlamaIndex PropertyGraphIndex：便于集成 Python
• Microsoft GraphRAG：基于社区的方法
• Amazon Neptune：托管式图数据库
• LangChain GraphCypherQAChain：文本转 Cypher 链