標準RAGの限界とGraphRAGによる複雑データ解析の解決策

import os
import logging
from pathlib import Path
from graphrag.api import build_index, query
from graphrag.config import load_config
from graphrag.storage import DocumentStore, GraphStore
import asyncio

# ロギング設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class GraphRAGPipeline:
    """GraphRAGのパイプラインを管理するクラス"""
    
    def __init__(self, input_dir: str, output_dir: str):
        self.input_dir = Path(input_dir)
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.logger = logging.getLogger(self.__class__.__name__)
        
        # 設定の初期化
        self.config = self._load_configuration()
    
    def _load_configuration(self):
        """設定ファイルを読み込む"""
        config_path = self.output_dir / "settings.yaml"
        if config_path.exists():
            self.logger.info(f"設定ファイルを読み込み: {config_path}")
            return load_config(config_path)
        
        self.logger.warning("設定ファイルが存在しません。デフォルト設定を使用します。")
        return None
    
    async def build_index(self, incremental: bool = False):
        """知識グラフを構築する"""
        self.logger.info(f"インデックス構築開始: {self.input_dir}")
        
        try:
            if incremental:
                self.logger.info("増分ビルドモードで実行")
            
            result = await build_index(
                input_dir=str(self.input_dir),
                output_dir=str(self.output_dir),
                config=self.config,
                incremental=incremental
            )
            
            self.logger.info(f"インデックス構築完了: {result.get('entity_count', 0)} エンティティ抽出")
            return result
            
        except FileNotFoundError as e:
            self.logger.error(f"入力ディレクトリが見つかりません: {e}")
            raise
        except ValueError as e:
            self.logger.error(f"設定エラー: {e}")
            raise
        except Exception as e:
            self.logger.error(f"インデックス構築中に予期しないエラー: {e}")
            raise
    
    async def query_local(self, question: str, community_level: int = 2):
        """ローカル検索で特定の事実に関する質問に回答"""
        self.logger.info(f"ローカル検索実行: {question[:50]}...")
        
        try:
            result = await query(
                method="local",
                query=question,
                community_level=community_level,
                response_type="Multiple Paragraphs",
                data_dir=str(self.output_dir)
            )
            
            self.logger.info("ローカル検索完了")
            return result
            
        except RuntimeError as e:
            if "Index not found" in str(e):
                self.logger.error("インデックスが構築されていません。先にbuild_indexを実行してください。")
                raise
            raise
        except Exception as e:
            self.logger.error(f"クエリ実行中にエラー: {e}")
            raise
    
    async def query_global(self, question: str, response_type: str = "Comprehensive"):
        """グローバル検索で集約的な質問に回答"""
        self.logger.info(f"グローバル検索実行: {question[:50]}...")
        
        try:
            result = await query(
                method="global",
                query=question,
                response_type=response_type,
                data_dir=str(self.output_dir)
            )
            
            self.logger.info("グローバル検索完了")
            return result
            
        except RuntimeError as e:
            if "Index not found" in str(e):
                self.logger.error("インデックスが構築されていません。先にbuild_indexを実行してください。")
                raise
            raise
        except Exception as e:
            self.logger.error(f"クエリ実行中にエラー: {e}")
            raise
    
    async def query_drift(self, question: str):
        """DRIFT検索で動的な探索的質問に対応"""
        self.logger.info(f"DRIFT検索実行: {question[:50]}...")
        
        try:
            result = await query(
                method="drift",
                query=question,
                data_dir=str(self.output_dir)
            )
            
            self.logger.info("DRIFT検索完了")
            return result
            
        except Exception as e:
            self.logger.error(f"DRIFTクエリ実行中にエラー: {e}")
            raise


async def main():
    """メイン実行関数"""
    # 環境変数の確認
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        logger.error("OPENAI_API_KEYが設定されていません")
        raise ValueError("API key required")
    
    # パイプライン初期化
    pipeline = GraphRAGPipeline(
        input_dir="./data/documents",
        output_dir="./output/graphrag_index"
    )
    
    try:
        # インデックスの構築
        logger.info("=" * 50)
        logger.info("ステップ1: 知識グラフの構築")
        await pipeline.build_index()
        
        # ローカル検索の例
        logger.info("=" * 50)
        logger.info("ステップ2: ローカル検索の実行")
        local_result = await pipeline.query_local(
            question="プロジェクトXで使用された主要な技術スタックは何ですか？",
            community_level=2
        )
        print("\n【ローカル検索結果】")
        print(local_result)
        
        # グローバル検索の例
        logger.info("=" * 50)
        logger.info("ステップ3: グローバル検索の実行")
        global_result = await pipeline.query_global(
            question="この企業文化を特徴づける価値観と行動特性は何ですか？",
            response_type="Comprehensive"
        )
        print("\n【グローバル検索結果】")
        print(global_result)
        
    except Exception as e:
        logger.error(f"パイプライン実行中にエラー: {e}")
        raise


if __name__ == "__main__":
    asyncio.run(main())

from dataclasses import dataclass
from typing import Optional

@dataclass
class QueryResult:
    """GraphRAGクエリ結果を封装"""
    response: str
    sources: list[dict]
    confidence: float
    method: str
    community_level: Optional[int] = None
    
    def to_markdown(self) -> str:
        """結果をMarkdown形式で出力"""
        output = f"## 回答\n\n{self.response}\n\n"
        output += f"**検索方法**: {self.method}\n"
        output += f"**信憑性スコア**: {self.confidence:.2f}\n\n"
        
        if self.sources:
            output += "### 参照元\n"
            for i, source in enumerate(self.sources, 1):
                output += f"{i}. {source.get('title', 'Unknown')} (信頼度: {source.get('relevance', 0):.2f})\n"
        
        return output


class ResultPostProcessor:
    """クエリ結果の後処理"""
    
    def __init__(self, confidence_threshold: float = 0.7):
        self.threshold = confidence_threshold
    
    def validate_result(self, result: QueryResult) -> tuple[bool, str]:
        """結果の妥当性を検証"""
        if result.confidence < self.threshold:
            return False, f"信憑性スコアが閾値 ({self.threshold}) を下回っています"
        
        if not result.response or len(result.response) < 50:
            return False, "回答が不自然に短いです"
        
        if result.response.count("[UNABLE TO ANSWER]") > 0:
            return False, "LLMが回答不能と判断しています"
        
        return True, "検証通過"
    
    def enrich_with_metadata(self, result: QueryResult) -> dict:
        """結果にメタデータを付与"""
        return {
            "response": result.response,
            "confidence": result.confidence,
            "validation_passed": self.validate_result(result)[0],
            "source_count": len(result.sources),
            "method": result.method,
            "tokens_estimate": len(result.response) // 4  # 概算トークン数
        }