Zenn
❄️

Cortex Analystを使ってネイティブアプリを作成する

2025/01/28に公開
Streamlit
Snowflake
cortex
LLM
tech

はじめに

Cortex Analystの記事としては3本目になります。1,2本目の記事は↓で、この2つの記事でCortex Analystを使ったStreamlitアプリ作成をしています。

https://zenn.dev/tree_and_tree/articles/fe6f261748b560
https://zenn.dev/tree_and_tree/articles/e1c9dc57cdde28

今回は、作成したアプリをネイティブアプリケーションとして公開する話をしようと思います。
Cortex Analystの概要や基本的なText to SQLアプリの話は上記記事を参照ください。

ネイティブアプリとは

https://docs.snowflake.com/ja/developer-guide/native-apps/native-apps-about

Snowflake Native App Frameworkのことで、Snowflake内部でアプリを作成し、そのアプリを公開することのできるフレームワークを指します。Streamlit in Snowflakeでアプリを作成してもアカウント内部でしか利用できませんが、ネイティブアプリにすることで外部に公開することができ、プロダクトとして提供することができます。

アプリは以下のパターンで公開できます。

  • 特定のコンシューマー（アプリ利用側のSnowflakeアカウント）に向けて共有（この記事ではこちらを扱います）
  • マーケットプレイスで一般公開

公式ドキュメントにチュートリアルがあり、とりあえず触ってみたい方はチュートリアルをやってみるのがおすすめです。
https://docs.snowflake.com/ja/developer-guide/native-apps/tutorials/getting-started-tutorial

Cortex Analystを使ってネイティブアプリを開発する

今回は、筆者が実際に開発した内容をベースに説明したいと思います。

作成したネイティブアプリの全体像

アプリケーションパッケージというものを作成し、その中にアプリに必要なテーブル、Streamlitを動かすpyファイルなどを入れてパッケージにします。
コンシューマーは、そのパッケージをインストールすることでアプリを利用することができます。

用語の補足

  • アプリケーションパッケージ: アプリケーションに必要なデータコンテンツ、アプリケーションロジック、メタデータ、セットアップスクリプトをカプセル化したコンテナー
  • アプリケーション：アプリケーションパッケージから作成できるオブジェクト。アプリをインストールするとアプリケーションオブジェクトが作成されます

作成手順

ローカルでファイルを整備

以下のフォルダ構成で、それぞれ必要なファイルを準備していきます。

フォルダ構成

CortexAnalyst_app/		プロジェクトフォルダ（任意の名前でOK）
├ readme.md			    readme
├ manifest.yml		    マニフェストファイル
├ scripts/
│   └ setup.sql		    セットアップスクリプト
└ streamlit/			Streamlitフォルダ
    └ cortex_analyst.py
    └ environment.yml	インストールする外部ライブラリのバージョンを明示的に指定

主なファイルの説明

  • manifest.yml
    セットアップスクリプトの場所やバージョン定義など、アプリケーションパッケージに必要なプロパティを定義するファイル
  • setup.sql
    アプリケーションパッケージをもtにアプリが作成される際に自動的に実行されるSQLクエリを記載するファイル
  • cortex_analyst.py
    Streamlitのアプリケーションを記述するファイル
  • environment.yml
    Pythonの外部ライブラリを明示的に指定するのに使用するファイル（任意）
各種ファイルのソースコード（例）

manifest.yml

manifest_version: 1

artifacts:
  setup_script: scripts/setup.sql
  readme: readme.md

configuration:
  log_level: debug
  trace_level: always

setup.sql

-- アプリケーションロールの作成とスキーマの作成
CREATE APPLICATION ROLE IF NOT EXISTS app_public;

-- ビューのスキーマ作成
CREATE OR ALTER VERSIONED SCHEMA code_schema;
GRANT USAGE ON SCHEMA code_schema TO APPLICATION ROLE app_public;
-- ビューの作成
CREATE VIEW IF NOT EXISTS code_schema.shared_view
  AS SELECT *
  -- FROM shared_data.shared_view;
  FROM shared_data.shared_table;
GRANT SELECT ON VIEW code_schema.shared_view TO APPLICATION ROLE app_public;

-- Streamlitオブジェクトの追加
CREATE STREAMLIT IF NOT EXISTS code_schema.cortex_analyst_pos
  FROM '/streamlit'
  MAIN_FILE = '/cortex_analyst.py'
;
GRANT USAGE ON STREAMLIT code_schema.cortex_analyst_pos TO APPLICATION ROLE app_public;

cortex_analyst.py
※このままでは動かないので注意（データに合わせてセマンティックモデルを記述する必要があります）

import _snowflake
import json
import streamlit as st
import time
from snowflake.snowpark.context import get_active_session

semantic_model = """
[セマンティックモデルを記述]
"""

def send_message(prompt: str) -> dict:
    """Calls the REST API and returns the response."""
    request_body = {
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt
                    }
                ]
            }
        ],
        "semantic_model": semantic_model,
    }
    resp = _snowflake.send_snow_api_request(
        "POST",
        f"/api/v2/cortex/analyst/message",
        {},
        {},
        request_body,
        {},
        30000,
    )
    if resp["status"] < 400:
        return json.loads(resp["content"])
    else:
        raise Exception(
            f"Failed request with status {resp['status']}: {resp}"
        )

def process_message(prompt: str) -> None:
    """Processes a message and adds the response to the chat."""
    st.session_state.messages.append(
        {"role": "user", "content": [{"type": "text", "text": prompt}]}
    )
    with st.chat_message("user"):
        st.markdown(prompt)
    with st.chat_message("assistant"):
        with st.spinner("Generating response..."):
            response = send_message(prompt=prompt)
            content = response["message"]["content"]
            display_content(content=content)
    st.session_state.messages.append({"role": "assistant", "content": content})


def display_content(content: list, message_index: int = None) -> None:
    """Displays a content item for a message."""
    message_index = message_index or len(st.session_state.messages)
    for item in content:
        if item["type"] == "text":
            st.markdown(item["text"])
        elif item["type"] == "suggestions":
            with st.expander("Suggestions", expanded=True):
                for suggestion_index, suggestion in enumerate(item["suggestions"]):
                    if st.button(suggestion, key=f"{message_index}_{suggestion_index}"):
                        st.session_state.active_suggestion = suggestion
        elif item["type"] == "sql":
            with st.expander("SQL Query", expanded=False):
                st.code(item["statement"], language="sql")
            with st.expander("Results", expanded=True):
                with st.spinner("Running SQL..."):
                    session = get_active_session()
                    df = session.sql(item["statement"]).to_pandas()
                    if len(df.index) > 1:
                        data_tab, line_tab, bar_tab = st.tabs(
                            ["Data", "Line Chart", "Bar Chart"]
                        )
                        data_tab.dataframe(df)
                        if len(df.columns) > 1:
                            df = df.set_index(df.columns[0])
                        with line_tab:
                            st.line_chart(df)
                        with bar_tab:
                            st.bar_chart(df)
                    else:
                        st.dataframe(df)


st.title("Cortex analyst")
st.markdown(f"Semantic Model: `{FILE}`")

if "messages" not in st.session_state:
    st.session_state.messages = []
    st.session_state.suggestions = []
    st.session_state.active_suggestion = None

for message_index, message in enumerate(st.session_state.messages):
    with st.chat_message(message["role"]):
        display_content(content=message["content"], message_index=message_index)

if user_input := st.chat_input("What is your question?"):
    process_message(prompt=user_input)

if st.session_state.active_suggestion:
    process_message(prompt=st.session_state.active_suggestion)
    st.session_state.active_suggestion = None

environment.yml

name: sf_env
channels:
- snowflake
dependencies:
- streamlit=1.35.0

ファイルをSnowflakeへアップロード

ローカルで用意したファイルをSnowflakeにアップロードします。

ファイルをアップロードする前に、以下の手順でアプリケーションパッケージを準備していきます。

  • アプリケーションパッケージの作成
  • スキーマ、名前付きステージの作成
  • アプリケーション外部のテーブルを連携
    • 外部データベースを以下のように準備
      • データベース：CORTEX_ANALYST_DEMO、スキーマ：CORTEX_ANALYST_POSを作成
      • ワークシートでTableをテーブルにコピーする
-- アプリケーションパッケージの作成
CREATE APPLICATION PACKAGE cortex_analyst_demo_package;
-- 作成されたことを確認
SHOW APPLICATION PACKAGES;

-- コンテキストを指定したアプリケーションパッケージに指定
USE APPLICATION PACKAGE cortex_analyst_demo_package;
-- スキーマを作成
CREATE SCHEMA stage_content;
-- 名前付きステージを作成
CREATE OR REPLACE STAGE cortex_analyst_demo_package.stage_content.cortex_analyst_demo_stage
  FILE_FORMAT = (TYPE = 'csv' FIELD_DELIMITER = '|' SKIP_HEADER = 1);

-- 外部のデータソースと連携するための作業
-- アプリケーションパッケージにREFERENCE_USAGE権限を付与
GRANT REFERENCE_USAGE ON DATABASE POSIS_DEV
  TO SHARE IN APPLICATION PACKAGE cortex_analyst_demo_package;
-- アプリケーションパッケージにパッケージにスキーマを作成
CREATE SCHEMA IF NOT EXISTS shared_data;
SHOW SCHEMAS LIKE 'SHARED_DATA' IN DATABASE cortex_analyst_demo_package;
-- テーブルを作成
CREATE TABLE IF NOT EXISTS cortex_analyst_demo_package.shared_data.shared_table
CLONE DB.SCHEMA.TABLE;
-- スキーマとテーブルの使用権を付与
GRANT USAGE ON SCHEMA cortex_analyst_demo_package.shared_data 
  TO SHARE IN APPLICATION PACKAGE cortex_analyst_demo_package;
GRANT SELECT ON TABLE cortex_analyst_demo_package.shared_data.shared_table
  TO SHARE IN APPLICATION PACKAGE cortex_analyst_demo_package;

アプリケーションパッケージのステージに、ローカルと同じフォルダ構成になるようにファイルをアップロードしていきます。筆者はGUIで行いました。

アップロードされたことを確認

LIST @cortex_analyst_demo_package.stage_content.cortex_analyst_demo_stage;
-- 以下のようなファイル構成になる
-- cortex_analyst_demo_stage/manifest.yml
-- cortex_analyst_demo_stage/readme.md
-- cortex_analyst_demo_stage/scripts/setup.sql
-- cortex_analyst_demo_stage/streamlit/cortex_analyst_pos.py
-- cortex_analyst_demo_stage/streamlit/environment.yml

プロバイダー側でアプリを作成

コンシューマーに提供する前に、プロバイダー（開発側のアカウント）でアプリケーションを作成して動くかどうかを確認します。

-- アプリケーションを作成
CREATE APPLICATION cortex_analyst_demo_app
  FROM APPLICATION PACKAGE cortex_analyst_demo_package
  USING '@cortex_analyst_demo_package.stage_content.cortex_analyst_demo_stage';
-- 作成されたことを確認
SHOW APPLICATIONS;
-- アプリケーションを削除(必要に応じて)
-- DROP APPLICATION cortex_analyst_demo_app;
-- アプリケーションの更新（必要に応じて）
-- ALTER APPLICATION cortex_analyst_demo_app UPGRADE
  -- USING '@cortex_analyst_demo_package.stage_content.cortex_analyst_demo_stage';

SHOW APPLICATIONSで作成されたアプリが出てくる

GUIで作成したアプリを確認する

バージョンとパッチを設定

アプリケーション公開の前にアプリのバージョンとパッチ設定します

-- バージョンを設定
ALTER APPLICATION PACKAGE cortex_analyst_demo_package
  ADD VERSION v1_0 USING '@cortex_analyst_demo_package.stage_content.cortex_analyst_demo_stage';
-- アプリケーションのバージョンとパッチを表示
SHOW VERSIONS IN APPLICATION PACKAGE cortex_analyst_demo_package;

-- デフォルトのリリースディレクティブを設定
ALTER APPLICATION PACKAGE cortex_analyst_demo_package
  SET DEFAULT RELEASE DIRECTIVE
  VERSION = v1_0
  PATCH = 0;

アプリケーションを公開

マーケットプレイスで一般公開することもできますが、今回は特定のコンシューマーに向けて公開してみます。

プロバイダーStudioを開き、リストを作成する

リスト名をつけ、「指定したコンシューマーのみ」を選択して次へ

リストにアプリケーションパッケージを追加し、説明を書く

インストールしたいアカウントのアカウント識別子を入力して公開

コンシューマー側のインストール

プロバイダー側で用意したアプリをインストールし、コンシューマー側で使えるようにします。

データ製品→アプリを開くと、共有されているアプリが表示されるので、「取得」を押してインストールします。
アンインストールしてしまっても、データ製品→プライベート共有から再びインストール可能です。

ウェアハウスを選択してインストール

インストール後、アプリを押下すると起動できます。

余談： リファレンスありでのネイティブアプリ作成

開発時の理想形としては、下図のようにアプリケーションのみをユーザーにインストールしてもらい、データソースはユーザー環境にあるものを参照するようなアプリを想定していました。

このような構成にする場合、アプリケーション外のデータを参照する必要があり、通常はリファレンスという機能で対応できるのですが、どうやらCortex Analystのセマンティックモデルがリファレンスに対応していないみたいでうまくいきませんでした。
2024年12月時点で、公式サポートからも対応していないとの連絡があり、この構成は断念しました……

おわりに

Coretx Analystは体感でかなり高精度なText to SQLを実現してくれるサービスで、個人的にビジネス活用の可能性を大いに秘めていると思います。
ネイティブアプリ化ができれば外部公開ができるようになるので、Text to SQLで未来のアプリを配布するのも夢じゃないですね！

Discussion