❄️
【Snowflake World Tour Tokyo】Rookies Camp Demoでみたチャットボットを作ってみよう！
こんにちは。Rookies Camp幹事のnaoki_yokozawa（横澤 直樹）です。
Xはこちらのアカウントで主にデータマネジメント関連の発信をしていますのでフォローいただけますと幸いです！
さて、2025年9月11(木)、12(金)に開催しているSnowflake World Tour Tokyoにて我らが
Rookies Campですが『SNOWFLAKE COMMUNITY HUB』から4枠いただき念願のオフライン開催いたしました。ご参加いただきました皆さん本当にありがとうございました。
参加前にこのブログを見ていただいている方は是非楽しみにしていてください。
今回は、4枠のうちの夕方の2枠を利用して開催する『Rookies Camp Demo』についてのブログとなります。デモではどんな動きになるのかをお見せしただけで実際にどうやって作るのかは解説しませんでしたので、こちらでご説明させていただきます。
はじめに
本日記載する内容はデモの参考元としている以下のQuick Startの手順について、更にわかりやすいように補足を加えてご説明をさせていただきます。
Streamlit in Snowflakeのコード詳細やSemantic Modelについてなどは10/31(金)に開催予定の第5回Rookies Campで詳しくお話させていただきます。
※当日のデモ内容はQuick Startから少しカスタマイズしているため、その点も
更に深く知りたいという方は是非以下URLから奮ってお申し込みください。
Cotex Analystとは？
Snowflakeが提供する機能で、大規模言語モデル(LLM)を活用して自然言語で投げかけた質問をSQLクエリに変換してくれるサービスです。いわゆるText to SQLというやつです。
チャットボットの活用に利用することなどが想定されます。
📓全体手順
全体の手順は以下のとおりです。
- 各種オブジェクト（データベース、スキーマ、ウェアハウス、ステージ、テーブル）の準備
- ファイルの内部ステージへのアップ及びデータロード
- Cortex Searchの作成
- Streamlit in Snowflakeの作成
処理の流れは以下のとおりです。チャットボット画面としてStreamlit in Snowflakeを利用します。
各種オブジェクトの準備
オブジェクト作成用SQLファイルのダウンロード
「2. Set up the Snowflake environment」にも記載されていますがGitからcreate_snowflake_objects.sqlをダウンロードします。
以下の赤枠のボタンを押下するとSQLファイルがダウンロード可能です。
create_snowflake_objects.sql
/*--
• Database, schema, warehouse, and stage creation
--*/
USE ROLE SECURITYADMIN;
CREATE ROLE cortex_user_role;
GRANT DATABASE ROLE SNOWFLAKE.CORTEX_USER TO ROLE cortex_user_role;
-- TODO: Replace <your_user> with your username
GRANT ROLE cortex_user_role TO USER <your_user>;
USE ROLE sysadmin;
-- Create demo database
CREATE OR REPLACE DATABASE cortex_analyst_demo;
-- Create schema
CREATE OR REPLACE SCHEMA cortex_analyst_demo.revenue_timeseries;
-- Create warehouse
CREATE OR REPLACE WAREHOUSE cortex_analyst_wh
WAREHOUSE_SIZE = 'large'
WAREHOUSE_TYPE = 'standard'
AUTO_SUSPEND = 60
AUTO_RESUME = TRUE
INITIALLY_SUSPENDED = TRUE
COMMENT = 'Warehouse for Cortex Analyst demo';
GRANT USAGE ON WAREHOUSE cortex_analyst_wh TO ROLE cortex_user_role;
GRANT OPERATE ON WAREHOUSE cortex_analyst_wh TO ROLE cortex_user_role;
GRANT OWNERSHIP ON SCHEMA cortex_analyst_demo.revenue_timeseries TO ROLE cortex_user_role;
GRANT OWNERSHIP ON DATABASE cortex_analyst_demo TO ROLE cortex_user_role;
USE ROLE cortex_user_role;
-- Use the created warehouse
USE WAREHOUSE cortex_analyst_wh;
USE DATABASE cortex_analyst_demo;
USE SCHEMA cortex_analyst_demo.revenue_timeseries;
-- Create stage for raw data
CREATE OR REPLACE STAGE raw_data DIRECTORY = (ENABLE = TRUE);
/*--
• Fact and Dimension Table Creation
--*/
-- Fact table: daily_revenue
CREATE OR REPLACE TABLE cortex_analyst_demo.revenue_timeseries.daily_revenue (
date DATE,
revenue FLOAT,
cogs FLOAT,
forecasted_revenue FLOAT,
product_id INT,
region_id INT
);
-- Dimension table: product_dim
CREATE OR REPLACE TABLE cortex_analyst_demo.revenue_timeseries.product_dim (
product_id INT,
product_line VARCHAR(16777216)
);
-- Dimension table: region_dim
CREATE OR REPLACE TABLE cortex_analyst_demo.revenue_timeseries.region_dim (
region_id INT,
sales_region VARCHAR(16777216),
state VARCHAR(16777216)
);
use role accountadmin;
-- create a Git API integration for Snowflake Labs
-- This integration allows access to GitHub repositories under Snowflake-Labs
-- It is used for accessing demo data and scripts from the Snowflake Labs GitHub organization
CREATE OR REPLACE API INTEGRATION snowflake_labs_git_integration
API_PROVIDER = git_https_api
API_ALLOWED_PREFIXES = ('https://github.com/Snowflake-Labs/')
ENABLED = TRUE;
USE ROLE cortex_user_role;
-- Create a schema for Git repositories
-- This schema will contain Git repositories for the Cortex Analyst demo
CREATE OR REPLACE SCHEMA cortex_analyst_demo.git_repos;
-- Create a Git repository for the Cortex Analyst demo
-- This repository contains scripts and data for the Cortex Analyst demo
CREATE OR REPLACE GIT REPOSITORY cortex_analyst_demo.git_repos.getting_started_with_cortex_analyst
API_INTEGRATION = snowflake_labs_git_integration
ORIGIN = 'https://github.com/Snowflake-Labs/sfguide-getting-started-with-cortex-analyst';
-- Fetch the latest content from the Git repository
ALTER GIT REPOSITORY cortex_analyst_demo.git_repos.getting_started_with_cortex_analyst FETCH;
ワークシートを作成してSQLを実行
ワークシートを選択。
SQLワークシートを選択。
先程のSQLをコピペしてワークシートに貼り付けて、<your_user>の部分を自分のユーザ名に書き換え。
※私はnaokiyokozawaに書き換えました
SQLを実行（赤丸部分を押下でもいいですし、Ctrl + Enterでも実行可能）
各種オブジェクトが作成されました
ファイルの内部ステージへのアップロード及びデータロード
ファイルをダウンロード
Gitリポジトリから以下のファイルをダウンロードします
ステージへのアップロード
以下の手順でステージにアップロードします。
- [カタログ] - [データベースエクスプローラ]を押下
- [CORTEX_ANALYST_DEMO] - [REVENUE_TIMESERIES] - [ステージ] - [RAW_DATA]を押下
- 右上の [+ファイル]を押下
- 先ほどダウンロードした4ファイルをドラッグしてアップロード
データロード用SQLファイルのダウンロード
load_data.sqlをダウンロード
load_data.sql
/*--
• looad data into tables
--*/
USE ROLE CORTEX_USER_ROLE;
USE DATABASE CORTEX_ANALYST_DEMO;
USE SCHEMA CORTEX_ANALYST_DEMO.REVENUE_TIMESERIES;
USE WAREHOUSE CORTEX_ANALYST_WH;
COPY INTO CORTEX_ANALYST_DEMO.REVENUE_TIMESERIES.DAILY_REVENUE
FROM @raw_data
FILES = ('daily_revenue.csv')
FILE_FORMAT = (
TYPE=CSV,
SKIP_HEADER=1,
FIELD_DELIMITER=',',
TRIM_SPACE=FALSE,
FIELD_OPTIONALLY_ENCLOSED_BY=NONE,
REPLACE_INVALID_CHARACTERS=TRUE,
DATE_FORMAT=AUTO,
TIME_FORMAT=AUTO,
TIMESTAMP_FORMAT=AUTO
EMPTY_FIELD_AS_NULL = FALSE
error_on_column_count_mismatch=false
)
ON_ERROR=CONTINUE
FORCE = TRUE ;
COPY INTO CORTEX_ANALYST_DEMO.REVENUE_TIMESERIES.PRODUCT_DIM
FROM @raw_data
FILES = ('product.csv')
FILE_FORMAT = (
TYPE=CSV,
SKIP_HEADER=1,
FIELD_DELIMITER=',',
TRIM_SPACE=FALSE,
FIELD_OPTIONALLY_ENCLOSED_BY=NONE,
REPLACE_INVALID_CHARACTERS=TRUE,
DATE_FORMAT=AUTO,
TIME_FORMAT=AUTO,
TIMESTAMP_FORMAT=AUTO
EMPTY_FIELD_AS_NULL = FALSE
error_on_column_count_mismatch=false
)
ON_ERROR=CONTINUE
FORCE = TRUE ;
COPY INTO CORTEX_ANALYST_DEMO.REVENUE_TIMESERIES.REGION_DIM
FROM @raw_data
FILES = ('region.csv')
FILE_FORMAT = (
TYPE=CSV,
SKIP_HEADER=1,
FIELD_DELIMITER=',',
TRIM_SPACE=FALSE,
FIELD_OPTIONALLY_ENCLOSED_BY=NONE,
REPLACE_INVALID_CHARACTERS=TRUE,
DATE_FORMAT=AUTO,
TIME_FORMAT=AUTO,
TIMESTAMP_FORMAT=AUTO
EMPTY_FIELD_AS_NULL = FALSE
error_on_column_count_mismatch=false
)
ON_ERROR=CONTINUE
FORCE = TRUE ;
ワークシートを作成してSQLを実行
先程も説明した手順でload_data.sqlを実行
投入データのER図
以下構成のデータが投入されました。
※このデータだと、日別収益と地域マスタ結合すると増幅してしまうのですが、元データがそういうデータなので一旦このまま進めます。
Cortex Searchの作成
Cortex Search作成用SQLファイルのダウンロード
cortex_search_create.sqlをダウンロード
cortex_search_create.sql
USE ROLE cortex_user_role;
USE DATABASE cortex_analyst_demo;
USE SCHEMA revenue_timeseries;
CREATE OR REPLACE CORTEX SEARCH SERVICE product_line_search_service
ON product_dimension
WAREHOUSE = cortex_analyst_wh
TARGET_LAG = '1 hour'
AS (
SELECT DISTINCT product_line AS product_dimension FROM product_dim
);
ワークシートを作成してSQLを実行
cortex_search_create.sqlを実行
Streamlit in Snowflakeの作成
Streamlit in Snowflake用Pythoコードのダウンロード
cortex_analyst_sis_demo_app.pyをダウンロード
cortex_analyst_sis_demo_app.py
"""
Cortex Analyst App
====================
This app allows users to interact with their data using natural language.
"""
import json # To handle JSON data
import time
from datetime import datetime
from typing import Dict, List, Optional, Tuple, Union
import _snowflake # For interacting with Snowflake-specific APIs
import pandas as pd
import streamlit as st # Streamlit library for building the web app
from snowflake.snowpark.context import (
get_active_session,
) # To interact with Snowflake sessions
from snowflake.snowpark.exceptions import SnowparkSQLException
# List of available semantic model paths in the format: <DATABASE>.<SCHEMA>.<STAGE>/<FILE-NAME>
# Each path points to a YAML file defining a semantic model
AVAILABLE_SEMANTIC_MODELS_PATHS = [
"CORTEX_ANALYST_DEMO.REVENUE_TIMESERIES.RAW_DATA/revenue_timeseries.yaml"
]
API_ENDPOINT = "/api/v2/cortex/analyst/message"
FEEDBACK_API_ENDPOINT = "/api/v2/cortex/analyst/feedback"
API_TIMEOUT = 50000 # in milliseconds
# Initialize a Snowpark session for executing queries
session = get_active_session()
def main():
# Initialize session state
if "messages" not in st.session_state:
reset_session_state()
show_header_and_sidebar()
if len(st.session_state.messages) == 0:
process_user_input("What questions can I ask?")
display_conversation()
handle_user_inputs()
handle_error_notifications()
display_warnings()
def reset_session_state():
"""Reset important session state elements."""
st.session_state.messages = [] # List to store conversation messages
st.session_state.active_suggestion = None # Currently selected suggestion
st.session_state.warnings = [] # List to store warnings
st.session_state.form_submitted = (
{}
) # Dictionary to store feedback submission for each request
def show_header_and_sidebar():
"""Display the header and sidebar of the app."""
# Set the title and introductory text of the app
st.title("Cortex Analyst")
st.markdown(
"Welcome to Cortex Analyst! Type your questions below to interact with your data. "
)
# Sidebar with a reset button
with st.sidebar:
st.selectbox(
"Selected semantic model:",
AVAILABLE_SEMANTIC_MODELS_PATHS,
format_func=lambda s: s.split("/")[-1],
key="selected_semantic_model_path",
on_change=reset_session_state,
)
st.divider()
# Center this button
_, btn_container, _ = st.columns([2, 6, 2])
if btn_container.button("Clear Chat History", use_container_width=True):
reset_session_state()
def handle_user_inputs():
"""Handle user inputs from the chat interface."""
# Handle chat input
user_input = st.chat_input("What is your question?")
if user_input:
process_user_input(user_input)
# Handle suggested question click
elif st.session_state.active_suggestion is not None:
suggestion = st.session_state.active_suggestion
st.session_state.active_suggestion = None
process_user_input(suggestion)
def handle_error_notifications():
if st.session_state.get("fire_API_error_notify"):
st.toast("An API error has occured!", icon="🚨")
st.session_state["fire_API_error_notify"] = False
def process_user_input(prompt: str):
"""
Process user input and update the conversation history.
Args:
prompt (str): The user's input.
"""
# Clear previous warnings at the start of a new request
st.session_state.warnings = []
# Create a new message, append to history and display imidiately
new_user_message = {
"role": "user",
"content": [{"type": "text", "text": prompt}],
}
st.session_state.messages.append(new_user_message)
with st.chat_message("user"):
user_msg_index = len(st.session_state.messages) - 1
display_message(new_user_message["content"], user_msg_index)
# Show progress indicator inside analyst chat message while waiting for response
with st.chat_message("analyst"):
with st.spinner("Waiting for Analyst's response..."):
time.sleep(1)
response, error_msg = get_analyst_response(st.session_state.messages)
if error_msg is None:
analyst_message = {
"role": "analyst",
"content": response["message"]["content"],
"request_id": response["request_id"],
}
else:
analyst_message = {
"role": "analyst",
"content": [{"type": "text", "text": error_msg}],
"request_id": response["request_id"],
}
st.session_state["fire_API_error_notify"] = True
if "warnings" in response:
st.session_state.warnings = response["warnings"]
st.session_state.messages.append(analyst_message)
st.rerun()
def display_warnings():
"""
Display warnings to the user.
"""
warnings = st.session_state.warnings
for warning in warnings:
st.warning(warning["message"], icon="⚠️")
def get_analyst_response(messages: List[Dict]) -> Tuple[Dict, Optional[str]]:
"""
Send chat history to the Cortex Analyst API and return the response.
Args:
messages (List[Dict]): The conversation history.
Returns:
Optional[Dict]: The response from the Cortex Analyst API.
"""
# Prepare the request body with the user's prompt
request_body = {
"messages": messages,
"semantic_model_file": f"@{st.session_state.selected_semantic_model_path}",
}
# Send a POST request to the Cortex Analyst API endpoint
# Adjusted to use positional arguments as per the API's requirement
resp = _snowflake.send_snow_api_request(
"POST", # method
API_ENDPOINT, # path
{}, # headers
{}, # params
request_body, # body
None, # request_guid
API_TIMEOUT, # timeout in milliseconds
)
# Content is a string with serialized JSON object
parsed_content = json.loads(resp["content"])
# Check if the response is successful
if resp["status"] < 400:
# Return the content of the response as a JSON object
return parsed_content, None
else:
# Craft readable error message
error_msg = f"""
🚨 An Analyst API error has occurred 🚨
* response code: `{resp['status']}`
* request-id: `{parsed_content['request_id']}`
* error code: `{parsed_content['error_code']}`
Message:
```
{parsed_content['message']}
```
"""
return parsed_content, error_msg
def display_conversation():
"""
Display the conversation history between the user and the assistant.
"""
for idx, message in enumerate(st.session_state.messages):
role = message["role"]
content = message["content"]
with st.chat_message(role):
if role == "analyst":
display_message(content, idx, message["request_id"])
else:
display_message(content, idx)
def display_message(
content: List[Dict[str, Union[str, Dict]]],
message_index: int,
request_id: Union[str, None] = None,
):
"""
Display a single message content.
Args:
content (List[Dict[str, str]]): The message content.
message_index (int): The index of the message.
"""
for item in content:
if item["type"] == "text":
st.markdown(item["text"])
elif item["type"] == "suggestions":
# Display suggestions as buttons
for suggestion_index, suggestion in enumerate(item["suggestions"]):
if st.button(
suggestion, key=f"suggestion_{message_index}_{suggestion_index}"
):
st.session_state.active_suggestion = suggestion
elif item["type"] == "sql":
# Display the SQL query and results
display_sql_query(
item["statement"], message_index, item["confidence"], request_id
)
else:
# Handle other content types if necessary
pass
@st.cache_data(show_spinner=False)
def get_query_exec_result(query: str) -> Tuple[Optional[pd.DataFrame], Optional[str]]:
"""
Execute the SQL query and convert the results to a pandas DataFrame.
Args:
query (str): The SQL query.
Returns:
Tuple[Optional[pd.DataFrame], Optional[str]]: The query results and the error message.
"""
global session
try:
df = session.sql(query).to_pandas()
return df, None
except SnowparkSQLException as e:
return None, str(e)
def display_sql_confidence(confidence: dict):
if confidence is None:
return
verified_query_used = confidence["verified_query_used"]
with st.popover(
"Verified Query Used",
help="The verified query from [Verified Query Repository](https://docs.snowflake.com/en/user-guide/snowflake-cortex/cortex-analyst/verified-query-repository), used to generate the SQL",
):
with st.container():
if verified_query_used is None:
st.text(
"There is no query from the Verified Query Repository used to generate this SQL answer"
)
return
st.text(f"Name: {verified_query_used['name']}")
st.text(f"Question: {verified_query_used['question']}")
st.text(f"Verified by: {verified_query_used['verified_by']}")
st.text(
f"Verified at: {datetime.fromtimestamp(verified_query_used['verified_at'])}"
)
st.text("SQL query:")
st.code(verified_query_used["sql"], language="sql", wrap_lines=True)
def display_sql_query(
sql: str, message_index: int, confidence: dict, request_id: Union[str, None] = None
):
"""
Executes the SQL query and displays the results in form of data frame and charts.
Args:
sql (str): The SQL query.
message_index (int): The index of the message.
confidence (dict): The confidence information of SQL query generation
request_id (str): Request id from user request
"""
# Display the SQL query
with st.expander("SQL Query", expanded=False):
st.code(sql, language="sql")
display_sql_confidence(confidence)
# Display the results of the SQL query
with st.expander("Results", expanded=True):
with st.spinner("Running SQL..."):
df, err_msg = get_query_exec_result(sql)
if df is None:
st.error(f"Could not execute generated SQL query. Error: {err_msg}")
elif df.empty:
st.write("Query returned no data")
else:
# Show query results in two tabs
data_tab, chart_tab = st.tabs(["Data 📄", "Chart 📉"])
with data_tab:
st.dataframe(df, use_container_width=True)
with chart_tab:
display_charts_tab(df, message_index)
if request_id:
display_feedback_section(request_id)
def display_charts_tab(df: pd.DataFrame, message_index: int) -> None:
"""
Display the charts tab.
Args:
df (pd.DataFrame): The query results.
message_index (int): The index of the message.
"""
# There should be at least 2 columns to draw charts
if len(df.columns) >= 2:
all_cols_set = set(df.columns)
col1, col2 = st.columns(2)
x_col = col1.selectbox(
"X axis", all_cols_set, key=f"x_col_select_{message_index}"
)
y_col = col2.selectbox(
"Y axis",
all_cols_set.difference({x_col}),
key=f"y_col_select_{message_index}",
)
chart_type = st.selectbox(
"Select chart type",
options=["Line Chart 📈", "Bar Chart 📊"],
key=f"chart_type_{message_index}",
)
if chart_type == "Line Chart 📈":
st.line_chart(df.set_index(x_col)[y_col])
elif chart_type == "Bar Chart 📊":
st.bar_chart(df.set_index(x_col)[y_col])
else:
st.write("At least 2 columns are required")
def display_feedback_section(request_id: str):
with st.popover("📝 Query Feedback"):
if request_id not in st.session_state.form_submitted:
with st.form(f"feedback_form_{request_id}", clear_on_submit=True):
positive = st.radio(
"Rate the generated SQL", options=["👍", "👎"], horizontal=True
)
positive = positive == "👍"
submit_disabled = (
request_id in st.session_state.form_submitted
and st.session_state.form_submitted[request_id]
)
feedback_message = st.text_input("Optional feedback message")
submitted = st.form_submit_button("Submit", disabled=submit_disabled)
if submitted:
err_msg = submit_feedback(request_id, positive, feedback_message)
st.session_state.form_submitted[request_id] = {"error": err_msg}
st.session_state.popover_open = False
st.rerun()
elif (
request_id in st.session_state.form_submitted
and st.session_state.form_submitted[request_id]["error"] is None
):
st.success("Feedback submitted", icon="✅")
else:
st.error(st.session_state.form_submitted[request_id]["error"])
def submit_feedback(
request_id: str, positive: bool, feedback_message: str
) -> Optional[str]:
request_body = {
"request_id": request_id,
"positive": positive,
"feedback_message": feedback_message,
}
resp = _snowflake.send_snow_api_request(
"POST", # method
FEEDBACK_API_ENDPOINT, # path
{}, # headers
{}, # params
request_body, # body
None, # request_guid
API_TIMEOUT, # timeout in milliseconds
)
if resp["status"] == 200:
return None
parsed_content = json.loads(resp["content"])
# Craft readable error message
err_msg = f"""
🚨 An Analyst API error has occurred 🚨
* response code: `{resp['status']}`
* request-id: `{parsed_content['request_id']}`
* error code: `{parsed_content['error_code']}`
Message:
```
{parsed_content['message']}
```
"""
return err_msg
if __name__ == "__main__":
main()
- ロールを「CORTEX_USER_ROLE」に切替
- [プロジェクト] - [Streamlit]を押下
- [+ Streamlitアプリ]を押下
- 以下を指定して[作成]を押下
・ アプリタイトル：任意の名称
・ データベース：CORTEX_ANALYST_DEMO
・ スキーマ：REVENUE_TIMESERIES
・ ウェアハウス：CORTEX_ANALYST_WH - デフォルトのコードを削除したうえでダウンロードしたPythonコードをコピペする
- 右上の[▶実行]を押下
以下の画面が完成
さいごに
手順に則って実施いただければ、簡単にチャットボットを作ることができたのではないでしょうか。今回の内容を応用していただければ自社データで活用することも可能です。
ただ、この中で触れたStreamlit in SnowflakeやSemantic Modelについては詳細説明ができていないため、繰り返しになりますが是非第5回Rookies Campをお申し込みください！
長文でしたが、お付き合いいただきありがとうございました！！
Discussion