【2026年最新】Vector Databases at Scale：企業レベルでの運用・スケーリング戦略完全ガイド

import hashlib
from typing import List, Dict, Any
import numpy as np

class VectorShardingManager:
    def __init__(self, shard_count: int, replica_count: int = 3):
        self.shard_count = shard_count
        self.replica_count = replica_count
        self.shard_nodes = self._initialize_shard_topology()

    def _initialize_shard_topology(self) -> Dict[int, List[str]]:
        """シャード番号からノードリストへのマッピング"""
        topology = {}
        for shard_id in range(self.shard_count):
            # プライマリ + レプリカノードを配置
            nodes = [
                f"vector-node-{shard_id}-primary",
                *[f"vector-node-{shard_id}-replica-{r}"
                  for r in range(self.replica_count)]
            ]
            topology[shard_id] = nodes
        return topology

    def get_shard_for_vector(self, vector_id: str) -> int:
        """一貫性ハッシュによるシャード決定"""
        hash_value = int(hashlib.md5(vector_id.encode()).hexdigest(), 16)
        return hash_value % self.shard_count

    def route_query(self, query_vector: np.ndarray, k: int = 10) -> List[Dict]:
        """全シャードに並列クエリを送信"""
        results = []

        # 並列クエリ実行（実装は簡略化）
        for shard_id in range(self.shard_count):
            primary_node = self.shard_nodes[shard_id][0]
            shard_results = self._query_shard(primary_node, query_vector, k)
            results.extend(shard_results)

        # 全シャード結果をマージしてtop-k選択
        return self._merge_and_rank(results, k)

    def _query_shard(self, node: str, vector: np.ndarray, k: int) -> List[Dict]:
        """個別シャードへのクエリ（実装は省略）"""
        # 実際はgRPC/HTTP APIでノードに送信
        pass

    def _merge_and_rank(self, results: List[Dict], k: int) -> List[Dict]:
        """分散検索結果のマージ"""
        sorted_results = sorted(results, key=lambda x: x['similarity'], reverse=True)
        return sorted_results[:k]

# 使用例：1000万ベクトルを10シャードに分散
shard_manager = VectorShardingManager(shard_count=10, replica_count=2)

# Kubernetes マルチリージョン構成例
apiVersion: v1
kind: ConfigMap
metadata:
  name: vector-db-topology
data:
  regions.yaml: |
    regions:
      us-west-2:
        primary_shards: [0, 1, 2]
        replica_shards: [6, 7, 8]
        read_preference: local
      eu-west-1:
        primary_shards: [3, 4, 5]
        replica_shards: [0, 1, 2]
        read_preference: local
      ap-northeast-1:
        primary_shards: [6, 7, 8]
        replica_shards: [3, 4, 5]
        read_preference: local

    routing_policy:
      query_timeout: 500ms
      failover_latency: 100ms
      consistency_level: eventual    

import faiss
import numpy as np
from typing import Tuple, List

class HierarchicalVectorIndex:
    def __init__(self, dimension: int, total_vectors: int):
        self.dimension = dimension
        self.total_vectors = total_vectors

        # 2段階インデックス：粗い検索 → 精密検索
        self.coarse_quantizer = faiss.IndexFlatIP(dimension)
        self.fine_index = faiss.IndexIVFFlat(
            self.coarse_quantizer,
            dimension,
            min(int(np.sqrt(total_vectors)), 65536)  # クラスター数
        )

        # PQ（Product Quantization）による圧縮
        self.compressed_index = faiss.IndexIVFPQ(
            self.coarse_quantizer,
            dimension,
            min(int(np.sqrt(total_vectors)), 65536),
            8,  # サブベクトル数
            8   # ビット数
        )

    def train_index(self, training_vectors: np.ndarray) -> None:
        """インデックスの訓練"""
        print(f"Training index with {len(training_vectors)} vectors...")

        # 正規化（内積検索を想定）
        faiss.normalize_L2(training_vectors)

        # インデックス訓練
        self.fine_index.train(training_vectors)
        self.compressed_index.train(training_vectors)

        print("Index training completed")

    def add_vectors(self, vectors: np.ndarray, batch_size: int = 100000) -> None:
        """大量ベクトルのバッチ挿入"""
        faiss.normalize_L2(vectors)

        # メモリ効率を考慮したバッチ処理
        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i+batch_size]
            self.fine_index.add(batch)

            # 圧縮インデックスにも追加（メモリ節約用）
            self.compressed_index.add(batch)

            if (i // batch_size) % 10 == 0:
                print(f"Added {i + len(batch)} vectors")

    def search(self, query_vectors: np.ndarray, k: int,
              use_compression: bool = False) -> Tuple[np.ndarray, np.ndarray]:
        """適応的検索：データサイズに応じてインデックスを選択"""
        faiss.normalize_L2(query_vectors)

        # メモリ使用量に基づく動的選択
        if use_compression or self.total_vectors > 100_000_000:
            # 超大規模：圧縮インデックスを使用
            similarities, indices = self.compressed_index.search(query_vectors, k)
        else:
            # 中規模：高精度インデックスを使用
            similarities, indices = self.fine_index.search(query_vectors, k)

        return similarities, indices

# 使用例：1億ベクトルのインデックス構築
index = HierarchicalVectorIndex(dimension=1536, total_vectors=100_000_000)

# 訓練データ（実際のデータセットの1%程度）
training_data = np.random.random((1_000_000, 1536)).astype('float32')
index.train_index(training_data)

# 本番データの追加
production_data = np.random.random((100_000_000, 1536)).astype('float32')
index.add_vectors(production_data)

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict, Any
import numpy as np

class OptimizedVectorSearchEngine:
    def __init__(self, index, batch_size: int = 100, max_workers: int = 8):
        self.index = index
        self.batch_size = batch_size
        self.max_workers = max_workers
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

        # クエリ統計
        self.query_stats = {
            'total_queries': 0,
            'avg_latency': 0,
            'p95_latency': 0,
            'cache_hit_rate': 0
        }

        # 簡易LRUキャッシュ
        self.query_cache = {}
        self.cache_max_size = 10000

    async def batch_search(self, queries: List[np.ndarray], k: int = 10) -> List[Dict]:
        """バッチクエリの非同期処理"""
        start_time = time.time()

        # クエリをバッチに分割
        batches = [queries[i:i+self.batch_size]
                  for i in range(0, len(queries), self.batch_size)]

        # 並列実行
        tasks = []
        for batch in batches:
            task = asyncio.create_task(self._process_batch(batch, k))
            tasks.append(task)

        batch_results = await asyncio.gather(*tasks)

        # 結果の結合
        all_results = []
        for batch_result in batch_results:
            all_results.extend(batch_result)

        # 統計更新
        elapsed_time = time.time() - start_time
        self._update_stats(len(queries), elapsed_time)

        return all_results

    async def _process_batch(self, batch: List[np.ndarray], k: int) -> List[Dict]:
        """個別バッチの処理"""
        loop = asyncio.get_event_loop()

        # CPUバウンドなタスクを別スレッドで実行
        result = await loop.run_in_executor(
            self.executor,
            self._search_batch_sync,
            batch, k
        )
        return result

    def _search_batch_sync(self, batch: List[np.ndarray], k: int) -> List[Dict]:
        """同期バッチ検索"""
        results = []
        batch_array = np.vstack(batch)

        # バッチ検索実行
        similarities, indices = self.index.search(batch_array, k)

        # 結果をフォーマット
        for i, (sims, idxs) in enumerate(zip(similarities, indices)):
            result = {
                'query_id': i,
                'matches': [
                    {'id': int(idx), 'similarity': float(sim)}
                    for idx, sim in zip(idxs, sims)
                    if idx != -1  # 無効インデックスを除外
                ]
            }
            results.append(result)

        return results

    def _update_stats(self, query_count: int, elapsed_time: float) -> None:
        """パフォーマンス統計の更新"""
        self.query_stats['total_queries'] += query_count
        avg_latency_per_query = elapsed_time / query_count * 1000  # ms

        # 移動平均での統計更新
        if self.query_stats['avg_latency'] == 0:
            self.query_stats['avg_latency'] = avg_latency_per_query
        else:
            self.query_stats['avg_latency'] = (
                self.query_stats['avg_latency'] * 0.95 +
                avg_latency_per_query * 0.05
            )

    def get_performance_report(self) -> Dict[str, Any]:
        """パフォーマンスレポートの生成"""
        return {
            'total_queries_processed': self.query_stats['total_queries'],
            'average_latency_ms': round(self.query_stats['avg_latency'], 2),
            'throughput_qps': round(1000 / self.query_stats['avg_latency'], 2),
            'cache_efficiency': self.query_stats['cache_hit_rate']
        }

# 使用例
search_engine = OptimizedVectorSearchEngine(index, batch_size=50)

# 大量クエリの処理
async def process_large_query_batch():
    queries = [np.random.random(1536).astype('float32') for _ in range(1000)]
    results = await search_engine.batch_search(queries, k=10)
    performance = search_engine.get_performance_report()
    print(f"Processed {len(queries)} queries: {performance}")

asyncio.run(process_large_query_batch())

import psutil
import pickle
import mmap
from pathlib import Path
from typing import Optional, Union
import numpy as np

class AdaptiveVectorStorage:
    def __init__(self, base_path: str, memory_limit_gb: int = 32):
        self.base_path = Path(base_path)
        self.memory_limit_bytes = memory_limit_gb * 1024**3
        self.base_path.mkdir(exist_ok=True)

        # メモリ階層
        self.hot_vectors = {}  # 最頻繁アクセス（メモリ）
        self.warm_vectors = {}  # 中頻度アクセス（メモリマップ）
        self.cold_storage_path = self.base_path / "cold_storage"
        self.cold_storage_path.mkdir(exist_ok=True)

        # アクセス統計
        self.access_counts = {}
        self.access_timestamps = {}

    def store_vector(self, vector_id: str, vector: np.ndarray,
                    access_pattern: str = 'unknown') -> None:
        """アクセスパターンに基づく適応的ストレージ"""
        current_memory = psutil.virtual_memory().used

        if access_pattern == 'hot' and current_memory < self.memory_limit_bytes:
            # 高頻度アクセス：メモリに格納
            self.hot_vectors[vector_id] = vector

        elif access_pattern == 'warm':
            # 中頻度アクセス：メモリマップファイル
            self._store_as_mmap(vector_id, vector)

        else:
            # 低頻度アクセス：圧縮してディスク
            self._store_as_compressed(vector_id, vector)

    def _store_as_mmap(self, vector_id: str, vector: np.ndarray) -> None:
        """メモリマップファイルとしての格納"""
        mmap_path = self.base_path / f"warm_{vector_id}.mmap"

        # バイナリファイルとして保存
        with open(mmap_path, 'wb') as f:
            np.save(f, vector)

        # メモリマップで読み込み
        self.warm_vectors[vector_id] = mmap_path

    def _store_as_compressed(self, vector_id: str, vector: np.ndarray) -> None:
        """圧縮ディスクストレージ"""
        compressed_path = self.cold_storage_path / f"{vector_id}.pkl.gz"

        with open(compressed_path, 'wb') as f:
            pickle.dump(vector, f, protocol=pickle.HIGHEST_PROTOCOL)

    def retrieve_vector(self, vector_id: str) -> Optional[np.ndarray]:
        """適応的ベクトル取得"""
        # アクセス統計更新
        self.access_counts[vector_id] = self.access_counts.get(vector_id, 0) + 1
        self.access_timestamps[vector_id] = time.time()

        # HOTストレージから検索
        if vector_id in self.hot_vectors:
            return self.hot_vectors[vector_id]

        # WARMストレージ（メモリマップ）から検索
        if vector_id in self.warm_vectors:
            mmap_path = self.warm_vectors[vector_id]
            with open(mmap_path, 'rb') as f:
                return np.load(f)

        # COLDストレージから検索
        cold_path = self.cold_storage_path / f"{vector_id}.pkl.gz"
        if cold_path.exists():
            with open(cold_path, 'rb') as f:
                vector = pickle.load(f)

            # アクセス頻度に基づく昇格
            if self.access_counts[vector_id] > 10:
                self._promote_to_warm(vector_id, vector)

            return vector

        return None

    def _promote_to_warm(self, vector_id: str, vector: np.ndarray) -> None:
        """COLDからWARMへの昇格"""
        # WARMストレージに移動
        self._store_as_mmap(vector_id, vector)

        # COLDストレージから削除
        cold_path = self.cold_storage_path / f"{vector_id}.pkl.gz"
        if cold_path.exists():
            cold_path.unlink()

    def optimize_storage(self) -> None:
        """アクセスパターンに基づくストレージ最適化"""
        current_memory = psutil.virtual_memory().used

        if current_memory > self.memory_limit_bytes * 0.9:
            # メモリ使用量が限界近い場合の降格処理
            self._demote_cold_vectors()

    def _demote_cold_vectors(self) -> None:
        """低頻度ベクトルの降格"""
        # アクセス頻度でソート
        sorted_vectors = sorted(
            self.access_counts.items(),
            key=lambda x: (x[1], self.access_timestamps.get(x[0], 0))
        )

        # 下位20%をCOLDストレージに移動
        demote_count = max(1, len(sorted_vectors) // 5)

        for vector_id, _ in sorted_vectors[:demote_count]:
            if vector_id in self.hot_vectors:
                vector = self.hot_vectors[vector_id]
                self._store_as_compressed(vector_id, vector)
                del self.hot_vectors[vector_id]

    def get_storage_stats(self) -> Dict[str, Any]:
        """ストレージ統計の取得"""
        return {
            'hot_vectors_count': len(self.hot_vectors),
            'warm_vectors_count': len(self.warm_vectors),
            'cold_vectors_count': len(list(self.cold_storage_path.glob("*.pkl.gz"))),
            'memory_usage_gb': psutil.virtual_memory().used / 1024**3,
            'disk_usage_gb': sum(f.stat().st_size for f in self.base_path.rglob("*") if f.is_file()) / 1024**3
        }

# 使用例
storage = AdaptiveVectorStorage("/opt/vector_storage", memory_limit_gb=64)

# ベクトルの格納（アクセスパターンを指定）
hot_vector = np.random.random(1536).astype('float32')
storage.store_vector("user_profile_123", hot_vector, 'hot')

warm_vector = np.random.random(1536).astype('float32')
storage.store_vector("document_456", warm_vector, 'warm')

import json
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Any, Optional
import hashlib

class TenantTier(Enum):
    FREE = "free"
    STANDARD = "standard"
    PREMIUM = "premium"
    ENTERPRISE = "enterprise"

@dataclass
class TenantConfig:
    tenant_id: str
    tier: TenantTier
    max_vectors: int
    max_queries_per_minute: int
    dedicated_resources: bool
    encryption_key: str
    compliance_requirements: List[str]

class MultiTenantVectorManager:
    def __init__(self):
        self.tenant_configs = {}
        self.tenant_indexes = {}
        self.query_quotas = {}

        # ティア別デフォルト設定
        self.tier_defaults = {
            TenantTier.FREE: {
                'max_vectors': 10000,
                'max_queries_per_minute': 100,
                'dedicated_resources': False
            },
            TenantTier.STANDARD: {
                'max_vectors': 1000000,
                'max_queries_per_minute': 1000,
                'dedicated_resources': False
            },
            TenantTier.PREMIUM: {
                'max_vectors': 10000000,
                'max_queries_per_minute': 10000,
                'dedicated_resources': True
            },
            TenantTier.ENTERPRISE: {
                'max_vectors': 100000000,
                'max_queries_per_minute': 100000,
                'dedicated_resources': True
            }
        }

    def register_tenant(self, tenant_id: str, tier: TenantTier,
                       custom_config: Dict[str, Any] = None) -> TenantConfig:
        """新規テナントの登録"""
        defaults = self.tier_defaults[tier]

        # カスタム設定のマージ
        if custom_config:
            defaults.update(custom_config)

        # 暗号化キーの生成
        encryption_key = hashlib.sha256(f"{tenant_id}-{tier.value}".encode()).hexdigest()

        config = TenantConfig(
            tenant_id=tenant_id,
            tier=tier,
            max_vectors=defaults['max_vectors'],
            max_queries_per_minute=defaults['max_queries_per_minute'],
            dedicated_resources=defaults['dedicated_resources'],
            encryption_key=encryption_key,
            compliance_requirements=custom_config.get('compliance', [])
        )

        self.tenant_configs[tenant_id] = config
        self._initialize_tenant_resources(config)

        return config

    def _initialize_tenant_resources(self, config: TenantConfig) -> None:
        """テナント専用リソースの初期化"""
        if config.dedicated_resources:
            # 専用インデックスの作成
            self.tenant_indexes[config.tenant_id] = self._create_dedicated_index(config)
        else:
            # 共有インデックス内での名前空間分離
            self.tenant_indexes[config.tenant_id] = {
                'type': 'shared',
                'namespace': f"tenant_{config.tenant_id}"
            }

        # クォータ管理の初期化
        self.query_quotas[config.tenant_id] = {
            'current_minute': 0,
            'query_count': 0,
            'last_reset': time.time()
        }

    def _create_dedicated_index(self, config: TenantConfig) -> Dict[str, Any]:
        """専用インデックスの作成"""
        if 'HIPAA' in config.compliance_requirements:
            # HIPAA準拠のセキュア設定
            index_config = {
                'type': 'dedicated',
                'encryption': 'AES-256',
                'access_logging': True,
                'data_residency': 'US',
                'backup_retention': '7_years'
            }
        elif 'GDPR' in config.compliance_requirements:
            # GDPR準拠の設定
            index_config = {
                'type': 'dedicated',
                'right_to_forget': True,
                'data_residency': 'EU',
                'pseudonymization': True
            }
        else:
            # 標準設定
            index_config = {
                'type': 'dedicated',
                'encryption': 'AES-128',
                'backup_retention': '30_days'
            }

        index_config['tenant_id'] = config.tenant_id
        return index_config

    def add_vectors(self, tenant_id: str, vectors: List[Dict]) -> Dict[str, Any]:
        """テナント別ベクトル追加"""
        config = self.tenant_configs[tenant_id]

        # 容量制限チェック
        current_count = self._get_vector_count(tenant_id)
        if current_count + len(vectors) > config.max_vectors:
            raise ValueError(f"Vector limit exceeded for tenant {tenant_id}")

        # データ暗号化（コンプライアンス要件に応じて）
        if 'HIPAA' in config.compliance_requirements:
            vectors = self._encrypt_vectors(vectors, config.encryption_key)

        # インデックスへの追加
        index_config = self.tenant_indexes[tenant_id]
        result = self._add_to_index(index_config, vectors)

        # 監査ログ
        self._log_tenant_activity(tenant_id, 'add_vectors', len(vectors))

        return result

    def search_vectors(self, tenant_id: str, query_vector: np.ndarray,
                      k: int = 10) -> List[Dict]:
        """テナント分離検索"""
        config = self.tenant_configs[tenant_id]

        # レート制限チェック
        if not self._check_query_quota(tenant_id):
            raise ValueError(f"Query rate limit exceeded for tenant {tenant_id}")

        # テナント専用検索実行
        index_config = self.tenant_indexes[tenant_id]
        results = self._search_tenant_index(index_config, query_vector, k)

        # データ復号化（必要に応じて）
        if 'HIPAA' in config.compliance_requirements:
            results = self._decrypt_results(results, config.encryption_key)

        # クォータ更新
        self._update_query_quota(tenant_id)

        return results

    def _check_query_quota(self, tenant_id: str) -> bool:
        """クエリレート制限チェック"""
        config = self.tenant_configs[tenant_id]
        quota = self.query_quotas[tenant_id]

        current_time = time.time()
        current_minute = int(current_time // 60)

        # 分が変わった場合はリセット
        if current_minute != quota['current_minute']:
            quota['current_minute'] = current_minute
            quota['query_count'] = 0
            quota['last_reset'] = current_time

        return quota['query_count'] < config.max_queries_per_minute

    def _log_tenant_activity(self, tenant_id: str, activity: str,
                           metadata: Any) -> None:
        """テナント活動のログ記録"""
        log_entry = {
            'timestamp': time.time(),
            'tenant_id': tenant_id,
            'activity': activity,
            'metadata': metadata,
            'compliance_logged': True
        }

        # ログをセキュアストレージに保存
        # 実際の実装では適切なログ管理システムを使用
        print(f"AUDIT_LOG: {json.dumps(log_entry)}")

# 使用例
tenant_manager = MultiTenantVectorManager()

# エンタープライズテナントの登録
enterprise_config = tenant_manager.register_tenant(
    "healthcare_corp",
    TenantTier.ENTERPRISE,
    custom_config={
        'compliance': ['HIPAA', 'SOC2'],
        'data_residency': 'US'
    }
)

print(f"Registered tenant: {enterprise_config}")