【セマンティック検索実装】BGE-M3+FAISSでドキュメント駆動開発革命！3ヶ月で検索効率10倍アップ

grep -r "JWT" docs/

Claude Code (MCP Client)
    ↓ MCP Protocol
semantic_server.py (MCP Server)
    ↓ Python API
HybridSearchManager
    ├── SemanticRetriever (BGE-M3 + FAISS)
    └── PatternMatcher (正規表現)

class SemanticEmbedder:
    def __init__(self):
        # BGE-M3モデルの初期化
        self.model = load_embedding_model('BAAI/bge-m3')
        self.device = self._detect_optimal_device()
        self.batch_size = self._calculate_batch_size()

    def encode(self, texts):
        # テキストをベクター化
        embeddings = self.model.encode(
            texts,
            batch_size=self.batch_size,
            convert_to_numpy=True
        )
        return embeddings

# 最初のテストコード（2024年10月20日深夜2:30）
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-m3')

test_queries = [
    "JWT認証の実装方法",
    "データベース設計パターン", 
    "React パフォーマンス最適化"
]

test_documents = [
    "token-based authentication using JWT",
    "OAuth 2.0 implementation guide",
    "database normalization strategies",
    "React.memo optimization techniques"
]

for query in test_queries:
    query_embedding = model.encode([query])
    doc_embeddings = model.encode(test_documents)

    # コサイン類似度計算
    similarities = cosine_similarity(query_embedding, doc_embeddings)[0]

    print(f"\nクエリ: {query}")
    for i, sim in enumerate(similarities):
        print(f"  {test_documents[i]}: {sim:.3f}")

クエリ: JWT認証の実装方法
  token-based authentication using JWT: 0.847
  OAuth 2.0 implementation guide: 0.734
  database normalization strategies: 0.112
  React.memo optimization techniques: 0.098

class FAISSVectorStore:
    def __init__(self):
        self.dimension = 1024  # BGE-M3準拠
        self.index = self.create_optimized_index()

    def create_optimized_index(self):
        # データ量に応じたインデックス最適化
        if self.is_small_dataset():
            return create_flat_index()      # 小規模: 完全精度
        elif self.is_medium_dataset():
            return create_ivf_index(10)     # 中規模: 速度と精度のバランス
        else:
            return create_ivf_index(100)    # 大規模: 高速検索

    def search(self, query_vector, k=5):
        # ベクター類似検索を実行
        return self.index.search(query_vector, k)

class GPUOptimizer:
    def __init__(self):
        self.vram_limit = self._detect_vram_capacity()  # 16GB（実測）
        self.batch_size = self._calculate_optimal_batch_size()

    def _calculate_optimal_batch_size(self):
        # VRAM使用量に基づくバッチサイズ最適化
        memory_per_sample = self._estimate_memory_usage()
        available_memory = self.vram_limit * 0.8  # 安全マージン
        optimal_batch = int(available_memory // memory_per_sample)
        return min(optimal_batch, 128)  # 最大バッチサイズ制限

    def optimize_for_gpu(self):
        # GPU最適化設定
        self._enable_mixed_precision()  # fp16使用でVRAM削減
        self._setup_memory_management()  # 自動キャッシュクリア
        torch.cuda.empty_cache()

class MemoryEfficientSearcher:
    def __init__(self, max_memory_mb=2048):
        self.max_memory = max_memory_mb * 1024 * 1024  # MB to bytes
        self.chunk_cache = {}
        self.lru_cache = LRUCache(maxsize=100)

    def chunked_embedding(self, documents, chunk_size=50):
        """大量文書のチャンク分割エンベディング"""
        embeddings = []

        for i in range(0, len(documents), chunk_size):
            chunk = documents[i:i + chunk_size]
            chunk_embeddings = self.model.encode(chunk)
            embeddings.extend(chunk_embeddings)

            # メモリ監視
            if self._get_memory_usage() > self.max_memory:
                self._clear_cache()

        return np.array(embeddings)

    def _get_memory_usage(self):
        import psutil
        return psutil.Process().memory_info().rss

    def _clear_cache(self):
        self.chunk_cache.clear()
        torch.cuda.empty_cache() if torch.cuda.is_available() else None

class SemanticSearchServer:
    def __init__(self):
        self.semantic_retriever = SemanticRetriever()
        self.pattern_matcher = PatternMatcher()
        self.error_handler = ErrorHandler()
        self.rate_limiter = RateLimiter(max_requests=100)

    async def search(self, query: str, k: int = 5, mode: str = "auto"):
        """統合検索ツール（MCP対応版）"""
        try:
            # レート制限チェック
            await self.rate_limiter.acquire()

            # 検索モードの自動判定
            if mode == "auto":
                mode = await self.detect_search_mode(query)

            # 適切な検索手法を選択
            if mode == "semantic":
                results = await self.semantic_search(query, k)
            else:
                results = await self.pattern_search(query, k)

            return self._format_results(results)

        except Exception as e:
            return await self.error_handler.handle_search_error(e, query)

    async def detect_search_mode(self, query: str) -> str:
        # 概念的クエリか具体的クエリかをAI判定
        conceptual_indicators = [
            "実装", "設計", "パターン", "ベストプラクティス", 
            "最適化", "アーキテクチャ", "戦略", "手法"
        ]

        return "semantic" if any(indicator in query for indicator in conceptual_indicators) else "pattern"

// .mcp.json設定（試行錯誤の痕跡）
{
  "semantic-search": {
    "type": "stdio",
    "command": "python",
    "args": [
      "semantic_server.py",
      "--log-level", "debug",
      "--timeout", "30"
    ],
    "cwd": ".",
    "env": {
      "PYTHONPATH": ".",
      "TOKENIZERS_PARALLELISM": "false"
    }
  }
}

def _format_results(self, results):
    """検索結果をMCP準拠フォーマットに変換"""
    formatted = []
    for result in results:
        try:
            # 日本語文字の安全なエスケープ処理
            safe_content = self._sanitize_japanese_text(result.content)
            safe_metadata = self._sanitize_metadata(result.metadata)

            formatted.append({
                "score": float(result.similarity_score),
                "file_path": result.file_path,
                "content": safe_content,
                "metadata": safe_metadata,
                "line_numbers": result.line_numbers
            })
        except Exception as e:
            self.logger.warning(f"結果フォーマット化エラー: {e}")
            continue

    return formatted

def _sanitize_japanese_text(self, text: str) -> str:
    """日本語テキストの安全な処理"""
    import unicodedata
    # 制御文字の除去
    cleaned = ''.join(char for char in text if unicodedata.category(char)[0] != 'C')
    # 長すぎるテキストの切り捨て
    return cleaned[:1000] + '...' if len(cleaned) > 1000 else cleaned

検索結果:
1. jwt-implementation-guide.md (類似度: 0.923)
   JWTトークンベースの認証システム実装の詳細手順...

2. oauth2-integration-patterns.md (類似度: 0.847) 
   OAuth 2.0プロバイダー統合のベストプラクティス...

3. session-security-architecture.md (類似度: 0.756)
   セッションベース認証のセキュリティ考慮事項...

# 田中さんの悩み：「認証の実装例を探したい」
$ find . -name "*.md" | grep auth
# → auth-ui.md, authentication-flow.md の2件のみ

$ grep -r "JWT" docs/
# → 27件ヒットするが、大半は無関係な内容

# 結果：30分かけて手動で文書を確認する羽目に

# Claude Codeから直接実行
mcp__semantic_search__search(query="認証システムの実装パターン")
# → わずか0.8秒で関連文書が意味的類似性でランク付けされて表示

query = "データベース設計のベストプラクティス"
results = await semantic_search(query, k=8)

# 検索結果（類似度順）:
# 1. orm-design-patterns.md (0.891) - ORMアーキテクチャの設計原則
# 2. index-optimization-strategies.md (0.834) - インデックス設計戦略
# 3. database-normalization-guide.md (0.798) - 正規化理論の実践
# 4. query-performance-tuning.md (0.776) - クエリ最適化手法
# 5. data-consistency-patterns.md (0.745) - データ整合性保証パターン

query = "React パフォーマンス最適化の実装"
results = await semantic_search(query, k=6)

# 検索結果:
# 1. react-memo-optimization.md (0.923) - React.memo とコンポーネント最適化
# 2. virtual-scrolling-implementation.md (0.867) - 仮想スクロール実装
# 3. bundle-splitting-strategies.md (0.845) - バンドル分割最適化
# 4. lazy-loading-patterns.md (0.812) - 遅延読み込みパターン
# 5. state-management-performance.md (0.789) - 状態管理の最適化

query = "メモリリーク対応とデバッグ手法"
results = await semantic_search(query, k=5)

# 検索結果:
# 1. memory-leak-debugging-guide.md (0.934) - メモリプロファイリング手法
# 2. nodejs-memory-optimization.md (0.887) - Node.js メモリ管理
# 3. garbage-collection-tuning.md (0.841) - GC チューニング戦略
# 4. production-monitoring-setup.md (0.798) - 本番監視設定
# 5. incident-response-checklist.md (0.756) - 障害対応チェックリスト

class PerformanceOptimizer:
    def __init__(self):
        self.query_cache = initialize_cache()
        self.batch_processor = BatchProcessor()

    async def optimized_search(self, query: str):
        # キャッシュ戦略による高速化
        if self.is_cached(query):
            return self.get_from_cache(query)

        # バッチ処理による効率化
        embedding = await self.create_embedding_batch([query])

        # ベクター検索実行
        results = await self.execute_vector_search(embedding)

        # 結果をキャッシュに保存
        self.cache_results(query, results)
        return results

    def adaptive_batch_sizing(self):
        # GPU/CPU リソースに応じた動的バッチサイズ調整
        return calculate_optimal_batch_size()

class AutoUpdateSystem:
    def __init__(self, threshold_seconds: int = 86400):  # 24時間
        self.threshold = threshold_seconds
        self.file_monitor = FileMonitor()

    async def check_and_update(self):
        """変更検出に基づく自動更新"""
        stats = await self.get_stats()
        last_scan = stats.get("time_since_last_scan", 0)

        if last_scan > self.threshold:
            changed_files = await self.file_monitor.get_changes()
            if changed_files:
                await self.incremental_update(changed_files)

# 更新戦略の設定例
await configure_update_strategy(
    strategy="auto",           # 自動更新
    time_threshold=86400,      # 24時間後
    count_threshold=100        # 100回検索後
)

# 解決策：バッチサイズ動的調整
def handle_vram_shortage(self):
    self.batch_size = max(self.batch_size // 2, 1)
    self.use_fp16 = True  # 混合精度でメモリ削減
    torch.cuda.empty_cache()  # キャッシュクリア

# 解決策：自動再構築
async def handle_index_corruption(self):
    await self.clear_index()
    await self.rebuild_index()
    logger.info("インデックスを再構築しました")