系列目录：本文是「AI 应用开发进阶实战」系列的第 3 篇。前两篇我们分别构建了生产级 RAG 系统和 MCP 工具链，本篇进入更高阶的领域：用知识图谱增强 LLM 的推理能力。

---

一、向量检索的"天花板"

1.1 一个真实的失败案例

用户问：“张三投资了哪些公司？这些公司的 CEO 分别是谁？他们之间有没有共同投资人？”

向量检索能做的：
  ✅ "张三 投资 公司" → 找到相关文档片段
  ❌ 但无法回答"共同投资人"——这需要关联多跳实体关系

Naive RAG 的工作流程：
  query → embedding → top-5 docs → LLM → "根据文档，张三投资了A公司和B公司，
  CEO分别是... 抱歉，文档中没有共同投资人的信息"
  
  问题：相关信息分散在多个文档中，向量检索无法发现它们的关联！

1.2 传统 RAG 的三大盲区

盲区	示例问题	为什么向量检索失败
多跳推理	“A 公司 CEO 的投资人还投资了哪些 AI 公司？”	需要 A→CEO→投资人→投资的AI公司，4跳推理
全局总结	“整个知识库中反复出现的主题是什么？”	需要聚合所有文档，而非检索 top-k
关系发现	“哪些人和张三有间接合作关系？”	需要遍历实体关系图，而非文本相似度

这些问题的本质是：文档之间的关联不是线性的文本相似度，而是复杂的实体关系图。

---

二、Graph RAG 的两种架构

2.1 架构对比

方案 A: Microsoft GraphRAG（全局图总结）
  
  文档 → 实体/关系抽取 → 构建知识图谱 → 社区检测(Leiden)
       → 每个社区生成摘要 → 检索社区摘要 → LLM 回答
  
  优势：全局视角，适合"总结""概览"类问题
  劣势：成本高（大量 LLM 调用），图更新需要重建


方案 B: GraphCypherQAChain（图查询增强）
  
  问题 → LLM 生成 Cypher 查询 → Neo4j 执行 → 返回子图
       → LLM 基于子图回答
  
  优势：灵活，适合"关系查询""路径发现"类问题
  劣势：依赖 Neo4j，LLM 写 Cypher 可能出错

本文结合两种方案，构建一个既能做全局分析又能做精确图查询的系统。

---

三、从零构建知识图谱

3.1 实体和关系抽取

# kg_builder.py
from openai import OpenAI
from neo4j import GraphDatabase
import json
from typing import List, Dict, Tuple


class KnowledgeGraphBuilder:
    """从非结构化文本构建知识图谱"""
    
    def __init__(
        self,
        neo4j_uri: str = "bolt://localhost:7687",
        neo4j_user: str = "neo4j",
        neo4j_password: str = "password",
        openai_api_key: str = None
    ):
        self.driver = GraphDatabase.driver(
            neo4j_uri, 
            auth=(neo4j_user, neo4j_password)
        )
        self.llm = OpenAI(api_key=openai_api_key)
    
    def extract_entities_relations(self, text: str) -> dict:
        """用 LLM 从文本中抽取实体和关系"""
        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "system",
                "content": """从文本中抽取实体和关系，返回严格 JSON 格式。

实体类型：PERSON（人）、ORG（组织/公司）、PRODUCT（产品）、TECH（技术）、EVENT（事件）
关系类型：WORKS_AT（就职于）、INVESTS_IN（投资）、FOUNDED（创立）、
         COMPETES_WITH（竞争）、USES（使用）、ACQUIRED（收购）

输出格式（不要Markdown代码块）：
{
  "entities": [
    {"name": "张三", "type": "PERSON", "description": "天使投资人"}
  ],
  "relations": [
    {"source": "张三", "target": "AI未来公司", "relation": "INVESTS_IN", "evidence": "张三领投了AI未来公司的A轮"}
  ]
}"""
            }, {
                "role": "user",
                "content": text
            }],
            temperature=0.1
        )
        
        result_text = response.choices[0].message.content
        
        # 清理 Markdown 标记
        if result_text.startswith("```"):
            result_text = result_text.split("\n", 1)[1]
            if result_text.endswith("```"):
                result_text = result_text[:-3]
        
        return json.loads(result_text)
    
    def build_graph(self, documents: List[str], batch_size: int = 5):
        """从文档集合构建知识图谱"""
        print(f"Building KG from {len(documents)} documents...")
        
        all_entities = {}
        all_relations = []
        
        # 批量处理
        for i in range(0, len(documents), batch_size):
            batch = documents[i:i+batch_size]
            combined = "\n\n---\n\n".join(batch)
            
            result = self.extract_entities_relations(combined)
            
            # 合并实体（去重 + 合并描述）
            for entity in result.get("entities", []):
                name = entity["name"]
                if name in all_entities:
                    # 保留更长的描述
                    if len(entity.get("description", "")) > len(all_entities[name].get("description", "")):
                        all_entities[name] = entity
                else:
                    all_entities[name] = entity
            
            all_relations.extend(result.get("relations", []))
            
            print(f"  Batch {i//batch_size + 1}: "
                  f"{len(result.get('entities',[]))} entities, "
                  f"{len(result.get('relations',[]))} relations")
        
        # 写入 Neo4j
        self._write_to_neo4j(all_entities, all_relations)
        
        return len(all_entities), len(all_relations)
    
    def _write_to_neo4j(self, entities: Dict, relations: List):
        """批量写入 Neo4j"""
        with self.driver.session() as session:
            # 创建实体节点
            for name, entity in entities.items():
                session.run(
                    f"""
                    MERGE (n:{entity['type']} {{name: $name}})
                    SET n.description = $desc
                    """,
                    name=name,
                    desc=entity.get("description", "")
                )
            
            # 创建关系边
            for rel in relations:
                session.run(
                    f"""
                    MATCH (a {{name: $source}})
                    MATCH (b {{name: $target}})
                    MERGE (a)-[r:{rel['relation']}]->(b)
                    SET r.evidence = $evidence
                    """,
                    source=rel["source"],
                    target=rel["target"],
                    evidence=rel.get("evidence", "")
                )
        
        print(f"Written to Neo4j: {len(entities)} nodes, {len(relations)} edges")
    
    def close(self):
        self.driver.close()


# 测试数据
sample_docs = [
    "张三是一位活跃的天使投资人，专注于AI和科技领域。他投资了AI未来、深度科技和智云数据三家公司。",
    "AI未来公司由李华于2020年创立，专注于大模型应用开发。公司CTO王强曾就职于Google。",
    "深度科技与AI未来在NLP领域存在竞争关系。深度科技的CEO赵敏同时也是多家AI创业公司的顾问。",
    "智云数据的核心技术是知识图谱，他们使用Neo4j作为底层存储。该公司与AI未来有战略合作。",
    "投资人刘伟和张三共同投资了深度科技。刘伟还在2023年投资了两家AI创业公司。"
]

# 构建图谱
builder = KnowledgeGraphBuilder(
    neo4j_password="your_password",
    openai_api_key="sk-xxx"
)
entity_count, rel_count = builder.build_graph(sample_docs)
print(f"\nKnowledge Graph: {entity_count} entities, {rel_count} relations")

3.2 知识图谱可视化

构建完成后，在 Neo4j Browser 中运行：

// 查看投资关系网络
MATCH (p:PERSON)-[r:INVESTS_IN]->(c:ORG)
RETURN p, r, c

// 查看竞争格局
MATCH (a:ORG)-[r:COMPETES_WITH]->(b:ORG)
RETURN a, r, b

// 2跳关系：找到张三的间接关联
MATCH (p:PERSON {name: "张三"})-[*1..2]-(connected)
RETURN p, connected

生成的图谱大概是这样的：

        张三 ──INVESTS_IN──→ AI未来 ←──FOUNDED── 李华
         │                      │
         │                   COMPETES_WITH
         │                      │
         ├──INVESTS_IN──→ 深度科技 ←──CEO── 赵敏
         │      ↑              │
         │      └──INVESTS_IN──┘
         │             刘伟
         │
         └──INVESTS_IN──→ 智云数据

---

四、GraphCypherQAChain：用图回答关系问题

4.1 让 LLM 写 Cypher

# graph_qa.py
from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
from langchain_openai import ChatOpenAI


class GraphQA:
    """基于知识图谱的问答系统"""
    
    def __init__(
        self,
        neo4j_uri: str = "bolt://localhost:7687",
        neo4j_user: str = "neo4j",
        neo4j_password: str = "password",
        openai_api_key: str = None
    ):
        self.graph = Neo4jGraph(
            url=neo4j_uri,
            username=neo4j_user,
            password=neo4j_password
        )
        
        # 刷新图 schema（让 LLM 知道有哪些实体类型和关系类型）
        self.graph.refresh_schema()
        print(f"Graph Schema:\n{self.graph.schema}")
        
        self.chain = GraphCypherQAChain.from_llm(
            llm=ChatOpenAI(
                model="gpt-4o",
                api_key=openai_api_key,
                temperature=0
            ),
            graph=self.graph,
            verbose=True,
            validate_cypher=True,  # 执行前验证 Cypher 语法
            top_k=10               # 返回最多 10 条结果
        )
    
    def ask(self, question: str) -> str:
        """向知识图谱提问"""
        try:
            result = self.chain.invoke({"query": question})
            return result["result"]
        except Exception as e:
            return f"查询失败: {e}\n可能是 Cypher 语法错误，请尝试换一种问法。"


# === 测试 ===
qa = GraphQA(neo4j_password="your_password", openai_api_key="sk-xxx")

questions = [
    "张三投资了哪些公司？",
    "谁和谁有竞争关系？",
    "哪些人和张三共同投资了同一家公司？",  # 多跳推理！
    "深度科技的相关公司有哪些？",          # 2跳关系
]

for q in questions:
    print(f"\nQ: {q}")
    answer = qa.ask(q)
    print(f"A: {answer}")

4.2 GraphCypherQAChain 的工作原理

用户: "哪些人和张三共同投资了同一家公司？"
          ↓
LLM:  生成 Cypher 查询
       MATCH (p1:PERSON {name: "张三"})-[:INVESTS_IN]->(c:ORG)
             <-[:INVESTS_IN]-(p2:PERSON)
       WHERE p1 <> p2
       RETURN p2.name, c.name
          ↓
Neo4j: 执行查询 → [(刘伟, 深度科技)]
          ↓
LLM:   基于结果回答
       "刘伟和张三共同投资了深度科技"

4.3 处理 LLM 生成错误 Cypher

class RobustGraphQA(GraphQA):
    """带错误恢复的 GraphQA"""
    
    def ask(self, question: str, max_retries: int = 3) -> str:
        """带重试和自动修正的查询"""
        
        for attempt in range(max_retries):
            try:
                result = self.chain.invoke({"query": question})
                return result["result"]
                
            except Exception as e:
                error_msg = str(e)
                
                if attempt == max_retries - 1:
                    return f"查询失败（已重试{max_retries}次）: {error_msg}"
                
                # 尝试自动修复：让 LLM 重写 Cypher
                print(f"  Retry {attempt+1}: {error_msg[:100]}...")
                
                self.chain = GraphCypherQAChain.from_llm(
                    llm=ChatOpenAI(model="gpt-4o", temperature=0),
                    graph=self.graph,
                    verbose=True,
                    validate_cypher=True,
                    top_k=5
                )
        
        return "查询失败"

---

五、Microsoft GraphRAG 风格的全局总结

5.1 社区检测与摘要生成

# community_summary.py
import networkx as nx
from collections import defaultdict


class CommunitySummarizer:
    """社区检测 + 摘要生成"""
    
    def __init__(self, neo4j_driver, openai_api_key: str):
        self.driver = neo4j_driver
        self.llm = OpenAI(api_key=openai_api_key)
    
    def export_to_networkx(self) -> nx.Graph:
        """从 Neo4j 导出为 NetworkX 图"""
        G = nx.Graph()
        
        with self.driver.session() as session:
            # 获取所有节点
            result = session.run("MATCH (n) RETURN n.name, labels(n) as labels, n.description as desc")
            for record in result:
                G.add_node(
                    record["n.name"],
                    labels=record["labels"],
                    description=record["desc"] or ""
                )
            
            # 获取所有关系
            result = session.run("""
                MATCH (a)-[r]->(b) 
                RETURN a.name, b.name, type(r) as rel_type
            """)
            for record in result:
                G.add_edge(
                    record["a.name"],
                    record["b.name"],
                    relation=record["rel_type"]
                )
        
        return G
    
    def detect_communities(self, G: nx.Graph) -> dict:
        """用 Louvain 算法检测社区"""
        from networkx.algorithms.community import louvain_communities
        
        communities = louvain_communities(G, seed=42)
        
        community_map = {}
        for i, community in enumerate(communities):
            for node in community:
                community_map[node] = i
        
        # 统计每个社区的实体
        community_entities = defaultdict(list)
        for node, comm_id in community_map.items():
            community_entities[comm_id].append(node)
        
        return {
            "map": community_map,
            "entities": dict(community_entities),
            "count": len(communities)
        }
    
    def generate_community_summary(
        self,
        community_entities: List[str],
        G: nx.Graph
    ) -> str:
        """为每个社区生成摘要"""
        
        # 收集社区内的实体信息
        entities_info = []
        for entity in community_entities:
            desc = G.nodes[entity].get("description", "")
            labels = G.nodes[entity].get("labels", [])
            
            # 收集该实体的关系
            relations = []
            for neighbor in G.neighbors(entity):
                edge_data = G.get_edge_data(entity, neighbor)
                relations.append(
                    f"  - {edge_data.get('relation','related_to')} → {neighbor}"
                )
            
            entities_info.append(
                f"{entity} ({', '.join(labels)}): {desc}\n" +
                "\n".join(relations[:5])
            )
        
        community_text = "\n\n".join(entities_info)
        
        # 让 LLM 生成摘要
        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "system",
                "content": """你是知识图谱分析专家。给出的社区实体及其关系，
生成 150 字以内的社区摘要，描述这个社区的主题、关键实体和核心关系。"""
            }, {
                "role": "user",
                "content": f"请总结以下知识图谱社区：\n\n{community_text}"
            }],
            temperature=0.3
        )
        
        return response.choices[0].message.content
    
    def global_search(self, question: str, top_communities: int = 3) -> str:
        """全局搜索：先检索相关社区摘要，再回答"""
        # 1. 导出图 + 检测社区
        G = self.export_to_networkx()
        communities = self.detect_communities(G)
        
        # 2. 生成所有社区摘要（可以缓存）
        summaries = {}
        for comm_id, entities in communities["entities"].items():
            summaries[comm_id] = self.generate_community_summary(entities, G)
        
        # 3. 用向量检索找到最相关的社区摘要
        relevant = self._find_relevant_communities(question, summaries, top_communities)
        
        # 4. 用 LLM 基于相关社区摘要回答
        context = "\n\n---\n\n".join(relevant.values())
        
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": f"""基于以下知识图谱的社区摘要回答用户问题。
这些摘要是从大量文档中聚合提取的全局视图。

社区摘要：
{context}"""
            }, {
                "role": "user",
                "content": question
            }]
        )
        
        return response.choices[0].message.content
    
    def _find_relevant_communities(
        self,
        question: str,
        summaries: dict,
        top_k: int
    ) -> dict:
        """用简单的关键词匹配找到相关社区"""
        scored = []
        for comm_id, summary in summaries.items():
            # 简单的关键词重叠打分
            keywords = set(question.lower().split())
            summary_words = set(summary.lower().split())
            score = len(keywords & summary_words)
            scored.append((comm_id, score))
        
        scored.sort(key=lambda x: x[1], reverse=True)
        
        return {
            comm_id: summaries[comm_id]
            for comm_id, _ in scored[:top_k]
        }

---

六、混合 Graph + Vector RAG

最佳实践：Graph RAG 和 Vector RAG 不是替代关系，而是互补关系。

class HybridGraphVectorRAG:
    """混合图检索 + 向量检索"""
    
    def __init__(self, graph_qa: GraphQA, vector_rag, summarizer: CommunitySummarizer):
        self.graph_qa = graph_qa
        self.vector_rag = vector_rag
        self.summarizer = summarizer
    
    def route_and_answer(self, question: str) -> dict:
        """智能路由：根据问题类型选择检索策略"""
        
        # Step 1: LLM 判断问题类型
        question_type = self._classify_question(question)
        
        results = {"type": question_type}
        
        if question_type == "relationship":
            # 关系查询 → 图检索
            results["graph_answer"] = self.graph_qa.ask(question)
            results["answer"] = results["graph_answer"]
        
        elif question_type == "global":
            # 全局总结 → 社区摘要检索
            results["community_answer"] = self.summarizer.global_search(question)
            results["answer"] = results["community_answer"]
        
        elif question_type == "hybrid":
            # 混合查询 → 两路检索 + 融合
            graph_result = self.graph_qa.ask(question)
            vector_docs = self.vector_rag.retrieve(question, top_k=3)
            
            # LLM 融合两个来源
            results["answer"] = self._fuse_results(
                question, graph_result, vector_docs
            )
        
        else:
            # 事实查询 → 向量检索
            docs = self.vector_rag.retrieve(question, top_k=5)
            results["answer"] = self.vector_rag.generate(question, docs)
        
        return results
    
    def _classify_question(self, question: str) -> str:
        """LLM 分类问题类型"""
        response = self.llm.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "system",
                "content": """将问题分类为以下之一：
- relationship: 涉及实体间关系、多跳推理（谁投资了谁、谁和谁竞争）
- global: 需要全局视角、总结、概览（整个知识库的主题、趋势）
- fact: 简单事实查询（某个概念的定义、某个公司做什么）
- hybrid: 既有关系查询又需要背景知识

只回复一个词。"""
            }, {
                "role": "user",
                "content": question
            }],
            temperature=0
        )
        return response.choices[0].message.content.strip().lower()
    
    def _fuse_results(self, question: str, graph_result: str, vector_docs: list) -> str:
        """融合图检索和向量检索的结果"""
        vector_context = "\n\n".join([d.page_content for d in vector_docs])
        
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": f"""你同时拥有两个信息源：
1. 知识图谱查询结果（精确的关系信息）
2. 文档检索结果（详细的背景信息）

综合两者回答用户问题。优先采信图谱的关系信息，用文档内容补充细节。

图谱结果：{graph_result}

文档上下文：{vector_context}"""
            }, {
                "role": "user",
                "content": question
            }]
        )
        return response.choices[0].message.content

---

七、总结与选型指南

                     你的问题
 
              ┌───────┼───────┐
              │       │       │
         关系/多跳  全局总结  简单事实
              │       │       │
         Graph RAG  Community  Vector RAG
        (Cypher)    Summary    (Hybrid)
              │       │       │
              └───────┼───────┘
                      │
                  LLM 融合

方案	适用问题	核心技术	延迟	成本
Vector RAG	事实查询	Embedding + ANN	低	低
GraphCypher	关系/多跳	LLM→Cypher→Neo4j	中	中
Community Summary	全局总结	Leiden + LLM摘要	高	高
Hybrid	复杂综合	智能路由+融合	中-高	中-高

实践建议

1. 先从 Vector RAG 起步，覆盖 80% 的查询
2. 当用户开始问关系类问题时，引入 Graph RAG
3. 知识库超过 10 万文档时，加入 Community Summary 做全局分析
4. 不要过早优化——Graph RAG 的构建和维护成本是 Vector RAG 的 5-10 倍

下一篇：AI Agent 工作流引擎——DAG 编排与动态路由。我们将离开"检索"领域，进入 Agent 的核心：如何设计可靠、可观测的工作流引擎。

---

本文完整代码已开源。下一篇：工作流引擎（即将发布）