AIコーディング 2026.07.29

ChatGPT APIで正しいデータが返ってこない理由|ハルシネーション対策と出力検証方法【2026年版】

タグ:ChatGPT API / LLMハルシネーション / データベース連携 / 出力検証 / プロンプト設計

ChatGPT APIのハルシネーション問題とは

ChatGPT APIを使ってデータベースクエリを実行する際、開発者が直面する落とし穴があります。クエリの実行は成功し、APIからのレスポンスコードは200番台ですが、返ってきたデータが実際のクエリ結果と異なるという現象です。これはLLMハルシネーション(LLMが根拠なく事実と異なる情報を生成すること)が原因の一つです。

データリーケージに関する研究によると、LLMが誤った情報を生成する理由は3つの異なるパターンに分類されます。これらを理解し対策することで、開発時のデバッグ時間と本番環境でのバグ発生を大幅に削減できます。

データリーケージの3つのパターン

パターン1: 学習データからの推測による誤り

LLMは学習段階で見たデータパターンに基づいて、「通常こういう結果が返ってくるだろう」と推測してしまうことがあります。特に以下のような場合に起きやすいです:

  • ユーザーIDと注文額の関係性が、実際のデータと異なる推測をする
  • よく見かけるデータ形式を「これは~という構造だろう」と決めつける
  • 統計的に多数派のパターンを優先的に出力する

例えば、同じアカウントの過去の取引履歴が10件あっても、LLMが「あ、このユーザーは月5件購入するはず」と学習データから推測して、実際の10件ではなく5件の結果を返す場合があります。

パターン2: プロンプトの文脈からの誤推測

プロンプト内に曖昧さや矛盾があると、LLMがその隙をついて自分の推測を入れてしまいます。

# 悪い例:曖昧なプロンプト
response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {
            "role": "user",
            "content": "ユーザーテーブルから年齢が高い順に取得して"
        }
    ]
)

このプロンプトでは「年齢が高い順」が相対的で、何件取得するのか、どのユーザーから始めるのかが不明確です。LLMは文脈から「きっと年収が高い順だろう」「有名ユーザーから始まるはず」と推測を加えてしまいます。

パターン3: スキーマ構造の理解不足による歪み

データベーススキーマが複雑な場合、LLMがカラム名や関連テーブルの意味を間違って解釈することがあります。

-- スキーマが複雑だと誤解しやすい例
SELECT 
    u.user_id,
    u.name,
    o.order_count,
    p.price_sum
FROM users u
LEFT JOIN order_stats o ON u.user_id = o.user_id
LEFT JOIN payment_summary p ON u.user_id = p.user_id

LLMが「LEFT JOINは外部結合か内部結合か」「NULLはどう扱うべきか」を誤解すると、結果セットの行数や値が間違ったものになります。

なぜこれが起きるのか:LLMの仕組みから理解する

LLMは統計的な確率に基づいて次の単語を予測するため、以下のような性質を持ちます:

  1. 訓練データの分布を反映する:よく見かけたパターンを優先する傾向
  2. 完全な入力がないと推測を加える:不完全な指示には「これぐらい加えておこう」と補完する
  3. 確率が高い結果を返す:実際のクエリ結果ではなく「統計的に尤もらしい」結果を返す可能性

例えば、「ユーザーAの最新の注文金額は?」というクエリがあっても、LLMが学習データから「ユーザーAは通常5000円ぐらい購入している」と知っていると、データベースの実際の値(例:2800円)ではなく、推測値(5000円)を返すことがあります。

ハルシネーションを検出する実装方法

方法1:結果検証ロジックの実装

import json
from openai import OpenAI

client = OpenAI(api_key="sk-...")

def query_database_with_validation(
    query_description: str,
    schema_definition: str,
    expected_fields: list[str]
) -> dict:
    """
    LLMにデータベースクエリを実行させ、
    結果を検証するラッパー関数
    """
    
    # ステップ1: スキーマを明確に認識させる
    system_prompt = f"""
あなたは正確なデータベースクエリ実行アシスタントです。
以下のスキーマに従ってSQL文を生成し、その結果を返してください。
絶対に推測や補完をしてはいけません。実際のクエリ結果のみを返す必要があります。

【スキーマ定義】
{schema_definition}

【ルール】
- 返すデータはJSON形式で、以下の構造にしてください:
  {{"success": bool, "result": list, "query": str}}
- 必ず実行したSQLクエリを含めてください
- 結果が空の場合でも、説得力のある推測を加えてはいけません
"""
    
    # ステップ2: LLMにクエリを実行させる
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": system_prompt
            },
            {
                "role": "user",
                "content": f"次のデータを取得してください:{query_description}"
            }
        ],
        temperature=0  # 決定論的な出力を強制
    )
    
    # ステップ3: 応答をJSON解析
    try:
        result_text = response.choices[0].message.content
        result_json = json.loads(result_text)
    except json.JSONDecodeError:
        return {
            "error": "LLMの応答がJSON形式ではありません",
            "raw_response": result_text
        }
    
    # ステップ4: 検証
    validation_errors = []
    
    # チェック1:必須フィールドの存在確認
    if not all(field in result_json for field in ["success", "result", "query"]):
        validation_errors.append("必須フィールドが不足しています")
    
    # チェック2:返されたカラムが期待値と一致するか
    if result_json.get("success") and result_json.get("result"):
        first_row = result_json["result"][0]
        actual_fields = set(first_row.keys())
        expected_fields_set = set(expected_fields)
        
        if actual_fields != expected_fields_set:
            validation_errors.append(
                f"カラムのずれ:期待値 {expected_fields_set}、"
                f"実際 {actual_fields}"
            )
    
    # チェック3:SQLクエリが正当なSELECT文か
    sql_query = result_json.get("query", "").strip().upper()
    if not sql_query.startswith("SELECT"):
        validation_errors.append("SQLクエリがSELECT文ではありません")
    
    return {
        "success": len(validation_errors) == 0,
        "validation_errors": validation_errors,
        "result": result_json
    }

方法2:二重検証による確認

def verify_query_result(
    query: str,
    initial_result: dict,
    database_connection
) -> bool:
    """
    LLMが返した結果が、実際のDBの結果と一致するか確認
    """
    # ステップ1:LLMが返したSQLを実際に実行
    try:
        cursor = database_connection.cursor()
        cursor.execute(initial_result["query"])
        actual_db_result = cursor.fetchall()
    except Exception as e:
        return False, f"DB実行エラー:{str(e)}"
    
    # ステップ2:LLMの結果と比較
    llm_result = initial_result.get("result", [])
    
    # 行数チェック
    if len(llm_result) != len(actual_db_result):
        return False, (
            f"行数の不一致:LLM={len(llm_result)}, "
            f"実DB={len(actual_db_result)}"
        )
    
    # 値の一致チェック
    for llm_row, db_row in zip(llm_result, actual_db_result):
        for key, llm_value in llm_row.items():
            db_value = db_row.get(key)
            if llm_value != db_value:
                return False, (
                    f"値の不一致:カラム'{key}', "
                    f"LLM={llm_value}, DB={db_value}"
                )
    
    return True, "検証成功"

方法3:結果の統計的検証

def statistically_validate_result(
    llm_result: list[dict],
    expected_value_ranges: dict
) -> tuple[bool, list[str]]:
    """
    LLMの結果が統計的に妥当か確認
    
    例:
    expected_value_ranges = {
        "age": (18, 100),  # 年齢は18~100の範囲
        "price": (100, 100000),  # 価格は100~10万円
    }
    """
    errors = []
    
    for row in llm_result:
        for field, (min_val, max_val) in expected_value_ranges.items():
            if field in row:
                value = row[field]
                if not (min_val <= value <= max_val):
                    errors.append(
                        f"範囲外:{field}={value} "
                        f"(期待値:{min_val}{max_val})"
                    )
    
    return len(errors) == 0, errors

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

プロンプト設計の最適化

# 脆弱なプロンプト例
bad_prompt = """
ユーザーテーブルから活動的なユーザーを取得して
"""

# 堅牢なプロンプト例
good_prompt = """
users テーブルから以下の条件で抽出してください:
- カラム:user_id, name, email, created_at
- 条件:last_login_at が過去30日以内のレコード
- 並び順:created_at の昇順
- 件数上限:100件

以下の形式で結果を返してください:
{
  "success": true,
  "query": "実行したSELECT文",
  "result": [
    {"user_id": 1, "name": "...", ...},
    ...
  ],
  "row_count": 50
}

重要:推測は一切加えず、実際のクエリ結果のみ返してください。
"""

Temperature パラメータの調整

# ハルシネーション削減のための設定
response = client.chat.completions.create(
    model="gpt-4",
    messages=messages,
    temperature=0,  # 0は決定論的(最も出現確率の高い選択肢のみ)
    top_p=0.9,  # 上位90%の確率質量を考慮
    max_tokens=2000,  # 冗長な追加情報を制限
)

スキーマドキュメントの明示

schema_doc = """
【テーブル構造】

orders テーブル:
- order_id (整数, 主キー)
- user_id (整数, users.user_id への外部キー)
- order_date (日付)
- total_amount (数値, 小数点以下2桁)
- status (文字列, 値の選択肢:pending, completed, cancelled)

users テーブル:
- user_id (整数, 主キー)
- name (文字列)
- email (文字列)
- created_at (タイムスタンプ)
- last_login_at (タイムスタンプ, NULLの場合あり)

【制約】
- LEFT JOIN の結果、users に対応する orders がない場合、
  orders のカラムはすべて NULL になります
- status = 'cancelled' のレコードは集計から除外してください
"""

よくある失敗パターンと対策

パターンA:「プロンプトに例を示した」だけで安心する

# 危険:例示だけでは不十分
messages = [
    {
        "role": "user",
        "content": """
例:
ユーザーAの注文合計:10000円
ユーザーBの注文合計:5000円

このように、全ユーザーの注文合計を取得してください
        """
    }
]

# 改善:例示 + 具体的な仕様
messages = [
    {
        "role": "user",
        "content": """
実行するSQLの仕様:
- SELECT user_id, SUM(total_amount) as total FROM orders WHERE status = 'completed' GROUP BY user_id
- user_id の昇順
- SUM が 0 以上のみを返す

【誤った例(これを返してはいけない)】
ユーザーAの注文合計:12000円(推測)

【正しい結果の形式】
{"user_id": 1, "total": 10000}
{"user_id": 2, "total": 5000}
        """
    }
]

パターンB:エラーハンドリングなしでLLM結果を信頼する

# 危険:常にエラーが 0 と仮定
def unsafe_query(query_desc: str):
    result = llm_query(query_desc)
    return result["data"]  # ← もしこのキーがなかったら?

# 安全:すべてのケースを想定
def safe_query(query_desc: str):
    try:
        result = llm_query(query_desc)
        
        # 1. レスポンス形式の確認
        if "data" not in result:
            raise ValueError("返された結果に 'data' キーがありません")
        
        # 2. データの整合性確認
        if not isinstance(result["data"], list):
            raise TypeError("'data' がリスト形式ではありません")
        
        # 3. 空結果の扱い
        if len(result["data"]) == 0:
            logging.warning(f"クエリが空結果:{query_desc}")
            return []
        
        return result["data"]
    
    except Exception as e:
        logging.error(f"LLMクエリ失敗:{str(e)}")
        return None

検証コードの実装例(完全版)

from openai import OpenAI
import json
import logging

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

class DatabaseQueryAgent:
    def __init__(self, api_key: str, schema_definition: str):
        self.client = OpenAI(api_key=api_key)
        self.schema_definition = schema_definition
    
    def execute_query(
        self,
        query_description: str,
        expected_fields: list[str]
    ) -> dict:
        """
        検証機能付きでLLMにクエリ実行させる
        """
        
        system_prompt = f"""
あなたはデータベースクエリの実行アシスタントです。
以下のルールを厳守してください:

【スキーマ】
{self.schema_definition}

【出力形式】
{{
  "success": true/false,
  "query": "実行したSQL",
  "result": [...],
  "note": "実行結果の説明"
}}

【禁止事項】
- 推測・補完の一切
- 学習データからの推測値の代入
- NULLの置き換え
- 存在しないレコードの作成
        """
        
        # LLMの実行
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query_description}
            ],
            temperature=0
        )
        
        # 応答の解析と検証
        raw_response = response.choices[0].message.content
        
        try:
            result_json = json.loads(raw_response)
        except json.JSONDecodeError:
            return {
                "error": f"JSON解析失敗:{raw_response}",
                "valid": False
            }
        
        # 検証実行
        validation_result = self._validate_result(
            result_json,
            expected_fields
        )
        
        return {
            "valid": validation_result["valid"],
            "errors": validation_result["errors"],
            "data": result_json,
            "sql": result_json.get("query")
        }
    
    def _validate_result(self, result_json: dict, expected_fields: list[str]) -> dict:
        """
        LLMの出力を複数の観点から検証
        """
        errors = []
        
        # チェック1:必須キー
        required_keys = {"success", "query", "result"}
        if not required_keys.issubset(result_json.keys()):
            errors.append(f"必須キーが不足:{required_keys - set(result_json.keys())}")
        
        # チェック2:SQLクエリの形式
        query = result_json.get("query", "").strip()
        if not query.upper().startswith("SELECT"):
            errors.append(f"クエリがSELECT文ではありません:{query[:50]}")
        
        # チェック3:結果のスキーマ
        result_list = result_json.get("result", [])
        if result_list and isinstance(result_list, list):
            first_row = result_list[0]
            if isinstance(first_row, dict):
                actual_fields = set(first_row.keys())
                expected_set = set(expected_fields)
                if actual_fields != expected_set:
                    errors.append(
                        f"カラム不一致:期待 {expected_set}, "
                        f"実際 {actual_fields}"
                    )
        
        # チェック4:成功フラグの妥当性
        success_flag = result_json.get("success")
        has_results = len(result_list) > 0
        if success_flag and not has_results:
            errors.append("成功フラグが True ですが結果が空です")
        
        return {
            "valid": len(errors) == 0,
            "errors": errors
        }

# 使用例
agent = DatabaseQueryAgent(
    api_key="sk-...",
    schema_definition="""
users テーブル:
  user_id (int, PK), name (string), email (string), age (int)

orders テーブル:
  order_id (int, PK), user_id (int, FK), order_date (date), amount (decimal)
    """
)

result = agent.execute_query(
    query_description="age が 30 以上のユーザーの注文総額を取得",
    expected_fields=["user_id", "name", "total_amount"]
)

if result["valid"]:
    print("✓ 検証成功")
    print(json.dumps(result["data"], indent=2, ensure_ascii=False))
else:
    print("✗ 検証失敗")
    for error in result["errors"]:
        print(f"  - {error}")

本番環境での運用方針

監視ログの設定

def log_query_execution(
    query_description: str,
    llm_result: dict,
    validation_result: dict,
    execution_time_ms: float
):
    """
    本番運用時の監視ログ
    """
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "query": query_description,
        "validation_passed": validation_result["valid"],
        "errors": validation_result.get("errors", []),
        "result_row_count": len(llm_result.get("result", [])),
        "execution_time_ms": execution_time_ms,
        "sql_executed": llm_result.get("query")
    }
    
    # エラー時のアラート
    if not validation_result["valid"]:
        logger.error(f"LLMクエリ検証失敗:{json.dumps(log_entry)}")
    else:
        logger.info(f"LLMクエリ成功:{json.dumps(log_entry)}")

参考:Claude Tool Useで構造化されたSQL生成

自由テキストではなくTool Use(構造化出力)でSQLクエリを生成させ、ハルシネーションの余地を減らす実装例です。

import anthropic
import sqlite3

client = anthropic.Anthropic()

SQL_TOOL = {
    "name": "execute_query",
    "description": "データベースに対してSELECTクエリを実行する",
    "input_schema": {
        "type": "object",
        "properties": {
            "table": {"type": "string"},
            "columns": {"type": "array", "items": {"type": "string"}},
            "where_clause": {"type": "string"},
            "limit": {"type": "integer"}
        },
        "required": ["table", "columns"]
    }
}

def generate_and_validate_query(user_request: str, db_connection) -> dict:
    """構造化出力でクエリ意図を取得し、実DBで検証してから返す"""
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        tools=[SQL_TOOL],
        tool_choice={"type": "tool", "name": "execute_query"},
        messages=[{"role": "user", "content": user_request}]
    )

    tool_use = next(b for b in response.content if b.type == "tool_use")
    params = tool_use.input

    # LLMの「推測」ではなく、構造化パラメータから実際にSQLを組み立てて実行
    columns_str = ", ".join(params["columns"])
    query = f"SELECT {columns_str} FROM {params['table']}"
    if params.get("where_clause"):
        query += f" WHERE {params['where_clause']}"
    if params.get("limit"):
        query += f" LIMIT {params['limit']}"

    cursor = db_connection.cursor()
    cursor.execute(query)
    real_results = cursor.fetchall()

    return {"generated_query": query, "actual_results": real_results}

# 使用例
conn = sqlite3.connect("example.db")
result = generate_and_validate_query("ユーザーテーブルから年齢が高い順に上位10件取得して", conn)
print(f"実行クエリ: {result['generated_query']}")
print(f"実際の結果: {result['actual_results']}")

LLMには「何を取得したいか」という構造化パラメータだけを生成させ、実際のSQL実行と結果取得はコード側で確実に行うことで、LLMが結果を「捏造」する余地を完全に排除できます。


まとめ:ハルシネーション対策の優先順位

  1. 最優先:プロンプトを具体的・明確にする(推測の余地をなくす)
  2. 高優先:temperature = 0 で決定論的にする
  3. 中優先:出力形式を固定(JSON化)して検証する
  4. 推奨:実際のDB結果と二重比較する
  5. 運用:失敗パターンをログして継続改善する

これらの対策を実装することで、ChatGPT API経由のデータベースクエリの信頼性を大幅に向上させることができます。


あわせて読みたい

参考ソース