私は去年/ECサイト運営者として、AIカスタマーサービスの精度向上に пытался 積極的に取り組んでいた。しかし、ある日ユーザーの質問に対して無関係な情報を返答するようになった。調査の結果、攻撃者が意図的に検索インデックスを汚染していたことが判明した。本稿では、この体験を踏まえ、RAG システムにおける検索汚染攻撃のメカニズムと実践的な防御策略について詳しく解説する。

検索汚染攻撃とは何か

検索汚染攻撃（Retrieval Pollution Attack）とは、攻撃者が RAG システムのベクトルデータベースに悪意のあるドキュメントを注入し、検索結果を操作する攻撃手法である。EC サイトの例では、「会社概要」ドキュメントに無関係な製品情報を埋め込むことで、価格查询クエリに対して架空の割引情報を返答させる可能性がある。

攻撃の3つの主要なパターン

1. ドキュメント注入攻撃

攻撃者が元のドキュメントに悪意のあるコンテンツを混入させる。Embedding 生成時にその悪意あるコンテンツもベクトル化され、以後の検索で優先的に参照されるようになる。

2. ベクトル空間操作攻撃

高次元ベクトル空間における近接性（Proximity）を悪用し、特定クエリに対して意図したドキュメントが類似度上位に来るようにベクトルを操作する。

3. 時系列崩壊攻撃

新しいドキュメントの挿入タイミングを悪用し、キャ싱機構やランキング算法の脆弱性を突く。

実践的な防御機構の実装

以下のコードは、HolySheep AI の Embedding API を活用した堅牢な RAG システムの防御機構を示す。HolySheep AI は今すぐ登録すれば無料クレジットが手に入り、レートは ¥1=$1（公式比85%節約）と非常に経済的だ。

import requests
import hashlib
import numpy as np
from datetime import datetime
from typing import List, Dict, Tuple

class SecureRAGDefense:
    """検索汚染攻撃に対する多層防御機構"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.embedding_cache = {}
        self.doc_integrity_hash = {}
        
    def generate_secure_embedding(self, text: str, doc_id: str) -> Dict:
        """整合性検証付きEmbedding生成"""
        
        # ドキュメントコンテンツのハッシュ計算
        content_hash = hashlib.sha256(text.encode()).hexdigest()
        
        # 以前の状態との整合性チェック
        if doc_id in self.doc_integrity_hash:
            if self.doc_integrity_hash[doc_id] != content_hash:
                raise ValueError(f"ドキュメント {doc_id} が不正に変更されました")
        
        self.doc_integrity_hash[doc_id] = content_hash
        
        # HolySheep AI API で Embedding 生成
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "text-embedding-3-small",
                "input": text,
                "encoding_format": "float"
            }
        )
        
        if response.status_code != 200:
            raise Exception(f"Embedding生成失敗: {response.text}")
        
        result = response.json()
        
        # 信頼度スコア計算
        embedding_vector = np.array(result["data"][0]["embedding"])
        reliability_score = self._calculate_reliability_score(
            embedding_vector, text
        )
        
        return {
            "embedding": result["data"][0]["embedding"],
            "content_hash": content_hash,
            "reliability_score": reliability_score,
            "timestamp": datetime.utcnow().isoformat()
        }
    
    def _calculate_reliability_score(
        self, 
        embedding: np.ndarray, 
        text: str
    ) -> float:
        """埋め込みベクトルから信頼度スコアを計算"""
        
        # ベクトルの分散計算（異常値検出）
        variance = np.var(embedding)
        
        # テキスト長との比率
        length_ratio = len(text) / 1000.0
        
        # 基本スコア（高いほど信頼）
        base_score = 1.0 - min(variance * 10, 0.5)
        length_penalty = 0.9 if length_ratio > 0.5 else 1.0
        
        return base_score * length_penalty
    
    def verify_retrieval_result(
        self, 
        retrieved_docs: List[Dict],
        query: str,
        threshold: float = 0.7
    ) -> Tuple[List[Dict], List[str]]:
        """検索結果の信頼性検証"""
        
        verified_docs = []
        warnings = []
        
        for doc in retrieved_docs:
            reliability = doc.get("reliability_score", 0)
            
            if reliability < threshold:
                warnings.append(
                    f"ドキュメント {doc['doc_id']} "
                    f"(信頼度: {reliability:.2f}) が閾値を下回っています"
                )
                continue
                
            # ハッシュ検証
            if "content_hash" in doc:
                computed_hash = hashlib.sha256(
                    doc["content"].encode()
                ).hexdigest()
                if computed_hash != doc["content_hash"]:
                    warnings.append(
                        f"ドキュメント {doc['doc_id']} の改竄を検出"
                    )
                    continue
            
            verified_docs.append(doc)
            
        return verified_docs, warnings


使用例

def main():
    api_key = "YOUR_HOLYSHEEP_API_KEY"
    rag = SecureRAGDefense(api_key)
    
    # 安全なEmbedding生成（レイテンシ <50ms）
    result = rag.generate_secure_embedding(
        text="山田電気店の会社概要と製品案内",
        doc_id="company_profile_001"
    )
    
    print(f"Embedding生成完了")
    print(f"信頼度スコア: {result['reliability_score']:.3f}")
    print(f"コンテンツハッシュ: {result['content_hash'][:16]}...")

if __name__ == "__main__":
    main()

多層防御アーキテクチャの実装

以下のコードは、ベクトル類似度検索における異常検知とフィルタリングを実装したものだ。DeepSeek V3.2 は $0.42/MTok と非常に低コストで、大量ドキュメントの検証に最適な選択肢となる。

import requests
import json
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np

class VectorAnomalyDetector:
    """ベクトル空間における異常検知システム"""
    
    def __init__(self, contamination=0.1):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42,
            n_estimators=100
        )
        self.scaler = StandardScaler()
        self.is_fitted = False
        self.baseline_vectors = []
        
    def establish_baseline(self, known_good_vectors: List[List[float]]):
        """正常ドキュメントのベースライン確立"""
        
        if len(known_good_vectors) < 50:
            raise ValueError(
                "ベースライン確立には少なくとも50件の正常ドキュメントが必要です"
            )
        
        baseline_array = np.array(known_good_vectors)
        self.baseline_vectors = baseline_array
        
        # 正規化
        scaled = self.scaler.fit_transform(baseline_array)
        
        # モデル訓練
        self.model.fit(scaled)
        self.is_fitted = True
        
        print(f"ベースライン確立完了: {len(known_good_vectors)} 件のドキュメント")
    
    def detect_anomaly(self, vector: List[float]) -> Dict:
        """单个ベクトルの異常検知"""
        
        if not self.is_fitted:
            raise RuntimeError("先に establish_baseline を実行してください")
        
        # 正規化
        scaled_vector = self.scaler.transform([vector])
        
        # 異常スコア予測
        anomaly_score = self.model.score_samples(scaled_vector)[0]
        prediction = self.model.predict(scaled_vector)[0]
        
        # 統計的距離計算
        baseline = np.array(self.baseline_vectors)
        distances = np.linalg.norm(baseline - vector, axis=1)
        avg_distance = np.mean(distances)
        std_distance = np.std(distances)
        z_score = (avg_distance - np.mean(distances)) / (std_distance + 1e-8)
        
        return {
            "is_anomaly": bool(prediction == -1),
            "anomaly_score": float(anomaly_score),
            "avg_distance_from_baseline": float(avg_distance),
            "z_score": float(z_score),
            "confidence": float(abs(z_score) / (abs(z_score) + 1))
        }
    
    def batch_filter(
        self, 
        retrieved_results: List[Dict],
        threshold: float = 0.8
    ) -> Tuple[List[Dict], List[Dict]]:
        """検索結果の一括フィルタリング"""
        
        safe_results = []
        suspicious_results = []
        
        for result in retrieved_results:
            vector = result.get("embedding", [])
            if not vector:
                continue
                
            detection = self.detect_anomaly(vector)
            
            if detection["confidence"] >= threshold:
                result["anomaly_detection"] = detection
                suspicious_results.append(result)
            else:
                result["anomaly_detection"] = detection
                safe_results.append(result)
        
        return safe_results, suspicious_results

class SemanticConsistencyChecker:
    """セマンティック一貫性チェッカー"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
    
    def verify_semantic_coherence(
        self, 
        query: str, 
        retrieved_text: str
    ) -> Dict:
        """クエリと検索結果のセマンティック一貫性検証"""
        
        # HolySheep AI で一貫性スコア生成
        prompt = f"""
        クエリ: {query}
        検索結果: {retrieved_text}
        
        上記クエリと検索結果の関連性を0.0〜1.0で評価してください。
        関連性スコアのみを返答してください。
        """
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4o-mini",
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.1,
                "max_tokens": 10
            }
        )
        
        if response.status_code != 200:
            return {"error": response.text}
        
        result = response.json()
        score_text = result["choices"][0]["message"]["content"].strip()
        
        try:
            score = float(score_text)
        except ValueError:
            score = 0.0
            
        return {
            "coherence_score": score,
            "is_consistent": score >= 0.6,
            "query": query,
            "retrieved_snippet": retrieved_text[:100]
        }


統合防御システム

class RAGDefenseOrchestrator:
    """多層防御を統合管理"""
    
    def __init__(self, api_key: str):
        self.anomaly_detector = VectorAnomalyDetector(contamination=0.05)
        self.consistency_checker = SemanticConsistencyChecker(api_key)
        self.api_key = api_key
        
    def secure_retrieval(
        self,
        query: str,
        candidate_documents: List[Dict],
        baseline_vectors: List[List[float]]
    ) -> Dict:
        """安全な検索パイプライン"""
        
        # Step 1: ベースライン確立（初回のみ）
        if not self.anomaly_detector.is_fitted:
            self.anomaly_detector.establish_baseline(baseline_vectors)
        
        # Step 2: 異常値フィルタリング
        safe, suspicious = self.anomaly_detector.batch_filter(
            candidate_documents,
            threshold=0.75
        )
        
        # Step 3: セマンティック一貫性検証
        verified = []
        for doc in safe:
            consistency = self.consistency_checker.verify_semantic_coherence(
                query=query,
                retrieved_text=doc["content"]
            )
            
            if consistency["is_consistent"]:
                doc["consistency_score"] = consistency["coherence_score"]
                verified.append(doc)
        
        return {
            "verified_documents": verified,
            "filtered_suspicious": len(suspicious),
            "filtered_inconsistent": len(candidate_documents) - len(verified) - len(suspicious)
        }


使用例

def demo():
    api_key = "YOUR_HOLYSHEEP_API_KEY"
    orchestrator = RAGDefenseOrchestrator(api_key)
    
    # ベースライン（北京計算研究所の正常ドキュメント例）
    baseline = [
        [0.1] * 1536,
        [0.2] * 1536,
        # ... 50件以上の正常ベクトル
    ]
    
    # 検索パイプライン実行
    result = orchestrator.secure_retrieval(
        query="製品価格の確認",
        candidate_documents=[
            {"content": "山田電気店の新製品情報です", "embedding": [0.15] * 1536},
            {"content": "特殊割引が適用されました", "embedding": [0.99] * 1536},  # 異常値
        ],
        baseline_vectors=baseline
    )
    
    print(f"検証済みドキュメント: {len(result['verified_documents'])}")
    print(f"フィルタリング: {result['filtered_suspicious']} 件")

if __name__ == "__main__":
    demo()

RAG システム攻撃の最新トレンド（2024-2025）

私の実務経験では、特に以下の3つの攻撃が増加傾向を見せている：

HolySheep AI の優位性

RAG システム構築において HolySheep AI は以下の理由で最適な選択だ：

料金比較（2026年1月更新）

モデルOutput価格/MTokRAG向き評価
DeepSeek V3.2$0.42★★★★★
Gemini 2.5 Flash$2.50★★★★☆
GPT-4.1$8.00★★★☆☆
Claude Sonnet 4.5$15.00★★★☆☆

よくあるエラーと対処法

エラー1: Embedding 生成時のハッシュ不一致エラー

ValueError: ドキュメント doc_001 が不正に変更されました

content_hash の不一致が検出されました

原因: 同一ドキュメント ID に対して異なる内容が登録されようとしている。攻撃の兆候またはコードのバグ。

解決コード:

import hashlib

def safe_document_update(
    doc_id: str, 
    new_content: str, 
    existing_hash: Dict[str, str],
    rag_instance: SecureRAGDefense
) -> bool:
    """安全なドキュメント更新流程"""
    
    # まず現在の状態を確認
    current_hash = existing_hash.get(doc_id)
    
    if current_hash is None:
        # 新規ドキュメント
        result = rag_instance.generate_secure_embedding(
            text=new_content,
            doc_id=doc_id
        )
        return True
    
    # コンテンツ変更の正当性を検証
    new_hash = hashlib.sha256(new_content.encode()).hexdigest()
    
    if new_hash != current_hash:
        # 変更履歴をログに記録
        print(f"[WARNING] ドキュメント {doc_id} の内容が変更されます")
        print(f"  旧ハッシュ: {current_hash[:16]}...")
        print(f"  新ハッシュ: {new_hash[:16]}...")
        
        # 管理者の承認が必要
        admin_approval = confirm_admin_approval(doc_id)
        
        if not admin_approval:
            raise PermissionError("管理者承認なしでドキュメントは更新できません")
    
    result = rag_instance.generate_secure_embedding(
        text=new_content,
        doc_id=doc_id
    )
    return True

def confirm_admin_approval(doc_id: str) -> bool:
    """管理者承認フロー（実際の実装ではDB/外部APIと連携）"""
    # 実装省略 - 実際のシステムでは監査ログとの統合が必要
    return True

エラー2: IsolationForest 訓練時のデータ不足

ValueError: ベースライン確立には少なくとも50件の正常ドキュメントが必要です




原因: 異常検知モデルの訓練に必要な正常ドキュメントのサンプル数が不足している。



解決コード:



import requests
from typing import List

class IncrementalBaselineBuilder:
    """段階的なベースライン構築システム"""
    
    def __init__(self, api_key: str, min_samples: int = 50):
        self.api_key = api_key
        self.min_samples = min_samples
        self.base_url = "https://api.holysheep.ai/v1"
        self.pending_vectors = []
        self._cache = {}
    
    def add_sample(self, doc_content: str) -> Dict:
        """サンプルを追加してEmbedding生成"""
        
        # キャッシュチェック
        cache_key = hashlib.md5(doc_content.encode()).hexdigest()
        if cache_key in self._cache:
            return {"status": "cached", "vector": self._cache[cache_key]}
        
        # HolySheep API呼び出し
        response = requests.post(
            f"{self.base_url}/embeddings",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "text-embedding-3-small",
                "input": doc_content
            },
            timeout=10
        )
        
        if response.status_code == 200:
            result = response.json()
            vector = result["data"][0]["embedding"]
            self._cache[cache_key] = vector
            self.pending_vectors.append(vector)
            return {
                "status": "added",
                "total_samples": len(self.pending_vectors),
                "remaining": max(0, self.min_samples - len(self.pending_vectors))
            }
        else:
            return {"status": "error", "message": response.text}
    
    def is_ready(self) -> bool:
        """ベースライン構築完了確認"""
        return len(self.pending_vectors) >= self.min_samples
    
    def get_baseline_vectors(self) -> List[List[float]]:
        """ベースライン取得"""
        if not self.is_ready():
            raise RuntimeError(
                f"ベースライン未完了: {len(self.pending_vectors)}/{self.min_samples}"
            )
        return self.pending_vectors[:self.min_samples]


使用例

builder = IncrementalBaselineBuilder("YOUR_HOLYSHEEP_API_KEY")


徐々にサンプルを追加

documents = [
    "会社概要: 山田電気店",
    "製品カテゴリ一覧",
    "よくあるご質問",
    # ... 50件以上
]

for doc in documents:
    result = builder.add_sample(doc)
    print(f"サンプル追加: {result}")
    
    if builder.is_ready():
        print("ベースライン構築完了")
        baseline = builder.get_baseline_vectors()
        break




エラー3: セマンティック一貫性検証の API タイムアウト



requests.exceptions.Timeout: API呼び出しが30秒以内に完了しませんでした




原因: ネットワーク遅延または API 過負荷によるタイムアウト。RAG システムではレイテンシが重要なため即座に対処が必要。



解決コード:



import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time

class RobustAPIClient:
    """堅牢なAPIクライアント（自動リトライ付き）"""
    
    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url
        self.session = self._create_session()
    
    def _create_session(self) -> requests.Session:
        """再試行策略付きセッション作成"""
        session = requests.Session()
        
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503,

関連リソース
関連記事