Knowledge Graph Builder

ナレッジグラフビルダー

関係性に基づく知識を通じて、AI システムの性能を高める構造化ナレッジグラフを構築します。

基本原則

ナレッジグラフは暗黙的な関係を明示化し、AI システムがつながりについて推論し、事実を検証し、ハルシネーションを回避できるようにします。

ナレッジグラフを使用する場面

ナレッジグラフを使用する場面：

• ✅ 複雑なエンティティ間の関係がドメインの中心となっている
• ✅ 構造化された知識と照合して、AI が生成した事実を検証する必要がある
• ✅ セマンティック検索と関係性のたどりが必要である
• ✅ データに豊富な相互関係がある（人物、組織、製品）
• ✅ 「X と Y はどのような関係にあるか？」という質問に答える必要がある
• ✅ 関係性に基づくレコメンデーションシステムを構築する
• ✅ つながりのあるデータ全体を対象とした不正検出やパターン認識を行う

ナレッジグラフを使用しない場面：

• ❌ 単純な表形式データの場合（リレーショナル DB を使用）
• ❌ ドキュメントのみを対象とした検索の場合（ベクトル DB を使った RAG を使用）
• ❌ エンティティ間に重要な関係がない
• ❌ チームにグラフモデリングの専門知識がない
• ❌ たどりが不要で読み取り中心のワークロードの場合（従来型の DB を使用）

---

ナレッジグラフ実装の 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 クエリ
• 短所：コミュニティが小さい、グラフ固有の機能が少ない
• 使用場面：ドキュメント DB とグラフを 1 つのシステムで扱う必要がある場合

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

LLM ベースの抽出（複雑な関係向け）：

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 との統合とハルシネーションの防止

目標：ナレッジグラフを使って LLM の応答を根拠付けし、ハルシネーションを検出します

ナレッジグラフ 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 の埋め込み
• API 層用の FastAPI

本番環境（エンティティ数が 10K～1M）の場合：

• Neo4j Enterprise または ArangoDB
• ベクトル検索用の Pinecone
• ETL 用の Airflow
• GraphQL API

エンタープライズ環境（エンティティ数が 1M 以上）の場合：

• Neo4j Enterprise または TigerGraph
• 分散型ベクトル DB（Pinecone、Weaviate）
• ストリーミング用の Kafka
• Kubernetes へのデプロイ

---

検証チェックリスト

• オントロジーを設計し、ドメインエキスパートと検証した
• グラフデータベースを選定してセットアップした
• エンティティ抽出パイプラインをテストした（精度 >85%）
• 関係抽出を検証した
• ハイブリッド検索（グラフ + ベクトル）を実装した
• クエリ API を作成し、文書化した
• AI との統合をテストした（RAG またはハルシネーション検出）
• パフォーマンスベンチマークを達成した（一般的なパターンのクエリ <100ms）
• データ品質の監視を導入した
• バックアップと復旧をテストした

---

関連リソース

関連スキル：

• rag-implementer - ハイブリッド KG+RAG システム向け
• multi-agent-architect - ナレッジグラフを活用するエージェント向け
• api-designer - KG API の設計向け

関連パターン：

• META/DECISION-FRAMEWORK.md - グラフ DB の選定
• STANDARDS/architecture-patterns/knowledge-graph-pattern.md - KG アーキテクチャ（作成時）

関連プレイブック：

• PLAYBOOKS/deploy-neo4j.md - Neo4j のデプロイ（作成時）
• PLAYBOOKS/build-kg-rag-system.md - KG-RAG の統合（作成時）