DOCUMENTATION
/
はじめるにあたり
ガイド
開発者
参照情報
リリース
チュートリアル
Snowflake Open Catalog
ステータス

チュートリアル1:Cortex Searchで簡単な検索アプリケーションを構築する

ガイドSnowflake AI & MLCortex Searchチュートリアルチュートリアル1:シンプルな検索アプリケーションの構築

チュートリアル1:Cortex Searchで簡単な検索アプリケーションを構築する¶

概要¶

このチュートリアルでは、簡単な検索アプリケーションのためにCortex Searchを使い始める方法を説明します。

学習内容¶

  • AirBnb リストデータセットからCortex Search Serviceを作成します。

  • Cortex Search Serviceにクエリできる Streamlit in Snowflake アプリを作成します。

前提条件¶

このチュートリアルを完了するには、以下の前提条件が必要です。

  • データベーステーブル、仮想ウェアハウスオブジェクト、Cortex Search Service、およびStreamlitアプリを作成するために必要な権限を付与するロールを持つSnowflakeアカウントとユーザーを持っています。

これらの要件を満たす手順については、 Snowflakeを20分で紹介 をご参照ください。

ステップ1: 設定する¶

サンプルデータを取得する¶

Huggingfaceでホストされている サンプルデータセットを使用し、単一の JSON ファイルとしてダウンロードします。このリンクをクリックして、ブラウザから直接ファイルをダウンロードします。

  • AirBnB リストデータセット

注釈

チュートリアル以外の設定では、自分のデータ(おそらくすでにSnowflakeテーブルにある)を持参することになります。

データベース、テーブル、ウェアハウスの作成¶

次のステートメントを実行して、データベース、2つのテーブル(csvおよびJSONデータ用)、およびこのチュートリアルに必要な仮想ウェアハウスを作成します。チュートリアルの完了後に、これらのオブジェクトをドロップできます。

CREATE DATABASE IF NOT EXISTS cortex_search_tutorial_db;

CREATE OR REPLACE WAREHOUSE cortex_search_tutorial_wh WITH
     WAREHOUSE_SIZE='X-SMALL'
     AUTO_SUSPEND = 120
     AUTO_RESUME = TRUE
     INITIALLY_SUSPENDED=TRUE;
Copy

次の点に注意してください。

  • CREATE DATABASE ステートメントはデータベースを作成します。データベースには、「public」という名前のスキーマが自動的に含まれます。

  • CREATE WAREHOUSE ステートメントは、中断された初期状態のウェアハウスを作成します。このステートメントは AUTO_RESUME = true も設定します。これにより、コンピューティングリソースを必要とする SQL ステートメントを実行すると、ウェアハウスが自動的に開始されます。

ステップ2: データをSnowflakeにロードする¶

検索サービスを作成する前に、Snowflakeにサンプルデータをロードする必要があります。

データセットのアップロードは、 Snowsight または SQL を使用します。 Snowsight にアップロードするには:

  1. 左のナビゲーションバーの上にある + Create ボタンを選択します。

  2. 次に Table » From File を選択します。

    Snowsight UI でデータを追加する
  3. 右上のドロップダウンから、新しく作成したウェアハウスをテーブルのウェアハウスとして選択します。

  4. JSON データファイルをダイアログにドラッグ&ドロップします。

  5. 上記で作成したデータベースを選択し、 PUBLIC スキーマを指定します。

  6. 最後に、 airbnb_listings という新しいテーブルの作成を指定し、 Next を選択します。

    Snowsight UI でデータを追加する
  7. Load Data into Table ダイアログで、以下の調整を行います。まず、このチュートリアルに適用されないため、 image_embeddings 、 images 、 text_embeddings の列のチェックを外します。次に、 amenities フィールドのデータ型を ARRAY に調整します。

    Snowsight UI でデータを追加する
    Snowsight UI でデータを追加する
  8. これらの調整を終えたら、 Load を選択して次に進みます。

  9. しばらくすると、データが読み込まれたことを示す確認ページが表示されます。

    Snowsight UI でデータを追加する
  10. Query Data を選択し、次のステップで使用する新しいSnowsightワークシートを開きます。

ステップ3: 検索サービスを作成する¶

次の SQL コマンドを実行して、新しいテーブルに対する検索サービスを作成します。

CREATE OR REPLACE CORTEX SEARCH SERVICE cortex_search_tutorial_db.public.airbnb_svc
ON listing_text
ATTRIBUTES room_type, amenities
WAREHOUSE = cortex_search_tutorial_wh
TARGET_LAG = '1 hour'
AS
    SELECT
        room_type,
        amenities,
        price,
        cancellation_policy,
        ('Summary\n\n' || summary || '\n\n\nDescription\n\n' || description || '\n\n\nSpace\n\n' || space) as listing_text
    FROM
    cortex_search_tutorial_db.public.airbnb_listings;
Copy
このコマンドの引数を分解してみましょう。
  • ON パラメーターは、検索するクエリの列を指定します。この場合、 listing_text、ベーステーブルの複数のテキスト列の連結としてソースクエリで生成されます。

  • ATTRIBUTES パラメーターは、検索結果をフィルターする列を指定します。この例では、 listing_text 列にクエリを発行する際に、 room_type と amenities のファイラーを使用します。

  • WAREHOUSE パラメーターと TARGET_LAG パラメーターはそれぞれ、ユーザーが提供するウェアハウスと、検索サービスの希望する鮮度を指定します。この例では、 cortex_search_tutorial_wh ウェアハウスを使用してインデックスを作成し、リフレッシュを実行し、サービスをソーステーブル AIRBNB_LISTINGS から '1 hour' 以内に保つことを指定しています。

  • AS フィールドは、サービスのソーステーブルを定義します。この例では、クエリが複数のフィールドを検索できるように、元のテーブルの複数のテキスト列を検索列 listing_text に連結しています。

ステップ4: Streamlitアプリを作成する¶

Python SDK (snowflake Pythonパッケージを使用)でサービスにクエリすることができます。このチュートリアルでは、 Streamlit in Snowflake アプリケーションでPython SDK を使用する例を示します。

まず、グローバル Snowsight UI ロールが、サービス作成ステップでサービスを作成するために使用したロールと同じであることを確認します。

  1. Snowsight にサインインします。

  2. 左側のナビゲーションメニューで Projects » Streamlit を選択します。

  3. + Streamlit App を選択します。

  4. 重要: アプリの場所の cortex_search_tutorial_db データベースと public スキーマを選択します。

  5. Streamlit in Snowflake エディターの左ペインで、 Packages を選択し、 snowflake (バージョン>= 0.8.0)を追加して、アプリケーションにパッケージをインストールします。

  6. サンプルアプリケーションコードを以下のStreamlitアプリに置き換えてください。

    # Import python packages
    import streamlit as st
    from snowflake.core import Root
    from snowflake.snowpark.context import get_active_session
    
    # Constants
    DB = "cortex_search_tutorial_db"
    SCHEMA = "public"
    SERVICE = "airbnb_svc"
    BASE_TABLE = "cortex_search_tutorial_db.public.airbnb_listings"
    ARRAY_ATTRIBUTES = {"AMENITIES"}
    
    
    def get_column_specification():
        """
        Returns the name of the search column and a list of the names of the attribute columns
        for the provided cortex search service
        """
        session = get_active_session()
        search_service_result = session.sql(f"DESC CORTEX SEARCH SERVICE {DB}.{SCHEMA}.{SERVICE}").collect()[0]
        st.session_state.attribute_columns = search_service_result.attribute_columns.split(",")
        st.session_state.search_column = search_service_result.search_column
        st.session_state.columns = search_service_result.columns.split(",")
    
    def init_layout():
        st.title("Cortex AI Search")
        st.markdown(f"Querying service: `{DB}.{SCHEMA}.{SERVICE}`".replace('"', ''))
    
    def query_cortex_search_service(query, filter={}):
        """
        Queries the cortex search service in the session state and returns a list of results
        """
        session = get_active_session()
        cortex_search_service = (
            Root(session)
            .databases[DB]
            .schemas[SCHEMA]
            .cortex_search_services[SERVICE]
        )
        context_documents = cortex_search_service.search(
            query,
            columns=st.session_state.columns,
            filter=filter,
            limit=st.session_state.limit)
        return context_documents.results
    
    @st.cache_data
    def distinct_values_for_attribute(col_name, is_array_attribute=False):
        session = get_active_session()
        if is_array_attribute:
            values = session.sql(f'''
            SELECT DISTINCT value FROM {BASE_TABLE},
            LATERAL FLATTEN(input => {col_name})
            ''').collect()
        else:
            values = session.sql(f"SELECT DISTINCT {col_name} AS VALUE FROM {BASE_TABLE}").collect()
        return [ x["VALUE"].replace('"', "") for x in values ]
    
    def init_search_input():
        st.session_state.query = st.text_input("Query")
    
    def init_limit_input():
        st.session_state.limit = st.number_input("Limit", min_value=1, value=5)
    
    def init_attribute_selection():
        st.session_state.attributes = {}
        for col in st.session_state.attribute_columns:
            is_multiselect = col in ARRAY_ATTRIBUTES
            st.session_state.attributes[col] = st.multiselect(
                col,
                distinct_values_for_attribute(col, is_array_attribute=is_multiselect)
            )
    
    def display_search_results(results):
        """
        Display the search results in the UI
        """
        st.subheader("Search results")
        for i, result in enumerate(results):
            result = dict(result)
            container = st.expander(f"[Result {i+1}]", expanded=True)
    
            # Add the result text.
            container.markdown(result[st.session_state.search_column])
    
            # Add the attributes.
            for column, column_value in sorted(result.items()):
                if column == st.session_state.search_column:
                    continue
                container.markdown(f"**{column}**: {column_value}")
    
    def create_filter_object(attributes):
        """
        Create a filter object for the search query
        """
        and_clauses = []
        for column, column_values in attributes.items():
            if len(column_values) == 0:
                continue
            if column in ARRAY_ATTRIBUTES:
                for attr_value in column_values:
                    and_clauses.append({"@contains": { column: attr_value }})
            else:
                or_clauses = [{"@eq": {column: attr_value}} for attr_value in column_values]
                and_clauses.append({"@or": or_clauses })
    
        return {"@and": and_clauses} if and_clauses else {}
    
    
    def main():
        init_layout()
        get_column_specification()
        init_attribute_selection()
        init_limit_input()
        init_search_input()
    
        if not st.session_state.query:
            return
        results = query_cortex_search_service(
            st.session_state.query,
            filter = create_filter_object(st.session_state.attributes)
        )
        display_search_results(results)
    
    
    if __name__ == "__main__":
        st.set_page_config(page_title="Cortex AI Search and Summary", layout="wide")
        main()
    
    Copy

上記のStreamlit-in-Snowflakeコードの主なコンポーネントを簡単に説明します。

  • get_column_specification は、 DESCRIBE SQL クエリを使用して、検索サービスで利用可能な属性に関する情報を取得し、Stremalit 状態に保存します。

  • init_layout はページのヘッダーとイントロを設定します。

  • query_cortex_search_service はPythonクライアントライブラリを介してCortex Search Serviceへのクエリを処理します。

  • create_filter_object は、Streamlitフォームから選択されたフィルター属性を、Cortex Searchへのクエリ用にPythonライブラリで使用される適切なオブジェクトに処理します。

  • distinct_values_for_attribute は、ドロップダウンメニューに入力するフィルタリング可能な属性の値を決定します。

  • init_search_input、 init_limit_input、 init_attribute_selection は、検索クエリの入力、結果数の制限、属性フィルターを初期化します。

  • display_search_results は、検索結果を結果ページに表示されるMarkdown要素にフォーマットします。

ステップ5: クリーンアップする¶

クリーンアップ(オプション)¶

次の DROP <オブジェクト> コマンドを実行して、システムをチュートリアルを開始する前の状態に戻します。

DROP DATABASE IF EXISTS cortex_search_tutorial_db;
DROP WAREHOUSE IF EXISTS cortex_search_tutorial_wh;
Copy

データベースをドロップすると、テーブルなどのすべての子データベースオブジェクトが自動的に削除されます。

次のステップ¶

おめでとうございます。あなたはSnowflakeでテキストデータの簡単な検索アプリを構築することに成功しました。 チュートリアル2 に進んで、 Cortex LLM 関数 を重ねて、Cortex Searchで AI チャットボットを構築する方法を見ることができます。

追加のリソース¶

さらに、以下のリソースを使って学習を続けることもできます。

  • Cortex Searchの概要

  • Cortex Search Serviceにクエリする

このページは役に立ちましたか?

Snowflakeに移動する
会話に参加する
Snowflakeで開発する
フィードバックを共有する
ブログでの最新情報を表示する
独自の認定を取得する
プライバシー通知サイト規約© 2025 Snowflake, Inc. All Rights Reserved.
言語: 日本語
  • English
  • Français
  • Deutsch
  • 日本語
  • 한국어
  • Português