SQL×RAGで構造化データも自在に検索

データベースに格納された構造化データと、文書ベースの非構造化データを横断して検索できないという課題は、多くの企業が直面している問題です。本記事では、自然言語でのDBクエリ生成とRAG（Retrieval-Augmented Generation）を融合することで、この課題を解決する手法について、LlamaIndexとLangChainの実装例とともに詳しく解説します。

従来の課題：データサイロ問題

多くの企業では、以下のようなデータサイロ問題を抱えています：

•構造化データ：データベースに格納された売上データ、顧客情報、在庫情報など
•非構造化データ：文書、レポート、メール、チャットログなど

これらのデータを統合的に検索・分析することは技術的に困難で、ユーザーは複数のシステムを使い分ける必要がありました。

SQL×RAG融合アプローチ

アーキテクチャ概要

SQL×RAGアプローチでは、以下の流れで処理を行います：

1. 自然言語クエリの解析：ユーザーの質問を分析し、構造化データと非構造化データのどちらが必要かを判定

2. SQLクエリ生成：構造化データが必要な場合、LLMがSQLクエリを自動生成

3. RAG検索：非構造化データが必要な場合、ベクトル検索でドキュメントを取得

4. 結果統合：両方のデータソースからの結果を統合して最終回答を生成

LlamaIndexによる実装

python

1from llama_index import SQLDatabase, VectorStoreIndex, ServiceContext
2from llama_index.query_engine import SQLJoinQueryEngine
3from llama_index.objects import SQLTableNodeMapping, ObjectIndex
4
5# データベース接続の設定
6sql_database = SQLDatabase.from_uri("postgresql://user:pass@localhost/db")
7
8# ドキュメントインデックスの作成
9documents = SimpleDirectoryReader("./docs").load_data()
10vector_index = VectorStoreIndex.from_documents(documents)
11
12# SQLテーブルのメタデータ設定
13table_node_mapping = SQLTableNodeMapping(sql_database)
14table_schema_objs = [
15    (table_node_mapping.from_objects([sql_database.get_single_table_info("sales")]))
16]
17
18# 統合クエリエンジンの作成
19obj_index = ObjectIndex.from_objects(table_schema_objs)
20query_engine = SQLJoinQueryEngine(
21    sql_database=sql_database,
22    obj_index=obj_index,
23    vector_query_engine=vector_index.as_query_engine()
24)
25
26# クエリ実行例
27response = query_engine.query(
28    "昨月の売上トップ10の製品について、製品レビューも含めて分析してください"
29)

LangChainでのSQLエージェント実装

python

1from langchain.agents import create_sql_agent
2from langchain.agents.agent_toolkits import SQLDatabaseToolkit
3from langchain.sql_database import SQLDatabase
4from langchain.llms import OpenAI
5from langchain.vectorstores import Chroma
6from langchain.embeddings import OpenAIEmbeddings
7
8class HybridSQLRAGAgent:
9    def __init__(self, db_uri: str, docs_path: str):
10        # SQL環境の設定
11        self.db = SQLDatabase.from_uri(db_uri)
12        self.sql_toolkit = SQLDatabaseToolkit(db=self.db, llm=OpenAI())
13        
14        # RAG環境の設定
15        embeddings = OpenAIEmbeddings()
16        self.vectorstore = Chroma.from_documents(
17            documents=self.load_documents(docs_path),
18            embedding=embeddings
19        )
20        
21        # エージェントの作成
22        self.agent = create_sql_agent(
23            llm=OpenAI(),
24            toolkit=self.sql_toolkit,
25            verbose=True,
26            agent_type="openai-functions"
27        )
28    
29    def query(self, question: str):
30        # まずSQL関連かRAG関連かを判定
31        if self._needs_database(question):
32            sql_result = self.agent.run(question)
33            
34        if self._needs_documents(question):
35            docs = self.vectorstore.similarity_search(question, k=5)
36            doc_context = "\n".join([doc.page_content for doc in docs])
37            
38        # 結果を統合して最終回答を生成
39        return self._synthesize_response(sql_result, doc_context, question)

実装のベストプラクティス

1. スキーマ設計とメタデータ管理

データベーススキーマの情報をLLMが理解しやすい形で提供することが重要です：

python

1# テーブル説明の追加
2table_info = {
3    "sales": {
4        "description": "売上データテーブル",
5        "columns": {
6            "product_id": "商品ID（products.idと結合）",
7            "sale_date": "売上日時",
8            "amount": "売上金額",
9            "customer_id": "顧客ID（customers.idと結合）"
10        }
11    }
12}

2. セキュリティ対策

SQLインジェクション攻撃を防ぐため、以下の対策を実装します：

python

1from sqlalchemy import text
2
3class SecureSQLGenerator:
4    def __init__(self, allowed_tables: list):
5        self.allowed_tables = allowed_tables
6        
7    def validate_query(self, sql_query: str) -> bool:
8        # 危険なキーワードのチェック
9        dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]
10        if any(keyword in sql_query.upper() for keyword in dangerous_keywords):
11            return False
12            
13        # 許可されたテーブルのみアクセス可能
14        referenced_tables = self.extract_table_names(sql_query)
15        return all(table in self.allowed_tables for table in referenced_tables)

3. パフォーマンス最適化

大規模なデータベースでは、クエリの最適化が重要です：

python

1class OptimizedSQLRAG:
2    def __init__(self):
3        self.query_cache = {}
4        self.embedding_cache = {}
5        
6    def cached_query(self, sql_query: str):
7        if sql_query in self.query_cache:
8            return self.query_cache[sql_query]
9            
10        result = self.execute_sql(sql_query)
11        self.query_cache[sql_query] = result
12        return result

実用的なユースケース

1. 営業分析レポート

「今四半期の売上トップ5の製品について、顧客からのフィードバックと併せて分析レポートを作成してください」

この質問では：

•SQL：売上データから今四半期のトップ5製品を抽出
•RAG：製品に関する顧客フィードバック文書を検索
•統合：数値データと定性データを組み合わせた分析レポートを生成

2. カスタマーサポート

「製品Aの返品に関する問い合わせが増えていますが、具体的な理由と対処法を教えてください」

この質問では：

•SQL：製品Aの返品データとトレンド分析
•RAG：製品仕様書、FAQ、過去のサポート履歴から関連情報を検索

まとめ

SQL×RAG統合アプローチにより、構造化データと非構造化データを横断した検索・分析が可能になります。LlamaIndexとLangChainを活用することで、比較的簡単に実装できる一方、セキュリティとパフォーマンスには十分な注意が必要です。

このアプローチは、データドリブンな意思決定を支援する強力なツールとなり、企業のDX推進において大きな価値を提供することが期待されます。

"SQL×RAG"で構造化データも自在に検索

Table of Contents

SQL×RAGで構造化データも自在に検索

従来の課題：データサイロ問題

SQL×RAG融合アプローチ

アーキテクチャ概要

LlamaIndexによる実装

LangChainでのSQLエージェント実装

実装のベストプラクティス

1. スキーマ設計とメタデータ管理

2. セキュリティ対策

3. パフォーマンス最適化

実用的なユースケース

1. 営業分析レポート

2. カスタマーサポート

まとめ

タグ