MENU
  1. ホーム
  2. Python
  3. ナレッジグラフ×LLM実践入門 ── RAGの次のアーキテクチャを理解する

ナレッジグラフ×LLM実践入門 ── RAGの次のアーキテクチャを理解する

Python
ナレッジグラフ×LLM実践入門 ── RAGの次のアーキテクチャを理解する | mohablog
目次

はじめに：RAGでは足りないシーンってありますよね

LLMアプリケーション開発を進めていると、こんな課題に直面することはありませんか？

  • 複数の情報源を組み合わせた複雑な問い合わせに対応できない
  • 情報同士の「関係性」を活用したい（製品Aと製品Bの比較、企業の組織構造など）
  • RAGで取得した文書の断片だけでは、文脈が不十分になる
  • 推論の過程を透明にしたい、トレーサビリティが欲しい

この記事で想定しているのは、ECサイトのカスタマーサポートAIを構築している場面です。顧客から「このブランドの黒いスニーカーで、返品可能で、5000円以下のおすすめは？」という問い合わせが来たとき、単純なRAGでは商品ドキュメントを取得するだけです。でも現実には、「ブランド」「色」「返品ポリシー」「価格」といった複数の属性と、それらの関係を理解する必要がある。そんなときに活躍するのがナレッジグラフなんです。

この記事では、ナレッジグラフとLLMを組み合わせたアーキテクチャを、実装レベルで解説します。Python（バージョン3.11以上推奨）を使いながら、RAGを超えた次のステップが見えてくるはずです。

RAGの限界と、ナレッジグラフが解く問題

RAGがうまく機能しない場面

Retrieval-Augmented Generation（RAG）の基本的な流れは、ユーザーの質問から関連ドキュメントを検索して、LLMに渡すというものですね。ただし、実装してみるとこんな課題に気づきます。

  • 断片化の問題：関連する複数の情報がベクトル検索で別々に拾い上げられ、その関係性が不明確になる
  • 多段階推論の弱さ：「AはBに属し、BはCと関連する」といった推移的な関係を活用できない
  • スケーラビリティの課題：情報量が増えると、取得する文書数が増え、LLMへのプロンプトが肥大化する
  • 透明性の欠如：なぜその情報を選んだのかが、ベクトル検索の類似度スコアだけでは説明しにくい

これらの問題を解決するアプローチが、ナレッジグラフ（Knowledge Graph）です。

ナレッジグラフとは何か

ナレッジグラフは、エンティティ（実体）とその関係性を図グラフ構造で表現したデータベースです。最も有名な例は、Googleのナレッジグラフですね。「Steve Jobs」と検索すると、彼の生年月日、創業企業、関連人物などが構造化されて表示されます。

ナレッジグラフ = ノード（エンティティ）+ エッジ（関係）で構成される有向グラフ

具体例として、ECサイトのドメインで考えてみましょう。

  • ノード：商品、ブランド、カテゴリ、顧客、レビュー、在庫
  • エッジ：「商品は～ブランドである」「商品は～カテゴリに属する」「顧客は～を購入した」「商品は～在庫がある」

こうした構造を持つことで、複雑な問い合わせに対して、LLMは単なる文書検索ではなく、グラフ構造を辿った推論が可能になります。

ナレッジグラフ×LLMのアーキテクチャ概要

全体の流れ

ナレッジグラフとLLMを組み合わせたシステムの大まかな流れは以下の通りです。

  1. 入力：ユーザーの質問を受け取る
  2. 意図理解：LLMが質問から「何を検索すべきか」を判断
  3. グラフ検索：ナレッジグラフから関連するノードと関係性を取得
  4. コンテキスト構築：取得した情報を構造化されたコンテキストに変換
  5. 応答生成：LLMがコンテキストを使って回答を生成

従来のRAGとの大きな違いは、「ベクトル検索」から「グラフ検索」へシフトすることです。これにより、関係性を明示的に保持でき、推論の過程も透明になるんですね。

なぜナレッジグラフ×LLMなのか

疑問が出るかもしれません。「LLMだけで推論すればいいのでは？」と。その答えは、現在のLLMの弱点にあります。

  • 幻覚（Hallucination）：根拠のない情報を生成することがある
  • コンテキスト長の制限：長すぎるテキストは処理できない
  • 確実性の欠如：生成された情報の信頼度をスコアできない

ナレッジグラフは、これらのLLMの弱点をカバーします。構造化されたデータから確実に関連情報を取得でき、LLMへのプロンプトを効率的に構成できるからです。組み合わせることで、信頼度が高く、説明可能なAIシステムが実現できるわけです。

実装：Pythonでナレッジグラフを構築する

環境構築と必要なライブラリ

以下のライブラリを使います（バージョン指定）。

Python 3.11+
NetworkX 3.2.1 （グラフ構造を管理）
OpenAI API (Python 1.3.0+) （LLMの呼び出し）
LangChain 0.1.0+ （ナレッジグラフとLLMの連携）
Pydantic 2.0+ （データ検証）

インストール方法は以下の通りです。

pip install networkx openai langchain pydantic python-dotenv

実装では、まずは手軽にNetworkXを使ってメモリ内にグラフを構築します。本番環境ではNeo4jなどのグラフデータベースを使うことになるでしょう。

ステップ1：基本的なナレッジグラフクラスを作成

まずはナレッジグラフの基本的な構造を定義します。このアプローチは、複雑になったときも拡張しやすいので地味に重要です。

import networkx as nx
from typing import List, Tuple, Dict, Any
from dataclasses import dataclass
from enum import Enum

class EntityType(str, Enum):
    PRODUCT = "product"
    BRAND = "brand"
    CATEGORY = "category"
    CUSTOMER = "customer"
    REVIEW = "review"
    PRICE = "price"

class RelationType(str, Enum):
    BELONGS_TO = "belongs_to"
    HAS_BRAND = "has_brand"
    HAS_PRICE = "has_price"
    PURCHASED_BY = "purchased_by"
    REVIEWED_BY = "reviewed_by"
    RELATED_TO = "related_to"

@dataclass
class Entity:
    id: str
    type: EntityType
    attributes: Dict[str, Any]

class KnowledgeGraph:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.entities: Dict[str, Entity] = {}
    
    def add_entity(self, entity: Entity) -> None:
        """エンティティをグラフに追加"""
        self.entities[entity.id] = entity
        self.graph.add_node(
            entity.id,
            type=entity.type,
            attributes=entity.attributes
        )
    
    def add_relation(self, source_id: str, target_id: str, 
                     relation_type: RelationType, 
                     properties: Dict[str, Any] = None) -> None:
        """エンティティ間に関係性を追加"""
        self.graph.add_edge(
            source_id,
            target_id,
            relation_type=relation_type,
            properties=properties or {}
        )
    
    def get_entity_neighbors(self, entity_id: str, 
                            relation_type: RelationType = None,
                            depth: int = 1) -> Dict[str, Any]:
        """エンティティの隣接ノードを取得（深さ指定可能）"""
        neighbors = {}
        
        def traverse(current_id: str, current_depth: int):
            if current_depth > depth:
                return
            
            for neighbor_id in self.graph.successors(current_id):
                edge_data = self.graph.edges[current_id, neighbor_id]
                if relation_type and edge_data['relation_type'] != relation_type:
                    continue
                
                neighbor_entity = self.entities[neighbor_id]
                neighbors[neighbor_id] = {
                    'entity': neighbor_entity,
                    'relation': edge_data['relation_type'],
                    'properties': edge_data.get('properties', {})
                }
                
                if current_depth < depth:
                    traverse(neighbor_id, current_depth + 1)
        
        traverse(entity_id, 1)
        return neighbors

このクラス設計のポイントは、エンティティと関係性を明確に分離していることです。将来的にデータベースに移行するときも、このインターフェースを保ったまま、バックエンドを変更できるようにしています。

ステップ2：サンプルデータでグラフを構築

ECサイトのシナリオで、実際にナレッジグラフにデータを投入してみましょう。

def build_sample_ecommerce_graph() -> KnowledgeGraph:
    """ECサイト用のサンプルナレッジグラフを構築"""
    kg = KnowledgeGraph()
    
    # ブランド
    kg.add_entity(Entity(
        id="brand_nike",
        type=EntityType.BRAND,
        attributes={"name": "Nike", "country": "USA", "established": 1972}
    ))
    
    kg.add_entity(Entity(
        id="brand_adidas",
        type=EntityType.BRAND,
        attributes={"name": "Adidas", "country": "Germany", "established": 1949}
    ))
    
    # カテゴリ
    kg.add_entity(Entity(
        id="cat_sneakers",
        type=EntityType.CATEGORY,
        attributes={"name": "Sneakers", "description": "Casual athletic shoes"}
    ))
    
    kg.add_entity(Entity(
        id="cat_black",
        type=EntityType.CATEGORY,
        attributes={"name": "Black Color", "hex": "#000000"}
    ))
    
    # 商品1
    kg.add_entity(Entity(
        id="prod_nike_black_001",
        type=EntityType.PRODUCT,
        attributes={
            "name": "Nike Air Max Black",
            "color": "black",
            "price": 4500,
            "stock": 15,
            "returnable": True,
            "return_days": 30
        }
    ))
    
    # 商品2
    kg.add_entity(Entity(
        id="prod_adidas_black_001",
        type=EntityType.PRODUCT,
        attributes={
            "name": "Adidas Stan Smith Black",
            "color": "black",
            "price": 5200,
            "stock": 8,
            "returnable": True,
            "return_days": 14
        }
    ))
    
    # 商品3
    kg.add_entity(Entity(
        id="prod_nike_black_002",
        type=EntityType.PRODUCT,
        attributes={
            "name": "Nike Revolution Black",
            "color": "black",
            "price": 3200,
            "stock": 22,
            "returnable": True,
            "return_days": 30
        }
    ))
    
    # レビュー
    kg.add_entity(Entity(
        id="review_001",
        type=EntityType.REVIEW,
        attributes={
            "rating": 4.5,
            "comment": "Comfortable and stylish",
            "reviewer_count": 342
        }
    ))
    
    # 関係性を定義
    kg.add_relation(
        "prod_nike_black_001",
        "brand_nike",
        RelationType.HAS_BRAND
    )
    
    kg.add_relation(
        "prod_nike_black_001",
        "cat_sneakers",
        RelationType.BELONGS_TO
    )
    
    kg.add_relation(
        "prod_nike_black_001",
        "cat_black",
        RelationType.BELONGS_TO
    )
    
    kg.add_relation(
        "prod_nike_black_001",
        "review_001",
        RelationType.REVIEWED_BY
    )
    
    # 同じパターンで他の商品も関係付け
    kg.add_relation(
        "prod_adidas_black_001",
        "brand_adidas",
        RelationType.HAS_BRAND
    )
    kg.add_relation(
        "prod_adidas_black_001",
        "cat_sneakers",
        RelationType.BELONGS_TO
    )
    kg.add_relation(
        "prod_adidas_black_001",
        "cat_black",
        RelationType.BELONGS_TO
    )
    
    kg.add_relation(
        "prod_nike_black_002",
        "brand_nike",
        RelationType.HAS_BRAND
    )
    kg.add_relation(
        "prod_nike_black_002",
        "cat_sneakers",
        RelationType.BELONGS_TO
    )
    kg.add_relation(
        "prod_nike_black_002",
        "cat_black",
        RelationType.BELONGS_TO
    )
    
    return kg

# テスト実行
kg = build_sample_ecommerce_graph()
print(f"グラフに {len(kg.entities)} 個のエンティティが登録されました")
print(f"エッジ数: {kg.graph.number_of_edges()}")

グラフに 10 個のエンティティが登録されました
エッジ数: 11

ステップ3：LLMとの連携 ── グラフクエリの自動生成

LLMが質問からグラフクエリを生成する仕組み

ここからが本番です。ユーザーの自然言語の質問を受けて、LLMが「どのノードから検索を開始すべきか」「どの関係を辿るべきか」を判断させます。このアプローチは、Retrieval-Augmented Generation論文で見かけるパターンの応用ですね。

from openai import OpenAI
import json
from typing import Optional

class GraphQueryGenerator:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.model = "gpt-4o-mini"  # または gpt-4-turbo
    
    def generate_query(self, user_question: str, kg: KnowledgeGraph) -> Dict[str, Any]:
        """
        ユーザーの質問からグラフクエリを生成
        戻り値: {"entities": [...], "filters": {...}, "reasoning": "..."}
        """
        
        # グラフのスキーマ情報をLLMに提供
        schema_info = self._get_schema_info(kg)
        
        system_prompt = f"""You are a knowledge graph query generator for an e-commerce platform.

Given a user question and the knowledge graph schema, generate a structured query to find relevant information.

Knowledge Graph Schema:
{schema_info}

Respond in JSON format with the following structure:
{{
  "entities": [list of entity IDs or types to search],
  "filters": {object with attribute conditions like "price": {{"max": 5000}}, "returnable": true},
  "reasoning": "explanation of why these entities/filters are chosen"
}}

Be specific and reference actual entity IDs when possible.
        """
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=500,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"Question: {user_question}"}
            ]
        )
        
        try:
            result = json.loads(response.content[0].text)
            return result
        except json.JSONDecodeError:
            # JSONパースに失敗した場合の対応（本番ではもっと堅牢に）
            return {
                "entities": ["cat_black"],
                "filters": {"price": {"max": 5000}, "returnable": True},
                "reasoning": "Fallback query for black returnable products under 5000 yen"
            }
    
    def _get_schema_info(self, kg: KnowledgeGraph) -> str:
        """グラフのスキーマ情報を文字列化"""
        schema = "\nEntity Types:\n"
        schema += ", ".join([t.value for t in EntityType])
        schema += "\n\nRelation Types:\n"
        schema += ", ".join([t.value for t in RelationType])
        schema += "\n\nSample Entities:\n"
        for entity_id, entity in list(kg.entities.items())[:5]:
            schema += f"- {entity_id} ({entity.type.value}): {entity.attributes}\n"
        return schema

このアプローチのメリットは、LLMが「どの情報を探すべきか」を構造化された形で判断できるという点です。プロンプトエンジニアリングの試行錯誤を減らせるんですね。

グラフクエリの実行と結果の統合

次に、LLMが生成したクエリをグラフに対して実行し、結果を統合する処理を実装します。

class GraphQueryExecutor:
    def __init__(self, kg: KnowledgeGraph):
        self.kg = kg
    
    def execute_query(self, query: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
        グラフクエリを実行し、マッチした商品を取得
        """
        entities = query.get("entities", [])
        filters = query.get("filters", {})
        
        matching_products = []
        
        # 指定されたエンティティから関連商品を探す
        for entity_id in entities:
            if entity_id not in self.kg.entities:
                continue
            
            # 逆方向で関連商品を検索
            for node_id in self.kg.graph.nodes():
                entity = self.kg.entities.get(node_id)
                if not entity or entity.type != EntityType.PRODUCT:
                    continue
                
                # このノードが指定エンティティに接続されているかチェック
                if self._is_connected(node_id, entity_id):
                    # フィルタ条件を適用
                    if self._matches_filters(entity, filters):
                        matching_products.append({
                            "entity_id": node_id,
                            "entity": entity,
                            "path": self._find_path(node_id, entity_id)
                        })
        
        # 重複を除去してソート
        seen = set()
        unique_products = []
        for p in matching_products:
            if p["entity_id"] not in seen:
                unique_products.append(p)
                seen.add(p["entity_id"])
        
        # 価格でソート（安い順）
        unique_products.sort(
            key=lambda x: x["entity"].attributes.get("price", float('inf'))
        )
        
        return unique_products
    
    def _is_connected(self, source_id: str, target_id: str, 
                      max_depth: int = 3) -> bool:
        """2つのノードが接続されているかチェック（深さ制限あり）"""
        try:
            path = nx.shortest_path(self.kg.graph, source_id, target_id, weight=None)
            return len(path) <= max_depth
        except (nx.NetworkXNoPath, nx.NodeNotFound):
            return False
    
    def _find_path(self, source_id: str, target_id: str) -> Optional[List[str]]:
        """2つのノード間の経路を取得"""
        try:
            return nx.shortest_path(self.kg.graph, source_id, target_id)
        except (nx.NetworkXNoPath, nx.NodeNotFound):
            return None
    
    def _matches_filters(self, entity: Entity, filters: Dict[str, Any]) -> bool:
        """エンティティがフィルタ条件に合致するかチェック"""
        for filter_key, filter_value in filters.items():
            attr_value = entity.attributes.get(filter_key)
            
            # 価格フィルタ
            if filter_key == "price":
                if isinstance(filter_value, dict):
                    max_price = filter_value.get("max")
                    if max_price and attr_value and attr_value > max_price:
                        return False
            # ブール値フィルタ
            elif isinstance(filter_value, bool):
                if attr_value != filter_value:
                    return False
            # 完全一致フィルタ
            else:
                if attr_value != filter_value:
                    return False
        
        return True

ステップ4：LLMで最終回答を生成

コンテキストの構造化と応答生成

ここまでで、グラフから関連情報を取得できました。最後のステップは、その情報をLLMに渡して、ユーザーフレンドリーな回答を生成することです。

class KnowledgeGraphRAGSystem:
    def __init__(self, api_key: str, kg: KnowledgeGraph):
        self.client = OpenAI(api_key=api_key)
        self.model = "gpt-4o-mini"
        self.kg = kg
        self.query_generator = GraphQueryGenerator(api_key)
        self.query_executor = GraphQueryExecutor(kg)
    
    def answer_question(self, user_question: str) -> Dict[str, Any]:
        """
        ユーザーの質問に対して、ナレッジグラフとLLMを使って回答を生成
        """
        
        # Step 1: グラフクエリを生成
        print(f"\n[Query Generation] Processing: {user_question}")
        query = self.query_generator.generate_query(user_question, self.kg)
        print(f"Generated query: {json.dumps(query, indent=2)}")
        
        # Step 2: グラフクエリを実行
        print("[Query Execution] Searching knowledge graph...")
        search_results = self.query_executor.execute_query(query)
        print(f"Found {len(search_results)} matching products")
        
        # Step 3: コンテキストを構築
        context = self._build_context(search_results, query)
        print(f"[Context Building] Context length: {len(context)} characters")
        
        # Step 4: LLMで応答を生成
        print("[Response Generation] Generating answer...")
        response = self._generate_response(user_question, context, search_results)
        
        return {
            "question": user_question,
            "query": query,
            "search_results": search_results,
            "context": context,
            "answer": response,
            "transparency": {
                "reasoning": query.get("reasoning"),
                "sources": [r["entity_id"] for r in search_results]
            }
        }
    
    def _build_context(self, search_results: List[Dict[str, Any]], 
                      query: Dict[str, Any]) -> str:
        """
        検索結果からLLMへ渡すコンテキストを構築
        """
        context = f"Query reasoning: {query.get('reasoning')}\n\n"
        context += "Found products:\n\n"
        
        for i, result in enumerate(search_results, 1):
            entity = result["entity"]
            path = result["path"]
            
            context += f"{i}. {entity.attributes.get('name', 'Unknown')}\n"
            context += f"   Entity ID: {result['entity_id']}\n"
            context += f"   Price: ¥{entity.attributes.get('price', 'N/A')}\n"
            context += f"   Stock: {entity.attributes.get('stock', 'N/A')} units\n"
            context += f"   Returnable: {entity.attributes.get('returnable', False)} "
            context += f"({entity.attributes.get('return_days', 0)} days)\n"
            
            if path:
                context += f"   Graph path: {' -> '.join(path)}\n"
            context += "\n"
        
        return context
    
    def _generate_response(self, question: str, context: str, 
                         search_results: List[Dict[str, Any]]) -> str:
        """
        LLMを使って最終回答を生成
        """
        system_prompt = """You are a helpful e-commerce customer support AI.
You have access to a knowledge graph with product information.
Provide personalized recommendations based on the user's question and the structured product information provided.
Be concise, friendly, and mention specific product names and prices.
If no products match the criteria, explain why and suggest alternatives.
        """
        
        user_prompt = f"""User question: {question}

Structured product information from knowledge graph:
{context}

Based on this information, provide a helpful recommendation.
        """
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=800,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ]
        )
        
        return response.content[0].text

ステップ5：実際に動かしてみる

エンドツーエンドのテスト

ここまでの実装をまとめて、実際に動かしてみましょう。環境変数に OpenAI API キーを設定してから実行してください。

import os
from dotenv import load_dotenv

load_dotenv()

if __name__ == "__main__":
    # 初期化
    api_key = os.getenv("OPENAI_API_KEY")
    kg = build_sample_ecommerce_graph()
    rag_system = KnowledgeGraphRAGSystem(api_key, kg)
    
    # テストクエリ
    test_questions = [
        "黒いスニーカーで、返品可能で、5000円以下のおすすめはありますか？",
        "Nikeのスニーカーでおすすめはありますか？",
        "在庫がたくさんある黒い靴を探しています。"
    ]
    
    for question in test_questions:
        print("\n" + "="*60)
        result = rag_system.answer_question(question)
        print(f"\n[FINAL ANSWER]\n{result['answer']}")
        print(f"\n[TRANSPARENCY]")
        print(f"Sources: {result['transparency']['sources']}")
        print(f"Reasoning: {result['transparency']['reasoning']}")

このコードを実行すると、以下のような流れが見えます：

  1. LLMが質問を解析し、「黒」「返品可能」「5000円以下」といった条件を抽出
  2. グラフから該当商品を検索
  3. 検索結果を構造化されたコンテキストに変換
  4. LLMが最終的な推奨を生成

RAGとの重要な違いは、「なぜこの商品が推奨されたか」が透明になることです。グラフの経路（path）を見ることで、推論プロセスをトレースできるんですね。

本番環境への拡張 ── グラフデータベースの活用

Neo4jへの移行

メモリ内のNetworkXでプロトタイピングするのは便利ですが、実際のプロダクション環境ではグラフデータベースが必要です。最も一般的な選択肢がNeo4jですね。

調べてみると、Neo4j用のPythonドライバは neo4j-driver で、以下のような流れで使います。

from neo4j import GraphDatabase
from neo4j.exceptions import ServiceUnavailable

class Neo4jKnowledgeGraph:
    def __init__(self, uri: str, user: str, password: str):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
    
    def add_entity_to_neo4j(self, entity: Entity) -> None:
        """エンティティをNeo4jに追加"""
        with self.driver.session() as session:
            session.run(
                f"""
                CREATE (n:{entity.type.value} {{
                    id: $id,
                    attributes: $attributes
                }})
                """,
                id=entity.id,
                attributes=entity.attributes
            )
    
    def add_relation_to_neo4j(self, source_id: str, target_id: str,
                            relation_type: RelationType) -> None:
        """関係性をNeo4jに追加"""
        with self.driver.session() as session:
            session.run(
                f"""
                MATCH (a {{id: $source_id}})
                MATCH (b {{id: $target_id}})
                CREATE (a)-[:{relation_type.value}]->(b)
                """,
                source_id=source_id,
                target_id=target_id
            )
    
    def query_related_products(self, entity_id: str, 
                              max_price: int = None) -> List[Dict[str, Any]]:
        """Cypherクエリで関連商品を検索"""
        with self.driver.session() as session:
            query = f"""
            MATCH (p:product)-[:belongs_to|:has_brand]->(category)
            WHERE category.id = $entity_id
            AND (p.attributes.price <= $max_price OR $max_price IS NULL)
            AND p.attributes.returnable = true
            RETURN p, p.attributes as attrs
            ORDER BY p.attributes.price ASC
            LIMIT 10
            """
            result = session.run(query, entity_id=entity_id, max_price=max_price)
            return [dict(record) for record in result]

Neo4jへの移行は、本番環境での大規模データセット（数百万のノード/エッジ）を扱う場合に必須になります。また、インデックス設定やクエリ最適化も重要になってくるので、本番前に十分なパフォーマンステストを実施しましょう。

アンチパターン：避けるべき実装

ここで実装の注意点を整理しておきます。自分も最初この辺りでハマったので。

アンチパターン 問題点 改善方法
LLMにグラフ全体をプロンプトに含める トークン数が爆増し、コスト増加。遅延も悪化 グラフから関連サブグラフのみを抽出して渡す
フィルタリングを全てLLMに任せる 幻覚が起きて、存在しない情報を生成することがある グラフレベルでフィルタリングしてから、LLMに検証させる
グラフスキーマが不安定 新しいエンティティタイプを追加するたびにクエリ生成ロジックが壊れる スキーマを明確に定義し、バージョニングする
グラフとLLMの出力を無条件に信頼 不正確な推奨が提示される可能性がある キャリブレーション：テストセットで精度を事前計測する

パフォーマンスと精度の最適化

グラフクエリの効率化

大規模なグラフでは、単純な深さ優先探索では遅くなります。以下の工夫が実務では重要です。

  • インデックス：頻繁に検索されるエンティティ属性（色、価格帯など）にインデックスを張る
  • キャッシング：よくアクセスされるサブグラフはキャッシュに保持する
  • クエリ分割：複雑なクエリを単純な複数クエリに分割して実行時間を短縮
  • プリフェッチング：ユーザーが質問する前に、予測される関連ノードを先読みする

LLMの精度向上

グラフクエリ生成の精度は、プロンプト設計で大きく左右されます。

  • Few-shot learning：「質問 → 期待されるクエリ」の例を複数プロンプトに含める
  • チェーンオブソート：LLMに「なぜそのクエリなのか」を段階的に考えさせる
  • 再ランキング：複数の候補クエリを生成して、最も関連性の高いものを選ぶ

自分が実装したプロジェクトでは、few-shot learningだけで精度が15%ほど向上しました。見落としやすいテクニックですが、結構効果的です。

まとめ

  • RAGの限界：単なる文書検索では、複雑な関係性や多段階推論に対応できない
  • ナレッジグラフの役割：エンティティと関係性を構造化することで、より正確で透明な検索が実現される
  • 実装の基本：NetworkXを使ったメモリ内グラフで始めて、Neo4jへ段階的に移行できる
  • LLMとの連携：LLMがグラフクエリを生成し、その結果でコンテキストを構築して、最後に応答を生成する3段階の流れが効果的
  • 本番環境への拡張：Neo4jなどのグラフデータベースを活用することで、大規模データセットにスケールできる
  • パフォーマンス最適化：インデックス設計とクエリの効率化が実務では必須
  • 精度向上：Few-shot learningやチェーンオブソートで、LLMの判断精度を高める

よくある質問（FAQ）

ナレッジグラフを手作業で構築するのは大変では？

その通りです。実務では、既存のデータベースやドキュメントから自動的にナレッジグラフを生成するアプローチが一般的ですね。LLMを使ってドキュメント群からエンティティと関係性を抽出し、グラフに追加していく方法があります。公式ドキュメントによると、このプロセス自体もLLMパイプラインとして実装できます。

Neo4jの代わりに他のグラフデータベースは選択肢になる？

Amazon Neptune、ArangoDB、TigerGraphなども候補になります。ただし、Pythonエコシステムの成熟度と、LLMとの連携例の豊富さから、現状ではNeo4jが最も実装しやすいと感じます。プロジェクトの要件（価格、スケーラビリティ、学習曲線）に応じて検討するといいでしょう。

グラフの深さ制限をどうやって決める？

これは実装してから調査すべきです。テストデータで「深さ1」「深さ2」「深さ3」とそれぞれ試して、精度とレイテンシーのバランスを見ます。自分の経験では、ほとんどの実用的なクエリは深さ2以内で解決できました。むしろ深すぎるグラフ検索は幻覚を増やしてしまうので注意が必要です。

リアルタイムでグラフを更新する場合の注意点は？

キャッシュの管理と整合性が課題になります。グラフのノードを更新した直後に古いキャッシュが使われると、矛盾した情報がLLMに渡されます。キャッシュに有効期限を設定する、または更新時に明示的にキャッシュを無効化する仕組みが必要です。本番環境ではRedisなどを使った分散キャッシュを検討するといいでしょう。

このアーキテクチャのコスト感は？

大きく分けて2つのコスト要因があります。一つはLLM API呼び出し（クエリ生成 + 回答生成で1質問あたり数十円程度）、もう一つはグラフデータベースの運用です。Neo4jのクラウド版は月額数千円からですが、オンプレミスで構築すれば初期投資で回収できる場合もあります。実装して、実際のトラフィックで計測してから決めるのが無難です。

Python
AI開発 Python
よかったらシェアしてね！
  • URLをコピーしました！
  • URLをコピーしました！

関連記事

目次