Knowledge Graph Builder

知识图谱构建器

通过关系知识构建结构化知识图谱，提升 AI 系统性能。

核心原则

知识图谱将隐含关系显式化，使 AI 系统能够推理关联、验证事实并避免幻觉。

何时使用知识图谱

适合使用知识图谱的情况：

• ✅ 复杂的实体关系是你的领域核心
• ✅ 需要根据结构化知识验证 AI 生成的事实
• ✅ 需要语义搜索和关系遍历
• ✅ 数据包含丰富的相互关联（人员、组织、产品）
• ✅ 需要回答“X 和 Y 有什么关系？”之类的查询
• ✅ 基于关系构建推荐系统
• ✅ 在关联数据中检测欺诈或识别模式

不适合使用知识图谱的情况：

• ❌ 简单的表格数据（使用关系型数据库）
• ❌ 纯文档搜索（使用结合向量数据库的 RAG）
• ❌ 实体之间没有显著关系
• ❌ 团队缺乏图建模专业知识
• ❌ 以读取为主且无需遍历的工作负载（使用传统数据库）

---

知识图谱实施的 6 个阶段

阶段 1：本体设计

目标：为你的领域定义实体、关系和属性

实体类型（节点）：

• 人员、组织、地点、产品、概念、事件、文档

关系类型（边）：

• 层级关系：IS_A、PART_OF、REPORTS_TO
• 关联关系：WORKS_FOR、LOCATED_IN、AUTHORED_BY、RELATED_TO
• 时间关系：CREATED_ON、OCCURRED_BEFORE、OCCURRED_AFTER

属性（特征）：

• 节点属性：id、name、type、created_at、metadata
• 边属性：type、confidence、source、timestamp

本体示例：

# RDF/Turtle format
@prefix : <http://example.org/ontology#> .

:Person a owl:Class ;
    rdfs:label "Person" .

:Organization a owl:Class ;
    rdfs:label "Organization" .

:worksFor a owl:ObjectProperty ;
    rdfs:domain :Person ;
    rdfs:range :Organization ;
    rdfs:label "works for" .

验证：

• 实体涵盖所有领域概念
• 关系能够捕捉关键关联
• 与领域专家共同审查本体
• 定义分类层级（is-a 关系）

---

阶段 2：图数据库选择

决策矩阵：

Neo4j（大多数情况下推荐）：

• 优点：成熟、使用 Cypher 查询语言、提供图算法、可视化效果出色
• 缺点：企业版许可费用高、扩展复杂
• 适用场景：需要复杂查询和图算法，且团队能够学习 Cypher

Amazon Neptune：

• 优点：托管服务、支持 Gremlin 和 SPARQL、可与 AWS 集成
• 缺点：受供应商锁定、比可自托管方案更昂贵
• 适用场景：使用 AWS 基础设施、需要托管服务或有合规要求

ArangoDB：

• 优点：多模型（图 + 文档 + 键值）、支持 JavaScript 查询
• 缺点：社区规模较小、图专属功能较少
• 适用场景：需要在一个系统中同时使用文档数据库和图数据库

TigerGraph：

• 优点：深度遍历性能最佳、支持并行处理
• 缺点：设置复杂、学习难度较高
• 适用场景：超大规模图（数十亿条边）、实时分析

技术栈：

graph_database: 'Neo4j Community' # or Enterprise for production
vector_integration: 'Pinecone' # For hybrid search
embeddings: 'text-embedding-3-large' # OpenAI
etl: 'Apache Airflow' # For data pipelines

Neo4j 模式设置：

// Create constraints for uniqueness
CREATE CONSTRAINT person_id IF NOT EXISTS
FOR (p:Person) REQUIRE p.id IS UNIQUE;

CREATE CONSTRAINT org_name IF NOT EXISTS
FOR (o:Organization) REQUIRE o.name IS UNIQUE;

// Create indexes for performance
CREATE INDEX entity_search IF NOT EXISTS
FOR (e:Entity) ON (e.name, e.type);

CREATE INDEX relationship_type IF NOT EXISTS
FOR ()-[r:RELATED_TO]-() ON (r.type, r.confidence);

---

阶段 3：实体提取与关系构建

目标：从数据源中提取实体和关系

数据源：

• 结构化：数据库、API、CSV 文件
• 非结构化：文档、网页内容、文本文件
• 半结构化：JSON、XML、知识库

实体提取流水线：

class EntityExtractionPipeline:
    def __init__(self):
        self.ner_model = load_ner_model()  # spaCy, Hugging Face
        self.entity_linker = EntityLinker()
        self.deduplicator = EntityDeduplicator()

    def process_text(self, text: str) -> List[Entity]:
        # 1. Extract named entities
        entities = self.ner_model.extract(text)

        # 2. Link to existing entities (entity resolution)
        linked_entities = self.entity_linker.link(entities)

        # 3. Deduplicate and resolve conflicts
        resolved_entities = self.deduplicator.resolve(linked_entities)

        return resolved_entities

关系提取：

class RelationshipExtractor:
    def extract_relationships(self, entities: List[Entity],
                            text: str) -> List[Relationship]:
        relationships = []

        # Use dependency parsing or LLM for extraction
        doc = self.nlp(text)
        for sent in doc.sents:
            rels = self.extract_from_sentence(sent, entities)
            relationships.extend(rels)

        # Validate against ontology
        valid_relationships = self.validate_relationships(relationships)
        return valid_relationships

基于大模型的提取（用于复杂关系）：

def extract_with_llm(text: str) -> List[Relationship]:
    prompt = f"""
    Extract entities and relationships from this text:
    {text}

    Format: (Entity1, Relationship, Entity2, Confidence)
    Only extract factual relationships.
    """

    response = llm.generate(prompt)
    relationships = parse_llm_response(response)
    return relationships

验证：

• 实体提取准确率 >85%
• 实体去重正常工作
• 根据本体验证关系
• 分配置信度评分

---

阶段 4：混合式知识图谱与向量架构

目标：将结构化图谱与语义向量搜索结合

架构：

class HybridKnowledgeSystem:
    def __init__(self):
        self.graph_db = Neo4jConnection()
        self.vector_db = PineconeClient()
        self.embedding_model = OpenAIEmbeddings()

    def store_entity(self, entity: Entity):
        # Store structured data in graph
        self.graph_db.create_node(entity)

        # Store embeddings in vector database
        embedding = self.embedding_model.embed(entity.description)
        self.vector_db.upsert(
            id=entity.id,
            values=embedding,
            metadata=entity.metadata
        )

    def hybrid_search(self, query: str, top_k: int = 10) -> SearchResults:
        # 1. Vector similarity search
        query_embedding = self.embedding_model.embed(query)
        vector_results = self.vector_db.query(
            vector=query_embedding,
            top_k=100
        )

        # 2. Graph traversal from vector results
        entity_ids = [r.id for r in vector_results.matches]
        graph_results = self.graph_db.get_subgraph(entity_ids, max_hops=2)

        # 3. Merge and rank results
        merged = self.merge_results(vector_results, graph_results)
        return merged[:top_k]

混合方案的优势：

• 向量搜索：语义相似性、灵活查询
• 图遍历：基于关系的推理、上下文扩展
• 结合使用：兼得两者优势

---

阶段 5：查询模式与 API 设计

常见查询模式：

1. 查找实体：

MATCH (e:Entity {id: $entity_id})
RETURN e

2. 查找关系：

MATCH (source:Entity {id: $entity_id})-[r]-(target)
RETURN source, r, target
LIMIT 20

3. 查找实体之间的路径：

MATCH path = shortestPath(
  (source:Person {id: $source_id})-[*..5]-(target:Person {id: $target_id})
)
RETURN path

4. 多跳遍历：

MATCH (p:Person {name: $name})-[:WORKS_FOR]->(o:Organization)-[:LOCATED_IN]->(l:Location)
RETURN p.name, o.name, l.city

5. 推荐查询：

// Find people similar to this person based on shared organizations
MATCH (p1:Person {id: $person_id})-[:WORKS_FOR]->(o:Organization)<-[:WORKS_FOR]-(p2:Person)
WHERE p1 <> p2
RETURN p2, COUNT(o) AS shared_orgs
ORDER BY shared_orgs DESC
LIMIT 10

知识图谱 API：

class KnowledgeGraphAPI:
    def __init__(self, graph_db):
        self.graph = graph_db

    def find_entity(self, entity_name: str) -> Entity:
        """Find entity by name with fuzzy matching"""
        query = """
        MATCH (e:Entity)
        WHERE e.name CONTAINS $name
        RETURN e
        ORDER BY apoc.text.levenshtein(e.name, $name)
        LIMIT 1
        """
        return self.graph.run(query, name=entity_name).single()

    def find_relationships(self, entity_id: str,
                         relationship_type: str = None,
                         max_hops: int = 2) -> List[Relationship]:
        """Find relationships within specified hops"""
        query = f"""
        MATCH (source:Entity {{id: $entity_id}})
        MATCH path = (source)-[r*1..{max_hops}]-(target)
        RETURN path, relationships(path) AS rels
        LIMIT 100
        """
        return self.graph.run(query, entity_id=entity_id).data()

    def get_subgraph(self, entity_ids: List[str],
                    max_hops: int = 2) -> Subgraph:
        """Get connected subgraph for multiple entities"""
        query = f"""
        MATCH (e:Entity)
        WHERE e.id IN $entity_ids
        CALL apoc.path.subgraphAll(e, {{maxLevel: {max_hops}}})
        YIELD nodes, relationships
        RETURN nodes, relationships
        """
        return self.graph.run(query, entity_ids=entity_ids).data()

---

阶段 6：AI 集成与幻觉预防

目标：使用知识图谱为大模型的回答提供依据，并检测幻觉

知识图谱 RAG：

class KnowledgeGraphRAG:
    def __init__(self, kg_api, llm_client):
        self.kg = kg_api
        self.llm = llm_client

    def retrieve_context(self, query: str) -> str:
        # Extract entities from query
        entities = self.extract_entities_from_query(query)

        # Retrieve relevant subgraph
        subgraph = self.kg.get_subgraph(
            [e.id for e in entities],
            max_hops=2
        )

        # Format subgraph for LLM
        context = self.format_subgraph_for_llm(subgraph)
        return context

    def generate_with_grounding(self, query: str) -> GroundedResponse:
        context = self.retrieve_context(query)

        prompt = f"""
        Context from knowledge graph:
        {context}

        User query: {query}

        Answer based only on the provided context. Include source entities.
        """

        response = self.llm.generate(prompt)

        return GroundedResponse(
            response=response,
            sources=self.extract_sources(context),
            confidence=self.calculate_confidence(response, context)
        )

幻觉检测：

class HallucinationDetector:
    def __init__(self, knowledge_graph):
        self.kg = knowledge_graph

    def verify_claim(self, claim: str) -> VerificationResult:
        # Parse claim into (subject, predicate, object)
        parsed_claim = self.parse_claim(claim)

        # Query knowledge graph for evidence
        evidence = self.kg.find_evidence(
            parsed_claim.subject,
            parsed_claim.predicate,
            parsed_claim.object
        )

        if evidence:
            return VerificationResult(
                is_supported=True,
                evidence=evidence,
                confidence=evidence.confidence
            )

        # Check for contradictory evidence
        contradiction = self.kg.find_contradiction(parsed_claim)

        return VerificationResult(
            is_supported=False,
            is_contradicted=bool(contradiction),
            contradiction=contradiction
        )

---

核心原则

1. 从本体开始

在摄取数据前定义好你的架构。之后更改本体的成本很高。

2. 实体消歧至关重要

积极去重实体。“Apple Inc”、“Apple”、“Apple Computer” → 同一实体。

3. 为所有内容设置置信度分数

每种关系都应有置信度分数（0.0-1.0）和来源。

4. 增量构建

不要试图一次性建模整个领域。从核心实体开始，再逐步扩展。

5. 混合架构效果最佳

结合图遍历（结构化）与向量搜索（语义），以获得最佳结果。

---

常见用例

1. 问答：

• 从问题中提取实体
• 遍历图谱以查找答案
• 返回路径作为解释

2. 推荐：

• 通过共享关系查找相似实体
• 根据关系强度排序
• 返回排名靠前的 K 项推荐

3. 欺诈检测：

• 将交易建模为图
• 查找可疑模式（环、异常）
• 标记以供审核

4. 知识发现：

• 识别隐含关系
• 建议缺失的关联
• 请领域专家验证

5. 语义搜索：

• 混合向量与图搜索
• 通过关系扩展上下文
• 返回丰富的关联结果

---

技术建议

适用于 MVP（少于 10K 个实体）：

• Neo4j Community Edition（免费）
• 使用 SQLite 存储元数据
• OpenAI 嵌入
• 使用 FastAPI 构建 API 层

适用于生产环境（10K–1M 个实体）：

• Neo4j Enterprise 或 ArangoDB
• 使用 Pinecone 进行向量搜索
• 使用 Airflow 进行 ETL
• GraphQL API

适用于企业级（超过 1M 个实体）：

• Neo4j Enterprise 或 TigerGraph
• 分布式向量数据库（Pinecone、Weaviate）
• 使用 Kafka 进行流处理
• Kubernetes 部署

---

验证清单

• 本体已设计，并经领域专家验证
• 已选择并设置图数据库
• 实体提取流水线已测试（准确率 >85%）
• 关系提取已验证
• 已实现混合搜索（图 + 向量）
• 已创建并记录查询 API
• AI 集成已测试（RAG 或幻觉检测）
• 已达到性能基准（常见模式的查询耗时 <100ms）
• 已建立数据质量监控
• 已测试备份与恢复

---

相关资源

相关技能：

• rag-implementer - 用于混合 KG+RAG 系统
• multi-agent-architect - 用于由知识图谱驱动的 Agent
• api-designer - 用于 KG API 设计

相关模式：

• META/DECISION-FRAMEWORK.md - 图数据库选择
• STANDARDS/architecture-patterns/knowledge-graph-pattern.md - 知识图谱架构（创建后提供）

相关操作手册：

• PLAYBOOKS/deploy-neo4j.md - Neo4j 部署（创建后提供）
• PLAYBOOKS/build-kg-rag-system.md - KG-RAG 集成（创建后提供）