Snowflake Cortex Search徹底理解｜ハイブリッド検索とRAG活用

Snowflake Cortex Searchは、キーワード検索とセマンティック検索を融合したハイブリッド検索をマネージドサービスとして提供します。 エンベディング生成・インデックス構築・スコア融合がすべて自動化されており、ユーザーはCREATE CORTEX SEARCH SERVICEでサービスを定義するだけで、 高品質な検索機能をアプリケーションに組み込めます。 RAG（Retrieval-Augmented Generation）のリトリーバとして利用することで、Cortex LLM関数と組み合わせた社内ドキュメントQAシステムを構築できます。

ハイブリッド検索とは

Cortex Searchが採用するハイブリッド検索は、2つの検索手法を内部的に組み合わせています。

検索フロー:

[ユーザークエリ: "Snowflakeのデータ保護機能"]
       │
       ├──→ キーワード検索 ──→ BM25スコア
       │     └─ "Snowflake", "データ", "保護" で転置インデックスを走査
       │
       ├──→ セマンティック検索 ──→ ベクトル類似度スコア
       │     └─ クエリをエンベディング化 → 近似近傍探索
       │
       └──→ スコア融合（Reciprocal Rank Fusion）
              └─ 最終ランキング結果を返却

CREATE CORTEX SEARCH SERVICE

Cortex Searchサービスの作成はDDLで行います。ソーステーブル・検索対象カラム・フィルタカラム・更新頻度を指定します。

CREATE OR REPLACE CORTEX SEARCH SERVICE knowledge_base_search
  ON doc_content                    -- 検索対象のテキストカラム
  ATTRIBUTES doc_title, category    -- フィルタ用カラム（任意）
  WAREHOUSE = search_wh             -- インデックス構築用WH
  TARGET_LAG = '1 hour'             -- ソースデータの反映遅延
AS (
  SELECT
    doc_id,
    doc_title,
    doc_content,
    category,
    updated_at
  FROM knowledge_base
  WHERE is_published = TRUE
);

主要パラメータ

検索クエリの実行

作成したCortex Searchサービスへのクエリは、Python SDK（推奨）またはREST APIで実行します。

# Python SDKでの検索
from snowflake.core import Root

root = Root(session)

search_service = (
    root.databases["MY_DB"]
    .schemas["PUBLIC"]
    .cortex_search_services["KNOWLEDGE_BASE_SEARCH"]
)

results = search_service.search(
    query="Snowflakeのタイムトラベル機能について",
    columns=["doc_id", "doc_title", "doc_content", "category"],
    filter={
        "@eq": {"category": "data_protection"}
    },
    limit=5
)

for result in results.results:
    print(f"Title: {result['doc_title']}")
    print(f"Score: {result['search_score']}")
    print(f"Content: {result['doc_content'][:200]}...")
    print("---")

フィルタ構文

ATTRIBUTESで指定したカラムに対して、以下のフィルタ演算子が使えます。

# 完全一致
{"@eq": {"category": "security"}}

# OR条件（複数値）
{"@or": [
  {"@eq": {"category": "security"}},
  {"@eq": {"category": "governance"}}
]}

# AND条件
{"@and": [
  {"@eq": {"category": "security"}},
  {"@eq": {"doc_type": "guide"}}
]}

RAGリトリーバとしての活用

Cortex SearchをリトリーバとしてCortex LLM関数と組み合わせることで、 社内ナレッジに基づいたRAGパイプラインを構築できます。

RAGパイプライン構成:

[ユーザー質問]
       │
       ▼
[Cortex Search]  ← リトリーバ
       │ 関連ドキュメント Top-K を取得
       ▼
[プロンプト構築]
       │ "以下のコンテキストに基づいて回答してください:
       │  {検索結果のdoc_content}"
       ▼
[SNOWFLAKE.CORTEX.COMPLETE()]  ← ジェネレータ
       │ コンテキスト付きで回答を生成
       ▼
[回答をユーザーに返却]

# RAGパイプラインの実装例（Streamlit in Snowflake）
import streamlit as st
from snowflake.core import Root
from snowflake.snowpark.context import get_active_session

session = get_active_session()
root = Root(session)

search_service = (
    root.databases["KB_DB"]
    .schemas["PUBLIC"]
    .cortex_search_services["KB_SEARCH"]
)

user_question = st.text_input("質問を入力してください")

if user_question:
    # Step 1: Cortex Searchで関連ドキュメント取得
    results = search_service.search(
        query=user_question,
        columns=["doc_content"],
        limit=3
    )

    context = "\n\n".join([r["doc_content"] for r in results.results])

    # Step 2: COMPLETE関数で回答生成
    prompt = f"""以下のコンテキストのみに基づいて質問に回答してください。
コンテキストに情報がない場合は「情報が見つかりませんでした」と回答してください。

コンテキスト:
{context}

質問: {user_question}"""

    answer = session.sql(
        "SELECT SNOWFLAKE.CORTEX.COMPLETE('llama3.1-70b', ?)",
        params=[prompt]
    ).collect()[0][0]

    st.write(answer)

コストモデル

• インデックス構築: 指定したウェアハウスのクレジットを消費
• インデックスの差分更新: TARGET_LAGに応じてサーバーレスクレジットを消費
• 検索クエリ実行: サーバーレスクレジットを消費（ウェアハウス不要）
• ストレージ: インデックスデータのストレージ費用が発生

Cortex Searchと他手法の比較

試験対策のポイント

• Cortex Searchはキーワード＋セマンティックのハイブリッド検索サービス
• CREATE CORTEX SEARCH SERVICEでON（検索対象）、TARGET_LAG（更新頻度）を指定
• 検索クエリ実行にウェアハウスは不要（サーバーレスコンピュート）
• ATTRIBUTESで指定したカラムはメタデータフィルタに使用
• RAGパイプラインではリトリーバとして活用し、COMPLETE関数と組み合わせる
• VECTOR型を使った手動のベクトル検索との違いを理解する

サンプル問題

よくある質問