Streamlit in Snowflake アプリカタログ
SNOWFLAKE WORLD TOUR TOKYO 2025
2025/9/11 - 12の2日間、Snowflake の国内最大のイベント『SNOWFLAKE WORLD TOUR TOKYO 2025』が開催されます!データ活用の最前線を是非体感しに来てください!私は『Snowflake Cortex AISQL を用いた高度な分析手法』というセッションで登壇させていただきます。AISQL の基本から応用までキャッチアップできますので是非ご登録いただければ嬉しいですー!(9/12の方がまだ空席がございます)
はじめに
Streamlit in Snowflake が大好きな紳士淑女の皆様、日々アプリ作りに勤しんでいるといつの間にか大量のアプリが Snowflake 上にできているのではないでしょうか?私はそうです。そしてアプリが増えてくると、どのアプリが何を目的としたアプリなのか、どれぐらい利用頻度の高いアプリなのか把握するのが大変になってくることが想定されます。
そこで Streamlit in Snowflake のアプリカタログを Streamlit in Snowflake で作ってみました!今回は自作アプリのご紹介でございます。もちろんこれで全てが解決するようなものではありませんが、Snowflake におけるアプリ管理の一手法としてご参考になれば幸いです。もし気に入っていただいたり、参考になった際には、フィードバックいただいたり 👍 を頂戴できるととても嬉しいです!
機能概要
機能一覧
- アプリポータル機能
- 検索とフィルタリング
- キーワード検索及びベクトル検索を両立したハイブリッド検索
- アプリの実行回数とアプリのいいね数による並び替え
- タグによるフィルタリング
- アプリケーション一覧の表示
- 簡易なカード表示と、詳細なリスト表示の切り替え
- アプリのいいね機能といいね数の表示
- アプリの過去30日間の実行回数の表示
- アプリのタグの表示
- アプリの説明文の表示
- アプリケーションへの直接遷移
- 検索とフィルタリング
- 管理画面
- ユーザーによるアプリの説明文の修正
- ユーザーによるアプリのタグの修正 (マルチタグに対応)
- 説明文の自動生成機能
- ソースコードからアプリの説明文を生成 AI で自動生成
- アプリの説明文の一括削除
- タグの自動生成機能
- タグに設定するカテゴリの編集
- アプリの説明文からタグを生成 AI で自動生成
- タグの一括削除
- ベクトルデータの自動生成機能
- アプリの説明文からベクトルデータを生成 AI で自動生成
- ベクトルデータの一括削除
完成イメージ
アプリポータルのカード表示
管理画面
説明文の自動生成機能
タグの自動生成機能
ベクトルデータの自動生成機能
前提条件
- Snowflake アカウント
- Cortex LLM が利用できる Snowflake アカウント (クロスリージョン推論のリリースによりクラウドやリージョンの制約がほぼ無くなりました)
- Streamlit in Snowflake のインストールパッケージ
- python 3.11 以降 (特に追加のパッケージは不要です)
手順
新規で Streamlit in Snowflake のアプリを作成
Snowsight の左ペインから『Streamlit』をクリックし、『+ Streamlit』ボタンをクリックし SiS アプリを作成します。
Streamlit in Snowflake のアプリを実行
Streamlit in Snowflake アプリの編集画面で以下コードをコピー&ペーストで貼り付けて完了です。(コードが長くなってしまったためアコーディオンを開いて表示してください)
ソースコード
import streamlit as st
from snowflake.snowpark.context import get_active_session
import pandas as pd
import json
# ページ設定
st.set_page_config(page_title="Streamlit in Snowflake アプリカタログ", layout='wide')
# Snowflakeアカウント情報 (環境に合わせて変更してください)
orgname = "<your_org_name>"
accountname = "<your_account_name>"
# Snowflakeセッションの取得
snowflake_session = get_active_session()
@st.cache_data
def get_streamlit_apps_info():
"""
アカウント全体のStreamlit in Snowflakeアプリ情報を取得します。
Native Appsなどは除外しています。
"""
query = "SHOW STREAMLITS IN ACCOUNT"
result = snowflake_session.sql(query).collect()
df = pd.DataFrame(result)
return df[df['owner_role_type'] == 'ROLE']
@st.cache_data
def create_app_descriptions_table():
"""
APP_DESCRIPTIONSテーブルとAPP_CATEGORIESテーブルが存在しない場合、作成します。
また、APP_DESCRIPTIONSに存在しないアプリの初期データを一括で登録します。
"""
# アプリ説明用テーブル
snowflake_session.sql("""
CREATE TABLE IF NOT EXISTS APP_DESCRIPTIONS (
database_name VARCHAR,
schema_name VARCHAR,
app_name VARCHAR,
description VARCHAR,
tags VARCHAR,
embedding VECTOR(FLOAT, 1024),
views INTEGER DEFAULT 0,
likes INTEGER DEFAULT 0,
last_updated TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
""").collect()
# カテゴリ管理用テーブル
snowflake_session.sql("""
CREATE TABLE IF NOT EXISTS APP_CATEGORIES (
category STRING PRIMARY KEY,
created_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
)
""").collect()
# デフォルトカテゴリの追加 (一括挿入)
default_categories = [
'データ分析', 'データエンジニアリング', 'ダッシュボード', 'BI',
'レポート', '管理ツール', '生成AI', '機械学習'
]
categories_values = ", ".join([f"('{category}')" for category in default_categories])
snowflake_session.sql(f"""
INSERT INTO APP_CATEGORIES (category)
SELECT category FROM (VALUES {categories_values}) AS v(category)
WHERE category NOT IN (SELECT category FROM APP_CATEGORIES)
""").collect()
# 現在のStreamlitアプリ一覧を取得し、一括で更新
apps_df = get_streamlit_apps_info()
if not apps_df.empty:
# アプリ一覧をテンポラリテーブルとして作成
apps_values = ", ".join([
f"('{app['database_name']}', '{app['schema_name']}', '{app['name']}')"
for _, app in apps_df.iterrows()
])
# 一括更新用のSQLを実行
snowflake_session.sql(f"""
INSERT INTO APP_DESCRIPTIONS (database_name, schema_name, app_name)
SELECT src.* FROM (
VALUES {apps_values}
) AS src(database_name, schema_name, app_name)
WHERE NOT EXISTS (
SELECT 1
FROM APP_DESCRIPTIONS dest
WHERE dest.database_name = src.database_name
AND dest.schema_name = src.schema_name
AND dest.app_name = src.app_name
)
""").collect()
def update_app_views():
"""
QUERY_HISTORYからアプリの実行回数を集計し、APP_DESCRIPTIONSテーブルを更新します。
"""
snowflake_session.sql("""
MERGE INTO APP_DESCRIPTIONS target
USING (
SELECT
REGEXP_SUBSTR(QUERY_TEXT, '"([^"]+)"\\."([^"]+)"\\."([^"]+)"', 1, 1, 'e', 1) as database_name,
REGEXP_SUBSTR(QUERY_TEXT, '"([^"]+)"\\."([^"]+)"\\."([^"]+)"', 1, 1, 'e', 2) as schema_name,
REGEXP_SUBSTR(QUERY_TEXT, '"([^"]+)"\\."([^"]+)"\\."([^"]+)"', 1, 1, 'e', 3) as app_name,
COUNT(*) as view_count
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE QUERY_TEXT LIKE 'execute streamlit%'
AND START_TIME >= DATEADD(month, -1, CURRENT_TIMESTAMP())
GROUP BY 1, 2, 3
) as source
ON target.database_name = source.database_name
AND target.schema_name = source.schema_name
AND target.app_name = source.app_name
WHEN MATCHED THEN
UPDATE SET views = source.view_count
""").collect()
def get_app_data():
"""
アプリ情報とメタデータを取得します。
キャッシュが存在しない場合のみデータベースから取得します。
"""
if 'apps_df' not in st.session_state or 'app_descriptions' not in st.session_state:
# アプリの実行回数を更新
update_app_views()
# アプリ情報の取得
apps_df = get_streamlit_apps_info()
app_data = snowflake_session.sql("""
SELECT
DATABASE_NAME,
SCHEMA_NAME,
APP_NAME,
DESCRIPTION,
TAGS,
EMBEDDING,
VIEWS,
LIKES,
LAST_UPDATED
FROM APP_DESCRIPTIONS
""").collect()
# アプリのメタデータを整理
app_descriptions = {}
for row in app_data:
# タグを配列に変換
current_tags = []
if row['TAGS']:
current_tags = [tag.strip() for tag in row['TAGS'].split(',') if tag.strip()]
app_descriptions[f"{row['DATABASE_NAME']}.{row['SCHEMA_NAME']}.{row['APP_NAME']}"] = {
'description': row['DESCRIPTION'],
'tags': current_tags, # 常に配列として保持
'views': row['VIEWS'] if row['VIEWS'] is not None else 0,
'likes': row['LIKES'] if row['LIKES'] is not None else 0,
'last_updated': row['LAST_UPDATED']
}
# session_stateにデータを保存
st.session_state['apps_df'] = apps_df
st.session_state['app_descriptions'] = app_descriptions
return st.session_state['apps_df'], st.session_state['app_descriptions']
def clear_app_cache():
"""
アプリデータのキャッシュをクリアします。
"""
if 'apps_df' in st.session_state:
del st.session_state['apps_df']
if 'app_descriptions' in st.session_state:
del st.session_state['app_descriptions']
def update_app_description(database, schema, app_name, description, tags):
"""
アプリの説明とタグを更新または挿入します。
"""
# タグをカンマ区切りの文字列に変換
tags_str = ','.join([tag['label'] if isinstance(tag, dict) else tag for tag in tags])
snowflake_session.sql("""
MERGE INTO APP_DESCRIPTIONS target
USING (
SELECT
? AS database_name,
? AS schema_name,
? AS app_name,
? AS description,
? AS tags
) AS source
ON target.database_name = source.database_name
AND target.schema_name = source.schema_name
AND target.app_name = source.app_name
WHEN MATCHED THEN
UPDATE SET
description = source.description,
tags = source.tags,
last_updated = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
INSERT (database_name, schema_name, app_name, description, tags)
VALUES (
source.database_name,
source.schema_name,
source.app_name,
source.description,
source.tags
)
""", params=[
database,
schema,
app_name,
description,
tags_str
]).collect()
# キャッシュをクリアして最新データを取得できるようにする
clear_app_cache()
@st.fragment
def render_like_button(app, likes_key):
"""
いいねボタンをレンダリングする関数
"""
# いいねの状態管理
if "like_states" not in st.session_state:
st.session_state.like_states = {}
current_likes = st.session_state[likes_key]
# formのスタイルを無効化
st.markdown("""
<style>
[data-testid="stForm"] {
border: none;
padding: 0;
margin: 0;
}
</style>
""", unsafe_allow_html=True)
# ボタンクリック時の処理
def on_like_click():
# いいねの更新
snowflake_session.sql("""
UPDATE APP_DESCRIPTIONS
SET likes = likes + 1,
last_updated = CURRENT_TIMESTAMP()
WHERE database_name = ?
AND schema_name = ?
AND app_name = ?
""", params=[
app['database_name'],
app['schema_name'],
app['name']
]).collect()
# session_stateのいいね数を更新
st.session_state[likes_key] = current_likes + 1
# バックグラウンドでキャッシュをクリア
clear_app_cache()
# このボタンのいいね状態を更新
st.session_state.like_states[likes_key] = True
# いいねボタンの表示
if st.button(
f"👍 {current_likes}",
key=f"like_button_{likes_key}",
on_click=on_like_click,
disabled=st.session_state.like_states.get(likes_key, False)
):
pass
@st.fragment
def render_app_card_view(app, metadata, app_key):
"""
アプリ情報をカード形式でレンダリングする関数
"""
app_title = app['title'] if app['title'] is not None else app['name']
app_description = metadata['description'] if metadata['description'] is not None else "説明なし"
app_url = f"https://app.snowflake.com/{orgname}/{accountname}/#/streamlit-apps/{app['database_name']}.{app['schema_name']}.{app['name']}"
# いいねの状態管理用のキー
likes_key = f"likes_{app_key}"
if likes_key not in st.session_state:
st.session_state[likes_key] = metadata.get('likes', 0)
# カードのコンテナ
tile = st.container(height=160, border=True)
# タイトルと基本情報
col1, col2 = tile.columns([3, 1])
with col1:
# タイトルをリンクとして表示(さらに小さく)
st.markdown(f"###### [{app_title}]({app_url})")
# メトリクスとタグを横並びで表示
st.caption(f"👁 {metadata.get('views', 0)}")
if metadata.get('tags'):
tags = metadata['tags']
if isinstance(tags, str):
tags = [tag.strip() for tag in tags.split(',') if tag.strip()]
st.caption(" ".join([f"#{tag}" for tag in tags]))
with col2:
# いいねボタン
render_like_button(app, likes_key)
# 説明文を最後に表示(全文表示)
tile.caption(app_description)
@st.fragment
def generate_description_with_progress(model_name):
"""
説明文の生成処理とプログレス表示を行います。
"""
with st.spinner("説明文を生成中..."):
apps_processed = generate_description_for_all_apps(model_name)
if apps_processed > 0:
st.success("全てのアプリの説明文を生成しました。")
@st.fragment
def description_generation_form():
"""
説明文生成のフォームを表示します。
"""
with st.form("description_generation_form"):
models = [
"deepseek-r1",
"claude-4-sonnet", "claude-3-7-sonnet", "claude-3-5-sonnet",
"mistral-large2", "mixtral-8x7b", "mistral-7b",
"llama4-maverick",
"llama3.3-70b",
"llama3.2-1b", "llama3.2-3b",
"llama3.1-8b", "llama3.1-70b", "llama3.1-405b",
"snowflake-llama-3.1-405b", "snowflake-llama-3.3-70b",
"snowflake-arctic",
"reka-flash", "reka-core",
"jamba-instruct", "jamba-1.5-mini", "jamba-1.5-large",
"gemma-7b",
"mistral-large", "llama3-8b", "llama3-70b", "llama2-70b-chat"
]
selected_model = st.selectbox("使用するモデルを選択", models)
submitted = st.form_submit_button("説明を一括生成")
if submitted:
generate_description_with_progress(selected_model)
@st.fragment
def tag_generation_with_progress():
"""
タグの生成処理とプログレス表示を行います。
"""
with st.spinner("タグを生成中..."):
generate_tags_for_all_apps()
st.success("全てのアプリのタグを生成しました。")
@st.fragment
def category_list():
"""
カテゴリ一覧を表示し、選択したカテゴリを削除できるようにします。
"""
if "category_message" not in st.session_state:
st.session_state.category_message = None
# カテゴリ一覧を取得
categories = [
row['CATEGORY'] for row in
snowflake_session.sql("SELECT category FROM APP_CATEGORIES ORDER BY category").collect()
]
with st.form("category_list_form", clear_on_submit=True):
selected = st.selectbox(
"カテゴリを選択",
categories,
key=f"category_selector_{len(categories)}"
)
submitted = st.form_submit_button("選択したカテゴリを削除")
if submitted and selected:
snowflake_session.sql(
"DELETE FROM APP_CATEGORIES WHERE category = ?",
params=[selected]
).collect()
st.session_state.category_message = f"カテゴリ '{selected}' を削除しました。"
st.rerun()
# メッセージ表示用のプレースホルダー(フォームの下に移動)
if st.session_state.category_message:
st.success(st.session_state.category_message)
st.session_state.category_message = None
@st.fragment
def add_category_form():
"""
新しいカテゴリを追加するフォームを表示します。
"""
if "add_category_message" not in st.session_state:
st.session_state.add_category_message = None
with st.form("add_category_form", clear_on_submit=True):
new_category = st.text_input("新しいカテゴリ名")
submitted = st.form_submit_button("追加")
if submitted and new_category:
snowflake_session.sql(
"INSERT INTO APP_CATEGORIES (category) SELECT ? WHERE NOT EXISTS (SELECT 1 FROM APP_CATEGORIES WHERE category = ?)",
params=[new_category, new_category]
).collect()
st.session_state.add_category_message = f"カテゴリ '{new_category}' を追加しました。"
st.rerun()
# メッセージ表示用のプレースホルダー(フォームの下に移動)
if st.session_state.add_category_message:
st.success(st.session_state.add_category_message)
st.session_state.add_category_message = None
@st.fragment
def category_management():
"""
カテゴリの管理機能を提供します。
"""
st.write("### カテゴリの管理")
st.write("タグ生成に使用するカテゴリを管理します。")
col1, col2 = st.columns(2)
with col1:
st.write("#### 現在のカテゴリ一覧")
category_list()
with col2:
st.write("#### 新しいカテゴリの追加")
add_category_form()
def description_generation_page():
"""
説明文の自動生成ページを表示します。
"""
st.title("説明文の自動生成")
# 説明文の一括生成
st.write("### 説明文の一括生成")
st.write("全てのアプリの説明文を自動生成します。")
description_generation_form()
st.divider()
# 説明文の一括削除
st.write("### 説明文の一括削除")
st.write("全てのアプリの説明文を削除します。")
confirm_and_delete_descriptions()
@st.fragment
def tag_generation_form():
"""
タグ生成のフォームを表示します。
"""
if "tag_generation_message" not in st.session_state:
st.session_state.tag_generation_message = None
with st.form("tag_generation_form"):
st.write("全てのアプリの説明文から、設定されたカテゴリを使用してタグを自動生成します。")
submitted = st.form_submit_button("タグを一括生成")
if submitted:
with st.spinner("タグを生成中..."):
apps_processed = generate_tags_for_all_apps()
if apps_processed > 0:
st.success("全てのアプリのタグを生成しました。")
def tag_generation_page():
"""
タグの自動生成ページを表示します。
"""
st.title("タグの自動生成")
# カテゴリの管理
category_management()
st.divider()
# タグの一括生成
st.write("### タグの一括生成")
tag_generation_form()
st.divider()
# タグの一括削除
st.write("### タグの一括削除")
st.write("全てのアプリのタグを削除します。")
confirm_and_delete_tags()
def main_page():
"""
メインページの表示を行います。
"""
st.title("Streamlit in Snowflake アプリカタログ")
# キャッシュされたデータを取得
apps_df, app_descriptions = get_app_data()
# 表示方法の選択
display_mode = st.radio("表示方法", ["カード表示", "リスト表示"], horizontal=True)
# 検索とフィルター用のコントロール
search_query, sort_by, selected_tags, _ = admin_search_filter()
# フィルタリングされたアプリの表示
filtered_apps = filter_apps(apps_df, app_descriptions, search_query, selected_tags)
# 並び替え
if sort_by == "人気順":
filtered_apps.sort(key=lambda x: x[1].get('views', 0), reverse=True)
elif sort_by == "評価順":
filtered_apps.sort(key=lambda x: x[1].get('likes', 0), reverse=True)
# アプリ一覧の表示
if display_mode == "カード表示":
# カード表示の場合は3列のグリッドレイアウトを作成
cols = st.columns(3)
for i, (app, metadata) in enumerate(filtered_apps):
with cols[i % 3]:
render_app_card_view(app, metadata, f"{app['database_name']}.{app['schema_name']}.{app['name']}")
if search_query.strip() and metadata.get('similarity', 0.0) > 0:
st.caption(f"一致率: {metadata['similarity']:.1%}")
else:
# リスト表示
for app, metadata in filtered_apps:
app_title = app['title'] if app['title'] is not None else app['name']
app_url = f"https://app.snowflake.com/{orgname}/{accountname}/#/streamlit-apps/{app['database_name']}.{app['schema_name']}.{app['name']}"
app_key = f"{app['database_name']}.{app['schema_name']}.{app['name']}"
# いいねの状態管理用のキー
likes_key = f"likes_{app_key}"
if likes_key not in st.session_state:
st.session_state[likes_key] = metadata.get('likes', 0)
# アプリ情報のコンテナ
with st.container():
# タイトルと基本情報
col1, col2 = st.columns([3, 1])
with col1:
st.markdown(f"### [{app_title}]({app_url})")
if search_query.strip() and metadata.get('similarity', 0.0) > 0:
st.caption(f"一致率: {metadata['similarity']:.1%}")
with col2:
# メトリクス
st.caption(f"👁 {metadata.get('views', 0)}")
# いいねボタン
render_like_button(app, likes_key)
# タグの表示
if metadata.get('tags'):
tags = metadata['tags']
if isinstance(tags, str):
tags = [tag.strip() for tag in tags.split(',') if tag.strip()]
st.caption(" ".join([f"#{tag}" for tag in tags]))
# 詳細情報を折りたたみで表示
with st.expander("詳細情報"):
# 説明文
if metadata.get('description'):
st.markdown(metadata['description'])
else:
st.markdown("*説明なし*")
# 詳細情報のテーブル
details = {
"アプリ名": app['name'],
"データベース": app['database_name'],
"スキーマ": app['schema_name'],
"所有者": app.get('owner', '不明'),
"作成日時": str(app.get('created_on', '不明'))[:19] if app.get('created_on') else '不明',
"URL": app_url
}
st.table(pd.DataFrame([details]).T.rename(columns={0: "値"}))
st.divider()
@st.fragment
def render_app_metadata_editor(app, metadata, categories):
"""
アプリのメタデータ編集フォームを表示します。
"""
app_key = f"{app['database_name']}.{app['schema_name']}.{app['name']}"
app_title = app['title'] if app['title'] is not None else app['name']
st.subheader(f"{app_title} ({app['name']})")
with st.form(key=f"metadata_form_{app_key}"):
col1, col2 = st.columns([3, 1])
with col1:
new_desc = st.text_area("説明", metadata['description'] or "", key=f"desc_{app_key}")
selected_app_tags = st.multiselect(
"タグ",
categories,
default=metadata['tags'],
key=f"tags_{app_key}"
)
if metadata['last_updated']:
st.caption(f"最終更新: {metadata['last_updated']}")
with col2:
submitted = st.form_submit_button("更新")
if submitted:
update_app_description(app['database_name'], app['schema_name'], app['name'], new_desc, selected_app_tags)
st.success(f"{app_title} のメタデータが更新されました。")
st.divider()
def admin_search_filter():
"""
管理画面の検索とフィルター用のコントロールを表示します。
"""
with st.form("search_form"):
col1, col2 = st.columns([2, 1])
with col1:
search_query = st.text_input("アプリを検索", "")
with col2:
sort_by = st.selectbox("並び替え", ["人気順", "評価順"])
# タグ一覧の取得
all_tags = set()
app_data = snowflake_session.sql("""
SELECT DISTINCT TRIM(value) as tag
FROM APP_DESCRIPTIONS,
LATERAL FLATTEN(input => SPLIT(tags, ',')) as t
WHERE tags IS NOT NULL
ORDER BY tag
""").collect()
for row in app_data:
if row['TAG']:
all_tags.add(row['TAG'])
selected_tags = st.multiselect("タグでフィルター", sorted(list(all_tags)))
search_button = st.form_submit_button("検索")
return search_query, sort_by, selected_tags, search_button
@st.cache_data
def get_cached_categories():
"""
カテゴリ一覧をキャッシュして取得します。
"""
return [row['CATEGORY'] for row in snowflake_session.sql("SELECT category FROM APP_CATEGORIES ORDER BY category").collect()]
def filter_apps(apps_df, app_descriptions, search_query, selected_tags):
"""
アプリ一覧をフィルタリングします。
キーワード検索とベクトル検索を組み合わせたハイブリッド検索を行います。
"""
filtered_apps = []
if search_query.strip():
# 検索クエリのエスケープ処理
escaped_query = search_query.replace("'", "''")
# ベクトル検索の結果を取得
vector_results = snowflake_session.sql(f"""
WITH search_vector AS (
SELECT SNOWFLAKE.CORTEX.EMBED_TEXT_1024('voyage-multilingual-2', '{escaped_query}') as query_embedding
)
SELECT
a.database_name,
a.schema_name,
a.app_name,
VECTOR_COSINE_SIMILARITY(a.embedding, s.query_embedding) as similarity
FROM
APP_DESCRIPTIONS a,
search_vector s
WHERE
a.embedding IS NOT NULL
ORDER BY
similarity DESC
""").collect()
# ベクトル検索の結果をディクショナリに変換
similarity_dict = {
f"{row['DATABASE_NAME']}.{row['SCHEMA_NAME']}.{row['APP_NAME']}": row['SIMILARITY']
for row in vector_results
}
# アプリをフィルタリング
for _, app in apps_df.iterrows():
app_key = f"{app['database_name']}.{app['schema_name']}.{app['name']}"
metadata = app_descriptions.get(app_key, {'description': "", 'tags': [], 'last_updated': None})
app_title = app['title'] if app['title'] is not None else app['name']
# 検索とタグフィルタリングの条件
show_app = True
similarity = 0.0
if search_query.strip():
# キーワード検索
keyword_match = (
search_query.lower() in app_title.lower() or
search_query.lower() in (metadata['description'] or "").lower()
)
# ベクトル検索の類似度を取得
vector_similarity = similarity_dict.get(app_key, 0.0)
# キーワードマッチまたは類似度が0.4以上の場合に表示
show_app = keyword_match or vector_similarity >= 0.4
similarity = vector_similarity if vector_similarity >= 0.4 else 0.0
if selected_tags:
# タグが文字列の場合は配列に変換
app_tags = metadata['tags']
if isinstance(app_tags, str):
app_tags = [tag.strip() for tag in app_tags.split(',') if tag.strip()]
# 選択された全てのタグを含むかをチェック(AND条件)
show_app = show_app and all(tag in app_tags for tag in selected_tags)
if show_app:
metadata['similarity'] = similarity
filtered_apps.append((app, metadata))
return filtered_apps
def admin_page():
"""
管理画面: アプリのメタデータを手動で編集します。
"""
st.title("管理画面: アプリのメタデータ編集")
# キャッシュされたデータを取得
apps_df, app_descriptions = get_app_data()
categories = get_cached_categories()
# 検索とフィルター用のコントロール
search_query, sort_by, selected_tags, _ = admin_search_filter()
# フィルタリングされたアプリの表示
filtered_apps = filter_apps(apps_df, app_descriptions, search_query, selected_tags)
# 並び替え
if sort_by == "人気順":
filtered_apps.sort(key=lambda x: x[1].get('views', 0), reverse=True)
elif sort_by == "評価順":
filtered_apps.sort(key=lambda x: x[1].get('likes', 0), reverse=True)
# アプリ一覧の表示
for app, metadata in filtered_apps:
render_app_metadata_editor(app, metadata, categories)
def get_app_source_code(database, schema, app_name):
"""
アプリのソースコードを取得します。
"""
try:
# アプリの詳細情報を取得
app_details = snowflake_session.sql(f"""
DESCRIBE STREAMLIT {database}.{schema}.{app_name}
""").collect()
app_details_df = pd.DataFrame(app_details)
# root_locationを取得
root_location = app_details_df['root_location'].iloc[0]
if not root_location:
return None
# ステージのファイル一覧を取得
files_df = pd.DataFrame(snowflake_session.sql(f"LIST '{root_location}'").collect())
# .pyファイルを探す
py_files = files_df[files_df['name'].str.endswith('.py')]['name'].tolist()
if not py_files:
return None
# 最初の.pyファイルの内容を取得
file_path = py_files[0]
stage_parts = root_location.lstrip('@').split('.')
if len(stage_parts) >= 3:
database = stage_parts[0]
schema = stage_parts[1]
stage = '.'.join(stage_parts[2:])
# ステージ名が含まれている場合は除去
stage_suffix = stage.strip('"').lower()
clean_file_path = file_path
if file_path.lower().startswith(stage_suffix + '/'):
clean_file_path = file_path[len(stage_suffix + '/'):]
# 完全修飾名でステージパスを構築
stage_path = f"@{database}.{schema}.{stage}/{clean_file_path}"
# ファイルの内容を取得
with snowflake_session.file.get_stream(stage_path) as file_stream:
content = file_stream.read()
# バイナリデータを文字列に変換
try:
return content.decode('utf-8')
except UnicodeDecodeError:
try:
return content.decode('shift-jis')
except UnicodeDecodeError:
st.error("ファイルのエンコーディングを特定できません")
return None
return None
except Exception as e:
st.error(f"ソースコード取得中にエラーが発生しました: {str(e)}")
return None
def generate_description_for_all_apps(model_name):
"""
説明文が未設定のアプリの説明を自動生成します。
戻り値: 処理したアプリの数
"""
# 説明文が未設定のアプリを取得
apps_df = get_streamlit_apps_info()
# 現在の説明文を持つアプリを取得
existing_descriptions = snowflake_session.sql("""
SELECT
database_name || '.' || schema_name || '.' || app_name as app_key
FROM APP_DESCRIPTIONS
WHERE description IS NOT NULL
AND TRIM(description) != ''
""").collect()
# 既に説明文があるアプリのキーをセットに変換
existing_app_keys = {row['APP_KEY'] for row in existing_descriptions}
# 説明文が未設定のアプリをフィルタリング
apps_to_process = []
for _, app in apps_df.iterrows():
app_key = f"{app['database_name']}.{app['schema_name']}.{app['name']}"
if app_key not in existing_app_keys:
apps_to_process.append(app)
total_apps = len(apps_to_process)
if total_apps == 0:
st.info("全てのアプリに説明文が設定されています。")
return 0
progress_bar = st.progress(0)
status_text = st.empty()
for i, app in enumerate(apps_to_process):
progress = (i + 1) / total_apps
progress_bar.progress(progress)
status_text.text(f"処理中... {i + 1}/{total_apps} アプリ ({int(progress * 100)}%)")
app_title = app['title'] if app['title'] is not None else app['name']
# アプリのソースコードを取得
source_code = get_app_source_code(app['database_name'], app['schema_name'], app['name'])
# プロンプトの作成(ソースコードは別のパラメータとして渡す)
prompt_template = """以下のStreamlitアプリの説明文を生成してください。
- 応答は説明文のみを出力してください。
- "はい、承知しました"などの応答は含めないでください。
- マークダウン形式ではなく、プレーンテキストで出力してください。
- 説明文は100文字程度で簡潔にまとめてください。
アプリ名: {app_title} ({app_name})"""
try:
# プロンプトとソースコードを別々のパラメータとして渡す
prompt_params = {
'app_title': app_title,
'app_name': app['name'],
'source_code': source_code if source_code else '※ソースコードは取得できませんでした。'
}
# SQLインジェクションを防ぐためにバインドパラメータを使用
description = snowflake_session.sql("""
SELECT SNOWFLAKE.CORTEX.COMPLETE(?,
CONCAT(
?,
'\n\nソースコード:\n',
?
)
)
""", params=[
model_name,
prompt_template.format(app_title=app_title, app_name=app['name']),
prompt_params['source_code']
]).collect()[0][0]
# 説明文の更新
snowflake_session.sql("""
MERGE INTO APP_DESCRIPTIONS
USING (SELECT
? AS db,
? AS sc,
? AS app,
? AS desc
) AS src
ON APP_DESCRIPTIONS.database_name = src.db
AND APP_DESCRIPTIONS.schema_name = src.sc
AND APP_DESCRIPTIONS.app_name = src.app
WHEN MATCHED THEN
UPDATE SET
description = src.desc,
last_updated = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
INSERT (database_name, schema_name, app_name, description)
VALUES (src.db, src.sc, src.app, src.desc)
""", params=[
app['database_name'],
app['schema_name'],
app['name'],
description
]).collect()
except Exception as e:
st.error(f"アプリ '{app_title}' の処理中にエラーが発生しました: {str(e)}")
continue
# 処理完了後にキャッシュをクリア
clear_app_cache()
progress_bar.progress(1.0)
status_text.text(f"完了! {total_apps}個のアプリの説明を生成しました。")
return total_apps
@st.fragment
def confirm_and_delete_descriptions():
"""
説明の一括削除の確認と実行を行います。
"""
if "show_desc_confirm" not in st.session_state:
st.session_state.show_desc_confirm = False
if "desc_message" not in st.session_state:
st.session_state.desc_message = None
container = st.container()
with container:
button_placeholder = st.empty()
message_placeholder = st.empty()
confirm_placeholder = st.empty()
if button_placeholder.button("説明を一括削除", type="secondary") or st.session_state.show_desc_confirm:
st.session_state.show_desc_confirm = True
with confirm_placeholder.container():
st.warning("本当に全ての説明を削除しますか?", icon="⚠️")
col1, col2 = st.columns(2)
with col1:
if st.button("削除", key="confirm_delete_desc", type="primary"):
with st.spinner("説明文を削除中..."):
clear_all_descriptions()
st.session_state.desc_message = "success"
st.session_state.show_desc_confirm = False
with col2:
if st.button("キャンセル", key="cancel_delete_desc"):
st.session_state.desc_message = "cancel"
st.session_state.show_desc_confirm = False
if st.session_state.desc_message == "success":
message_placeholder.success("全てのアプリの説明文を削除しました。")
confirm_placeholder.empty()
st.session_state.desc_message = None
elif st.session_state.desc_message == "cancel":
message_placeholder.info("削除をキャンセルしました。")
confirm_placeholder.empty()
st.session_state.desc_message = None
def clear_all_descriptions():
"""
全てのアプリの説明を削除します。
"""
snowflake_session.sql("""
UPDATE APP_DESCRIPTIONS
SET description = NULL,
last_updated = CURRENT_TIMESTAMP()
""").collect()
# キャッシュをクリア
clear_app_cache()
def clear_all_tags():
"""
全てのアプリのタグを削除します。
"""
snowflake_session.sql("""
UPDATE APP_DESCRIPTIONS
SET tags = NULL,
last_updated = CURRENT_TIMESTAMP()
""").collect()
# キャッシュをクリア
clear_app_cache()
@st.fragment
def confirm_and_delete_tags():
"""
タグの一括削除の確認と実行を行います。
"""
if "show_tags_confirm" not in st.session_state:
st.session_state.show_tags_confirm = False
if "tags_message" not in st.session_state:
st.session_state.tags_message = None
container = st.container()
with container:
button_placeholder = st.empty()
message_placeholder = st.empty()
confirm_placeholder = st.empty()
if button_placeholder.button("タグを一括削除", type="secondary") or st.session_state.show_tags_confirm:
st.session_state.show_tags_confirm = True
with confirm_placeholder.container():
st.warning("本当に全てのタグを削除しますか?", icon="⚠️")
col1, col2 = st.columns(2)
with col1:
if st.button("削除", key="confirm_delete_tags", type="primary"):
with st.spinner("タグを削除中..."):
clear_all_tags()
st.session_state.tags_message = "success"
st.session_state.show_tags_confirm = False
with col2:
if st.button("キャンセル", key="cancel_delete_tags"):
st.session_state.tags_message = "cancel"
st.session_state.show_tags_confirm = False
if st.session_state.tags_message == "success":
message_placeholder.success("全てのアプリのタグを削除しました。")
confirm_placeholder.empty()
st.session_state.tags_message = None
elif st.session_state.tags_message == "cancel":
message_placeholder.info("削除をキャンセルしました。")
confirm_placeholder.empty()
st.session_state.tags_message = None
def generate_tags_for_all_apps():
"""
タグが未設定のアプリのタグを自動生成します。
戻り値: 処理したアプリの数
"""
# 説明文が設定されているアプリの数を確認
apps_with_description = snowflake_session.sql("""
SELECT COUNT(*) as count
FROM APP_DESCRIPTIONS
WHERE description IS NOT NULL
AND TRIM(description) != ''
""").collect()[0]['COUNT']
if apps_with_description == 0:
st.warning("タグを生成するには、先にアプリの説明文を設定してください。")
return 0
# タグが未設定のアプリを取得
app_data = snowflake_session.sql("""
SELECT
database_name,
schema_name,
app_name,
description
FROM APP_DESCRIPTIONS
WHERE (tags IS NULL OR TRIM(tags) = '') -- 空文字列のみをチェック
AND description IS NOT NULL
AND TRIM(description) != ''
""").collect()
total_apps = len(app_data)
if total_apps == 0:
st.info("説明文が設定されている全てのアプリにタグが設定されています。")
return 0
# カテゴリ一覧を取得
categories = [row['CATEGORY'] for row in snowflake_session.sql("SELECT category FROM APP_CATEGORIES ORDER BY category").collect()]
categories_json = json.dumps(categories, ensure_ascii=False)
progress_bar = st.progress(0)
status_text = st.empty()
for i, row in enumerate(app_data):
try:
progress = (i + 1) / total_apps
progress_bar.progress(progress)
status_text.text(f"処理中... {i + 1}/{total_apps} アプリ ({int(progress * 100)}%)")
# CLASSIFY_TEXT関数の結果を取得
result = snowflake_session.sql("""
SELECT SNOWFLAKE.CORTEX.CLASSIFY_TEXT(?, PARSE_JSON(?))
""", params=[row['DESCRIPTION'], categories_json]).collect()[0][0]
# 結果がJSON文字列の場合はパース
if isinstance(result, str):
result = json.loads(result)
# タグを抽出(単一のタグの場合とリストの場合の両方に対応)
if isinstance(result, list):
tags = [item['label'] for item in result if isinstance(item, dict) and 'label' in item]
elif isinstance(result, dict) and 'label' in result:
tags = [result['label']]
else:
tags = []
# タグを更新
if tags: # タグが空でない場合のみ更新
snowflake_session.sql("""
UPDATE APP_DESCRIPTIONS
SET
tags = ?,
last_updated = CURRENT_TIMESTAMP()
WHERE database_name = ?
AND schema_name = ?
AND app_name = ?
""", params=[
','.join(tags),
row['DATABASE_NAME'],
row['SCHEMA_NAME'],
row['APP_NAME']
]).collect()
except Exception as e:
st.error(f"アプリ '{row['APP_NAME']}' のタグ生成中にエラーが発生しました: {str(e)}")
continue
# 処理完了後にキャッシュをクリア
clear_app_cache()
progress_bar.progress(1.0)
status_text.text(f"完了! {total_apps}個のアプリのタグを生成しました。")
return total_apps
def vector_generation_page():
"""
ベクトルデータの自動生成ページを表示します。
"""
st.title("ベクトルデータの自動生成")
st.write("アプリの説明文からベクトルデータを生成します。")
# 説明文が設定されているアプリの数を確認
apps_with_description = snowflake_session.sql("""
SELECT COUNT(*) as count
FROM APP_DESCRIPTIONS
WHERE description IS NOT NULL
AND TRIM(description) != ''
""").collect()[0]['COUNT']
if apps_with_description == 0:
st.warning("ベクトルデータを生成するには、先にアプリの説明文を設定してください。")
return
# ベクトルデータが未生成のアプリを取得
apps_without_vector = snowflake_session.sql("""
SELECT
DATABASE_NAME,
SCHEMA_NAME,
APP_NAME,
DESCRIPTION
FROM APP_DESCRIPTIONS
WHERE DESCRIPTION IS NOT NULL
AND TRIM(DESCRIPTION) != ''
AND EMBEDDING IS NULL -- NULLのみをチェック
""").collect()
total_apps = len(apps_without_vector)
if total_apps == 0:
st.info("説明文が設定されている全てのアプリにベクトルデータが設定されています。")
else:
st.write(f"{total_apps}個のアプリにベクトルデータを生成できます。")
if st.button("ベクトルデータを生成"):
progress_bar = st.progress(0)
status_text = st.empty()
for i, row in enumerate(apps_without_vector):
try:
progress = (i + 1) / total_apps
progress_bar.progress(progress)
status_text.text(f"処理中... {i + 1}/{total_apps} アプリ ({int(progress * 100)}%)")
# ベクトルデータを生成して保存
snowflake_session.sql("""
UPDATE APP_DESCRIPTIONS
SET
EMBEDDING = SNOWFLAKE.CORTEX.EMBED_TEXT_1024('voyage-multilingual-2', ?),
last_updated = CURRENT_TIMESTAMP()
WHERE database_name = ?
AND schema_name = ?
AND app_name = ?
AND EMBEDDING IS NULL -- NULLのみをチェック
""", params=[
row['DESCRIPTION'],
row['DATABASE_NAME'],
row['SCHEMA_NAME'],
row['APP_NAME']
]).collect()
except Exception as e:
st.error(f"アプリ '{row['APP_NAME']}' のベクトルデータ生成中にエラーが発生しました: {str(e)}")
continue
# 処理完了後にキャッシュをクリア
clear_app_cache()
progress_bar.progress(1.0)
status_text.text(f"完了! {total_apps}個のアプリのベクトルデータを生成しました。")
st.success("全ての対象アプリのベクトルデータが生成されました!")
st.divider()
# ベクトルデータの全削除
st.write("### ベクトルデータの一括削除")
st.write("全てのアプリのベクトルデータを削除します。")
confirm_and_delete_vectors()
@st.fragment
def confirm_and_delete_vectors():
"""
ベクトルデータの一括削除の確認と実行を行います。
"""
if "show_vectors_confirm" not in st.session_state:
st.session_state.show_vectors_confirm = False
if "vectors_message" not in st.session_state:
st.session_state.vectors_message = None
container = st.container()
with container:
button_placeholder = st.empty()
message_placeholder = st.empty()
confirm_placeholder = st.empty()
if button_placeholder.button("ベクトルデータを一括削除", type="secondary") or st.session_state.show_vectors_confirm:
st.session_state.show_vectors_confirm = True
with confirm_placeholder.container():
st.warning("本当に全てのベクトルデータを削除しますか?", icon="⚠️")
col1, col2 = st.columns(2)
with col1:
if st.button("削除", key="confirm_delete_vectors", type="primary"):
with st.spinner("ベクトルデータを削除中..."):
clear_all_vectors()
st.session_state.vectors_message = "success"
st.session_state.show_vectors_confirm = False
with col2:
if st.button("キャンセル", key="cancel_delete_vectors"):
st.session_state.vectors_message = "cancel"
st.session_state.show_vectors_confirm = False
if st.session_state.vectors_message == "success":
message_placeholder.success("全てのアプリのベクトルデータを削除しました。")
confirm_placeholder.empty()
st.session_state.vectors_message = None
elif st.session_state.vectors_message == "cancel":
message_placeholder.info("削除をキャンセルしました。")
confirm_placeholder.empty()
st.session_state.vectors_message = None
def clear_all_vectors():
"""
全てのアプリのベクトルデータを削除します。
"""
snowflake_session.sql("""
UPDATE APP_DESCRIPTIONS
SET embedding = NULL,
last_updated = CURRENT_TIMESTAMP()
""").collect()
def main():
"""
アプリケーションのメイン関数です。
"""
# アプリ起動時に1回だけテーブルの初期化と更新を実行
create_app_descriptions_table()
page = st.sidebar.radio(
"ページ選択",
["アプリポータル", "管理画面", "説明文の自動生成", "タグの自動生成", "ベクトルデータの自動生成"]
)
if page == "アプリポータル":
main_page()
elif page == "管理画面":
admin_page()
elif page == "説明文の自動生成":
description_generation_page()
elif page == "タグの自動生成":
tag_generation_page()
elif page == "ベクトルデータの自動生成":
vector_generation_page()
if __name__ == "__main__":
main()
最後に
いかがでしたでしょうか?アプリカタログを利用することで、様々な角度からアプリケーションを検索したりフィルタすることができるようになります。特に Streamlit in Snowflake の用途としては、ビジネスユーザー向けに利用していただくことも多いため、アプリケーションを楽に探せることは、より多くの方々にデータを身近に感じていただけるためには重要ではないかと思います。
宣伝
SNOWFLAKE DISCOVER (第2弾) 開催決定!
2025/7/8-9の2日間「Snowflake Discover (第2弾)」が開催されます!Snowflake Summit の振り返りからハンズオンまで凝縮したコンテンツを Webinar 形式でお届けしますので是非ご視聴ください!私は2日目のハンズオン「Snowflake Cortex AI で実現する次世代の VoC (顧客の声) 分析」をデリバリします。最新の Cortex AISQL 関数を用いた実践的な顧客の声分析の手法を体得することができますので是非お気軽にご登録ください!
開催日時: 2025/7/8-9
形式: Webinar (オンライン)
参加費: 無料
SNOWFLAKE DISCOVER で登壇しました!
2025/4/24-25に開催されました Snowflake のエンジニア向け大規模ウェビナー『SNOWFLAKE DISCOVER』において『ゼロから始めるSnowflake:モダンなデータ&AIプラットフォームの構築』という一番最初のセッションで登壇しました。Snowflake の概要から最新状況まで可能な限り分かりやすく説明しておりますので是非キャッチアップにご活用いただければ嬉しいです!
以下リンクでご登録いただけるとオンデマンドですぐにご視聴いただくことが可能です。
生成AI Conf 様の Webinar で登壇しました!
『生成AI時代を支えるプラットフォーム』というテーマの Webinar で NVIDIA 様、古巣の AWS 様と共に Snowflake 社員としてデータ*AI をテーマに LTをしました!以下が動画アーカイブとなりますので是非ご視聴いただければ幸いです!
X で Snowflake の What's new の配信してます
X で Snowflake の What's new の更新情報を配信しておりますので、是非お気軽にフォローしていただければ嬉しいです。
日本語版
Snowflake の What's New Bot (日本語版)
English Version
Snowflake What's New Bot (English Version)
変更履歴
(20250204) 新規投稿
(20250228) Cortex LLM のモデル llama3.3-70b
deepseek-r1
を追加、LLM のモデルを全体的に並べ替え
(20250508) 宣伝欄の修正
(20250510) Cortex LLM のモデル claude-3-7-sonnet
を追加
(20250524) Cortex LLM のモデル claude-4-sonnet
llama4-maverick
を追加
(20250629) 宣伝欄の修正
Discussion