ChatGPT APIでRAG実装「SQLに直結しない」設計——LangChain中間処理レイヤーのベストプラクティス【2026年版】

なぜ「LLMが直接SQLを書く」は本番環境で使えないか

ChatGPT APIを使ったRetrieval Augmented Generation（RAG）の実装では、LLMにデータベースクエリを生成させるアーキテクチャが一見簡潔に見えます。しかし、金融機関やコンプライアンスが必要な企業では、LLMが生成したSQLを直接実行することは避けるべきです。

この問題は、単なるセキュリティの懸念にとどまりません。AIメモリはデータベースの問題ではなく、アーキテクチャの問題なのです。つまり、LLMの出力をどのように信頼できるデータに変換するか、という層別設計が重要になります。

「LLMが直接SQL」の危険性

• SQLインジェクションリスク: LLMが生成したクエリが、想定外のデータベース操作を実行する可能性
• データ保護違反: アクセス制御を無視して機密情報にアクセスする可能性
• 監査証跡の欠落: どのようなクエリが実行されたかを追跡できない

金融機関では「AIが生成したSQLなど必要ない。統治された回答が必要」という指摘が重要です。つまり、LLMの出力は「参考情報」であり、最終的な実行権限は人間またはルールエンジンが握るべきです。

推奨アーキテクチャ：中間処理レイヤーとは

RAGの正しい実装では、ChatGPT APIの出力とデータベースの間に中間処理レイヤーを挟みます。

ユーザー質問
    ↓
[LangChain]
    ↓
ChatGPT API（意図解釈・パラメータ抽出）
    ↓
【中間処理レイヤー】← ここが重要
  ├─ クエリの正当性検証
  ├─ アクセス権チェック
  ├─ パラメータサニタイズ
  └─ ホワイトリストマッチング
    ↓
データベース（事前定義クエリ実行）
    ↓
結果セット
    ↓
ChatGPT API（回答文生成）
    ↓
ユーザー

この設計では、LLMは「クエリ文字列そのもの」ではなく、「クエリの意図」と「パラメータ」だけを生成します。実際のSQLは、ホワイトリストに登録された安全なクエリテンプレートのみが使われます。

LangChainで中間処理レイヤーを実装する

前提環境

• Python 3.9以上
• LangChain（最新版）
• OpenAI Python SDK
• SQLAlchemy（任意だが推奨）
• dotenv（APIキー管理）

ステップ1：安全なクエリテンプレートを事前定義

LLMに生成させるのではなく、ドメイン知識をコード化したテンプレートを用意します。

# queries.py
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class QueryTemplate:
    """実行を許可するクエリテンプレート"""
    name: str
    description: str
    template: str
    required_params: List[str]
    param_types: Dict[str, str]  # "customer_id": "int", "month": "str"

# 許可されたクエリのみをホワイトリストに登録
ALLOWED_QUERIES = {
    "customer_order_history": QueryTemplate(
        name="customer_order_history",
        description="顧客の注文履歴を月指定で取得",
        template="""
            SELECT order_id, order_date, total_amount
            FROM orders
            WHERE customer_id = %(customer_id)s
              AND MONTH(order_date) = %(month)s
            ORDER BY order_date DESC
        """,
        required_params=["customer_id", "month"],
        param_types={"customer_id": "int", "month": "int"}
    ),
    "product_stock_check": QueryTemplate(
        name="product_stock_check",
        description="商品の在庫状況を取得",
        template="""
            SELECT product_id, product_name, stock_quantity
            FROM inventory
            WHERE product_id = %(product_id)s
        """,
        required_params=["product_id"],
        param_types={"product_id": "int"}
    ),
}

def get_allowed_query(query_name: str) -> QueryTemplate:
    """ホワイトリストからクエリを取得"""
    if query_name not in ALLOWED_QUERIES:
        raise ValueError(f"Query '{query_name}' is not allowed")
    return ALLOWED_QUERIES[query_name]

ステップ2：LLMに「クエリ意図」と「パラメータ」だけを生成させる

LangChainのプロンプトテンプレートを使い、LLMの出力を制限します。

# llm_parser.py
import json
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

class QueryIntent(BaseModel):
    """LLMが生成する『意図』と『パラメータ』"""
    query_name: str = Field(
        description="実行するクエリの名前（customer_order_history など）"
    )
    parameters: dict = Field(
        description="クエリに必要なパラメータをJSON形式で"
    )
    reasoning: str = Field(
        description="なぜこのクエリを選んだか（監査用）"
    )

parser = PydanticOutputParser(pydantic_object=QueryIntent)

rag_prompt = ChatPromptTemplate.from_template("""
You are a database query assistant. The user asked a question and you must:
1. Identify which ALLOWED query should be executed
2. Extract required parameters from the user's question
3. Return ONLY valid query names from this list:
   - customer_order_history (requires: customer_id, month)
   - product_stock_check (requires: product_id)

User question: {user_question}

{format_instructions}

Respond in JSON format with query_name, parameters, and reasoning.
""")

rag_prompt = rag_prompt.partial(
    format_instructions=parser.get_format_instructions()
)

def parse_user_intent(user_question: str) -> QueryIntent:
    """ユーザー質問をLLMが解析"""
    llm = ChatOpenAI(
        model="gpt-4o",
        temperature=0.1  # 出力の一貫性を重視
    )
    chain = rag_prompt | llm | parser
    intent = chain.invoke({"user_question": user_question})
    return intent

ステップ3：パラメータ検証とサニタイズ

中間処理レイヤーでパラメータを厳密にチェック。

# validator.py
import re
from typing import Any, Dict
from queries import get_allowed_query, QueryTemplate

class QueryValidator:
    """LLM出力を検証する中間処理"""
    
    @staticmethod
    def validate_parameters(
        query_name: str,
        parameters: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        パラメータを検証・サニタイズ
        不正なパラメータはValueErrorをraiseする
        """
        template = get_allowed_query(query_name)
        
        # 1. 必須パラメータの存在確認
        missing = set(template.required_params) - set(parameters.keys())
        if missing:
            raise ValueError(f"Missing parameters: {missing}")
        
        # 2. 型チェックと変換
        validated = {}
        for param_name in template.required_params:
            value = parameters[param_name]
            expected_type = template.param_types[param_name]
            
            try:
                if expected_type == "int":
                    validated[param_name] = int(value)
                elif expected_type == "str":
                    # 文字列パラメータの場合、長さ制限とエスケープ
                    if len(str(value)) > 100:
                        raise ValueError(f"Parameter too long: {param_name}")
                    validated[param_name] = str(value)
                elif expected_type == "date":
                    # ISO形式の日付のみ許可（例: 2026-07-18）
                    if not re.match(r"^\d{4}-\d{2}-\d{2}$", str(value)):
                        raise ValueError(f"Invalid date format: {value}")
                    validated[param_name] = str(value)
                else:
                    validated[param_name] = value
            except (ValueError, TypeError) as e:
                raise ValueError(
                    f"Invalid type for {param_name}: expected {expected_type}, got {type(value).__name__}"
                ) from e
        
        # 3. 追加の不要なパラメータをチェック（攻撃検知）
        extra = set(parameters.keys()) - set(template.required_params)
        if extra:
            print(f"Warning: Extra parameters ignored: {extra}")
        
        return validated
    
    @staticmethod
    def check_access_control(user_id: str, query_name: str) -> bool:
        """
        ユーザーがこのクエリを実行する権限があるか確認
        （アクセス制御ルールはビジネスロジックに応じて実装）
        """
        # 例: 営業ユーザーは customer_order_history を実行可能
        permissions = {
            "sales": ["customer_order_history"],
            "inventory": ["product_stock_check", "customer_order_history"],
            "admin": ["customer_order_history", "product_stock_check"],
        }
        
        user_role = get_user_role(user_id)  # DB照会や認証システムから取得
        return query_name in permissions.get(user_role, [])

def get_user_role(user_id: str) -> str:
    # 実装例：認証システムから取得
    return "sales"  # ダミー

ステップ4：LangChainのエージェントチェーンを構築

# rag_agent.py
import logging
from langchain.agents import Tool, AgentExecutor, initialize_agent
from langchain.chat_models import ChatOpenAI
from llm_parser import parse_user_intent
from validator import QueryValidator
import sqlite3

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SafeQueryExecutor:
    """中間処理レイヤーを含む安全なクエリ実行"""
    
    def __init__(self, db_path: str = "data.db"):
        self.db_path = db_path
        self.validator = QueryValidator()
    
    def execute_safe_query(
        self,
        user_id: str,
        user_question: str
    ) -> str:
        """
        ユーザー質問 → LLM解析 → 検証 → 実行 → 回答生成
        """
        try:
            # 1. LLMが『意図』と『パラメータ』を抽出
            intent = parse_user_intent(user_question)
            logger.info(f"Parsed intent: query_name={intent.query_name}")
            
            # 2. パラメータ検証
            validated_params = self.validator.validate_parameters(
                intent.query_name,
                intent.parameters
            )
            logger.info(f"Parameters validated: {validated_params}")
            
            # 3. アクセス権チェック
            if not self.validator.check_access_control(user_id, intent.query_name):
                return "エラー: このクエリを実行する権限がありません。"
            
            # 4. ホワイトリストからテンプレートを取得して実行
            from queries import get_allowed_query
            template = get_allowed_query(intent.query_name)
            
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            # SQL インジェクション対策：パラメータ化クエリ
            # （SQLiteではスロープで安全に実行）
            cursor.execute(template.template, validated_params)
            result = cursor.fetchall()
            conn.close()
            
            logger.info(f"Query executed: {intent.query_name}, rows={len(result)}")
            
            # 5. LLMに『結果の解釈』をさせる（クエリ実行ではなく）
            answer = generate_answer(user_question, result)
            
            return answer
        
        except ValueError as e:
            logger.warning(f"Validation error: {e}")
            return f"エラー: {str(e)}"
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            return "システムエラーが発生しました。"

def generate_answer(user_question: str, result: list) -> str:
    """LLMに『結果の解釈と回答生成』をさせる"""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.5)
    
    from langchain.prompts import ChatPromptTemplate
    
    answer_prompt = ChatPromptTemplate.from_template("""
    ユーザーの質問: {user_question}
    
    データベースから取得した結果:
    {db_result}
    
    この結果をふまえて、ユーザーにわかりやすく丁寧に回答してください。
    """)
    
    chain = answer_prompt | llm
    response = chain.invoke({
        "user_question": user_question,
        "db_result": str(result)
    })
    
    return response.content

# 使用例
executor = SafeQueryExecutor()
answer = executor.execute_safe_query(
    user_id="user_123",
    user_question="2026年7月の顧客ID 456の注文履歴を教えて"
)
print(answer)

つまずきやすいポイントと解決策

ポイント1：LLMが「存在しないクエリ名」を生成する

問題: LLMが hallucination（幻覚）を起こし、ホワイトリストに無いクエリ名を生成する。

解決策: プロンプトで「許可されたクエリ名のリスト」を明示し、JSON Schemaで出力を厳密に制限。

# プロンプト内にクエリリストを埋め込む
query_list = ", ".join(ALLOWED_QUERIES.keys())

rag_prompt = ChatPromptTemplate.from_template(f"""
You MUST choose query_name from this list only:
{query_list}

If no query fits the user question, respond with {{
  "query_name": "none",
  "parameters": {{}},
  "reasoning": "User question cannot be answered with available queries"
}}
""")

ポイント2：パラメータの型変換でエラーが出る

問題: LLMが数値を文字列で返す、日付形式が不統一など。

解決策: 型チェック時に柔軟に変換し、失敗時は明確なエラーメッセージを返す。

# validator.py の改良版
def safe_convert(value: Any, target_type: str) -> Any:
    """柔軟な型変換"""
    try:
        if target_type == "int":
            # "123" → 123、"123.0" → 123 に対応
            return int(float(str(value)))
        elif target_type == "date":
            from datetime import datetime
            # 複数の日付形式に対応
            for fmt in ["%Y-%m-%d", "%Y/%m/%d", "%d-%m-%Y"]:
                try:
                    return datetime.strptime(str(value), fmt).strftime("%Y-%m-%d")
                except ValueError:
                    continue
            raise ValueError(f"Unrecognized date format: {value}")
        else:
            return str(value)
    except Exception as e:
        raise ValueError(f"Cannot convert {value} to {target_type}: {e}")

ポイント3：SQLインジェクション対策が甘い

問題: パラメータ化クエリを使っていても、文字列パラメータにSQLコードを埋め込まれる可能性。

解決策: SQLiteやPostgreSQLのパラメータバインディング機能を使い、SQLAlchemyと組み合わせる。

# SQLAlchemyを使った安全な実装例
from sqlalchemy import create_engine, text

engine = create_engine("sqlite:///data.db")

def execute_with_sqlalchemy(query_name: str, params: Dict) -> list:
    """SQLAlchemyのパラメータバインディングを使用"""
    template = get_allowed_query(query_name)
    
    with engine.connect() as conn:
        # text()とパラメータの分離が自動で行われる
        stmt = text(template.template)
        result = conn.execute(stmt, params).fetchall()
    
    return result

ポイント4：監査ログの記録漏れ

問題: クエリ実行の経路（LLMが何を判断したか）を記録していないため、後から問題が発生時に原因追跡ができない。

解決策: 全ステップを構造化ログに記録。

# logging_helper.py
import json
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

def log_query_execution(
    user_id: str,
    user_question: str,
    intent: dict,
    validated_params: dict,
    query_name: str,
    result_count: int,
    status: str = "success"
):
    """監査用ログをJSON形式で記録"""
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "user_id": user_id,
        "user_question": user_question,
        "llm_intent": intent,  # LLMが何を判断したか
        "validated_params": validated_params,  # 検証後のパラメータ
        "query_name": query_name,
        "result_count": result_count,
        "status": status
    }
    
    logger.info(json.dumps(log_entry))
    # さらにAWSCloudWatchやDatadog、Splunkなどに送信することも推奨

応用：複雑なクエリ生成が必要な場合

単純なホワイトリストでは対応できない複雑なクエリ（複数テーブル結合、動的WHERE句など）が必要な場合、クエリビルダーパターンを使用します。

# query_builder.py
from sqlalchemy import select, and_
from sqlalchemy.orm import declarative_base, Session
from sqlalchemy import Column, Integer, String, DateTime

Base = declarative_base()

class Order(Base):
    __tablename__ = "orders"
    order_id = Column(Integer, primary_key=True)
    customer_id = Column(Integer)
    order_date = Column(DateTime)
    total_amount = Column(Integer)

class DynamicQueryBuilder:
    """LLM出力から安全にクエリを組み立てる"""
    
    def build_customer_order_query(
        self,
        customer_id: int,
        month: int = None,
        year: int = None
    ):
        """
        LLMは『パラメータ』だけを返す。
        このメソッドが『クエリ構築』をする。
        """
        query = select(Order)
        conditions = [Order.customer_id == customer_id]
        
        if month and year:
            from sqlalchemy import extract
            conditions.append(extract("month", Order.order_date) == month)
            conditions.append(extract("year", Order.order_date) == year)
        
        return select(Order).where(and_(*conditions))

# このアプローチなら、LLMはパラメータだけを返し、
# 実際のクエリ組み立てはビジネスロジックコード（テスト可能）に任せる

本番環境での運用チェックリスト

• 全クエリテンプレートがホワイトリストに登録されているか
• パラメータ検証が「拒否優先」（許可リストベース）か
• SQLインジェクション対策にパラメータ化クエリを使用しているか
• アクセス制御（RBAC）が実装されているか
• 監査ログがJSON形式で記録されているか
• LLMの出力が100%正確だと仮定していないか
• エラー時のユーザーへの返答がセキュアか（情報漏洩なし）
• 定期的に実行されたクエリを監視する仕組みがあるか
• ChatGPT APIのレート制限やコスト対策は講じているか

まとめ：なぜ「中間処理レイヤー」が必須なのか

AIメモリはデータベースの問題ではなく、アーキテクチャの問題です。LLMが直接SQL を書く設計は、開発速度は速いかもしれませんが、本番環境では運用コスト（問題対応、監査、セキュリティインシデント）が極めて高くなります。

推奨の設計：

1. LLMは「何をしたいか」だけを判断（パラメータ抽出）
2. 中間処理レイヤーが「許可するか・しないか」を判定
3. 実際のクエリはホワイトリストのみ実行
4. 全ステップを監査ログに記録

この層別設計により、AIの柔軟性と組織のガバナンスを両立できます。

あわせて読みたい

• LangChain × ベクトルDB で RAG システムを実装｜5ステップで自社データAIを構築
• AIエージェントが会話を忘れる原因と解決方法【RAG・メモリ設計3つの実装パターン】
• LangChainで複数AIエージェント連携｜CI/CD自動化【実装3ステップ】