❄️
Streamlit in Snowflake (SiS) で自然言語によるデータ分析をしてみる
はじめに
Snowflake では Snowflake Copilot という LLM アシスタント機能がプレビュー機能として公開されています。この Snowflake Copilot を用いることで自然言語を用いてテーブルデータの分析を行うことが可能です。
一方で Streamlit in Snowflake (SiS) では簡単に生成 AI を組み込んだり、テーブルデータにセキュアにアクセスすることができるため、もっとお節介な自然言語データ分析ができるのではないか？と思い、自然言語で分析したり可視化したりできるアプリを作ってみました。
機能概要
実現したいこと
- SQL に馴染みの無いビジネスユーザーが利用できる
- DB / スキーマ / テーブルをマウス操作で選んで分析できる
- かなり曖昧な自然言語でも分析できる
実際の画面 (抜粋)
以下は Cortex LLM (snowflake-arctic) を用いて分析してみた結果の抜粋です。
※ こちらのワインの品質に関するデータセットを利用させていただきました
機能一覧
- 分析に使用する Cortex LLM を選択
- 分析対象となる DB / スキーマ / テーブル を選択
- 選択したテーブルのカラム情報とサンプルデータの表示
- ユーザーが入力した自然言語による分析
- 分析用 SQL の自動生成
- 生成された SQL の自動実行
- SQL クエリの実行結果の表示
- Cortex LLM によるデータの洞察
- データの可視化
- Cortex LLM による適切なグラフの選定と表示
- Cortex LLM による最終的なデータの解釈
動作条件
- Cortex LLM が利用できる Snowflake アカウント
- snowflake-ml-python 1.1.2 以降
注意事項
- データの Limit を設けていないため巨大なデータセットをご利用の際はご注意ください
- 生成された SQL クエリの結果サイズが大きいと処理に時間がかかることが想定されます
- 生成された SQL クエリの結果サイズが大きいと LLM のトークン数の上限値エラーが出ることが想定されます
ソースコード
from snowflake.snowpark.context import get_active_session
import streamlit as st
from snowflake.cortex import Complete as CompleteText
import snowflake.snowpark.functions as F
import pandas as pd
import numpy as np
import json
import plotly.express as px
# 現在のセッションの取得
session = get_active_session()
# アプリケーションのタイトル
st.title("自然言語によるデータ分析アプリ")
# Cortex LLMの設定
st.sidebar.title("Cortex LLM の設定")
lang_model = st.sidebar.radio("使用したい言語モデルを選んでください",
("snowflake-arctic", "reka-flash", "reka-core",
"mistral-large2", "mistral-large", "mixtral-8x7b", "mistral-7b",
"llama3.1-405b", "llama3.1-70b", "llama3.1-8b",
"llama3-70b", "llama3-8b", "llama2-70b-chat",
"jamba-instruct", "gemma-7b")
)
# カラム名をエスケープする関数
def escape_column_name(name):
return f'"{name}"'
# テーブル情報を取得する関数
def get_table_info(database, schema, table):
# テーブルの列情報を取得
columns = session.sql(f"DESCRIBE TABLE {database}.{schema}.{table}").collect()
# DataFrameを作成
column_df = pd.DataFrame(columns)
# テーブルの行数を取得
row_count = session.sql(f"SELECT COUNT(*) as count FROM {database}.{schema}.{table}").collect()[0]['COUNT']
# サンプルデータを取得
sample_data = session.sql(f"SELECT * FROM {database}.{schema}.{table} LIMIT 5").collect()
sample_df = pd.DataFrame(sample_data)
return column_df, row_count, sample_df
# データ分析関数
def analyze(df, query):
st.subheader("結果の分析")
# 基本的な統計情報の表示
st.subheader("基本統計情報")
st.write(df.describe())
# AIを使用してデータ分析
analysis_prompt = f"""
以下のデータフレームと元の質問に基づいて、データの分析を行ってください。
データから得られる洞察、傾向、特異点などを日本語で簡潔に説明してください。
可能であれば、以下の点についても言及してください：
1. データの分布や特徴
2. 異常値や外れ値の有無
3. カラム間の相関関係 （もし複数の数値カラムがある場合）
4. 時系列的な傾向 （もし日付や時間のデータがある場合）
5. カテゴリ別の特徴 （もしカテゴリで分けられるデータがある場合）
データフレーム:
{df.to_string()}
元の質問:
{query}
"""
analysis = CompleteText(lang_model, analysis_prompt)
st.write(analysis)
# データ可視化関数
def smart_data_visualization(df):
st.subheader("スマートデータ可視化")
if df.empty:
st.warning("データフレームが空です。可視化するデータがありません。")
return
# AIにグラフの提案を依頼
columns_info = "\n".join([f"{col} - type: {df[col].dtype}" for col in df.columns])
sample_data = df.head().to_string()
visualization_prompt = f"""
以下のデータフレームの情報を分析し、最も適切なグラフの種類と、そのx軸とy軸に使用すべきカラムを提案してください。
データの特性を考慮し、意味のある可視化となるよう選択してください。
カラムの情報:
{columns_info}
サンプルデータ:
{sample_data}
回答は以下の形式のJSONデータのみを提供してください:
{{
"graph_type": "散布図、棒グラフ、折れ線グラフ、ヒストグラム、箱ひげ図 のいずれか",
"x_axis": "x軸に使用するカラム名",
"y_axis": "y軸に使用するカラム名（該当する場合）",
"explanation": "選択理由の簡単な説明"
}}
"""
ai_suggestion = CompleteText(lang_model, visualization_prompt)
try:
suggestion = json.loads(ai_suggestion)
graph_type = suggestion['graph_type']
x_axis = suggestion['x_axis']
y_axis = suggestion.get('y_axis') # y軸が不要な場合もある
explanation = suggestion['explanation']
st.write(f"AI推奨のグラフ: {graph_type}")
st.write(f"選択理由: {explanation}")
if graph_type == "散布図":
fig = px.scatter(df, x=x_axis, y=y_axis, title=f"{x_axis} vs {y_axis}")
elif graph_type == "棒グラフ":
fig = px.bar(df, x=x_axis, y=y_axis, title=f"{y_axis} by {x_axis}")
elif graph_type == "折れ線グラフ":
fig = px.line(df, x=x_axis, y=y_axis, title=f"{y_axis} over {x_axis}")
elif graph_type == "ヒストグラム":
fig = px.histogram(df, x=x_axis, title=f"Distribution of {x_axis}")
elif graph_type == "箱ひげ図":
fig = px.box(df, x=x_axis, y=y_axis, title=f"Distribution of {y_axis} by {x_axis}")
else:
st.warning(f"未対応のグラフタイプ: {graph_type}")
return
st.plotly_chart(fig)
except json.JSONDecodeError:
st.error("AIの提案を解析できませんでした。もう一度試してください。")
except KeyError as e:
st.error(f"AIの提案に必要な情報が含まれていません: {str(e)}")
except Exception as e:
st.error(f"グラフの作成中にエラーが発生しました: {str(e)}")
# AI による可視化の解釈
visualization_interpretation_prompt = f"""
以下のデータセットと作成されたグラフに基づいて、データの傾向や特徴について詳細な解釈を日本語で提供してください。
考えられる洞察、パターン、異常値、または追加の分析が必要な領域を指摘してください。
データセットの情報:
カラム: {', '.join(df.columns)}
行数: {len(df)}
作成されたグラフ:
種類: {graph_type}
X軸: {x_axis}
Y軸: {y_axis if y_axis else 'なし'}
"""
ai_interpretation = CompleteText(lang_model, visualization_interpretation_prompt)
st.subheader("AI によるグラフの解釈")
st.write(ai_interpretation)
# AIの応答がSQLクエリだけじゃない場合の対処関数
def clean_sql_query(query):
# 先頭と末尾の空白を削除
query = query.strip()
# SQL キーワードで始まっていない場合、最初のSELECTまでを削除
if not query.upper().startswith(('SELECT')):
keywords = ['SELECT']
for keyword in keywords:
if keyword in query.upper():
query = query[query.upper().index(keyword):]
break
return query
# 自然言語を用いたDBへのクエリ関数
def data_analysis_and_natural_language_query():
# データベース選択
databases = session.sql("SHOW DATABASES").collect()
database_names = [row['name'] for row in databases]
selected_database = st.selectbox("データベースを選択してください", database_names)
if selected_database:
# スキーマ選択
schemas = session.sql(f"SHOW SCHEMAS IN DATABASE {selected_database}").collect()
schema_names = [row['name'] for row in schemas]
selected_schema = st.selectbox("スキーマを選択してください", schema_names)
if selected_schema:
# テーブル選択
tables = session.sql(f"SHOW TABLES IN {selected_database}.{selected_schema}").collect()
table_names = [row['name'] for row in tables]
selected_table = st.selectbox("テーブルを選択してください", table_names)
if selected_table:
# テーブル情報の取得
column_df, row_count, sample_df = get_table_info(selected_database, selected_schema, selected_table)
st.subheader("テーブル情報")
st.write(f"テーブル名: `{selected_database}.{selected_schema}.{selected_table}`")
st.write(f"総行数: **{row_count:,}**")
st.subheader("カラム情報")
st.dataframe(column_df)
st.subheader("サンプルデータ (5件のみ表示)")
st.dataframe(sample_df)
# テーブル情報の文字列化（AI用）
table_info = f"""
テーブル名: {selected_database}.{selected_schema}.{selected_table}
総行数: {row_count}
カラム情報:
{column_df.to_string(index=False)}
サンプルデータ:
{sample_df.to_string(index=False)}
"""
# ユーザーからの自然言語入力
user_query = st.text_area("選択したテーブルに関して知りたいことを自由に入力してください")
if st.button("分析開始"):
if user_query:
# AIを使用してSQLを生成
prompt = f"""
以下のテーブル情報と質問に基づいて、適切なSQLクエリを生成してください。
返事などは不要で、生成されたSQLクエリのみを返してください。
テーブル情報:
{table_info}
質問: {user_query}
注意:
- SQLはSnowflakeのSQL構文に従ってください。
- 集計関数を使って、クエリ結果が大量に出ないように調整してください。
- テーブル名は {selected_database}.{selected_schema}.{selected_table} を使用してください。
"""
generated_sql = CompleteText(lang_model, prompt)
generated_sql = clean_sql_query(generated_sql)
st.subheader("生成されたSQL:")
st.code(generated_sql, language='sql')
try:
# 生成されたSQLを実行
result = session.sql(generated_sql).collect()
df = pd.DataFrame(result)
st.subheader("クエリ結果:")
st.dataframe(df)
# 結果の分析
analyze(df, user_query)
# スマートデータ可視化
smart_data_visualization(df)
except Exception as e:
st.error(f"クエリの実行中にエラーが発生しました: {str(e)}")
else:
st.warning("質問を入力してください。")
# 実行部分
data_analysis_and_natural_language_query()
最後に
SQL や Python を使わないビジネスユーザーにとって自然言語で手軽にテーブルデータを分析できると企業におけるデータ活用の幅が広がるのではないでしょうか？また、複数テーブルを結合できるようにしたり、可視化のグラフの種類を増やしたりするなどで、より複雑な分析も行えるようになります。是非皆さまも Streamlit in Snowflake でデータ分析の民主化を実現してみてください。
宣伝
SNOWFLAKE WORLD TOUR TOKYO の参加登録受付中 (残席僅かです！)
Snowflake の最新情報を知ることができる大規模イベント『SNOWFLAKE WORLD TOUR TOKYO』が2024/9/12(木)@ANAインターコンチネンタル東京で開催されます！
残席が少なくなってきておりますので参加登録お急ぎください！
私も Snowflake の概要をお伝えするセッションで登壇する予定です！超分かりやすく Snowflake AI Data Cloud についてご紹介しますので、是非ご来場ください〜！
X で Snowflake の What's new の配信してます
X で Snowflake の What's new の更新情報を配信しておりますので、是非お気軽にフォローしていただければ嬉しいです。
日本語版
Snowflake の What's New Bot (日本語版)
English Version
Snowflake What's New Bot (English Version)
変更履歴
(20240710) 新規投稿
(20240901) 全体的に体裁修正、宣伝欄追加、ソースコードに mistral-large2 などを追加
Discussion