はじめに
クライアント案件で Streamlit in Snowflake（SIS）を使う流れが出てきたので、キャッチアップを兼ねて簡単なアプリを作ってみようと思いました。
以前から社内の振り返りを仕組み化したいと思っていたので、「AIフィードバック機能付きKPT振り返りアプリ」を題材にしてみました。
※KPT：Keep（続けたいこと）/ Problem（問題・課題）/ Try（次に取り組むこと）の3つの観点で振り返りをする手法
この記事の想定読者はこんな人です。
- Snowflakeは触ったことがある
- Streamlit in Snowflakeはまだ触っていない
- SIS + Cortexで、簡単なAIアプリを作ってみたい
ゴールは以下の通りです。
- Streamlit in Snowflake × Cortexで、簡単なAIアプリが作れるようになる
（ガチガチのプロダクトを作る話というより、サクッと動くものを簡単に作るイメージです）
アプリの概要
画面の流れはシンプルです。
- 今日のKPT（Keep / Problem / Try）を入力
- 「保存してAIフィードバックを受け取る」ボタンを押す
- AI(Snowflake Cortex)によるフィードバックを画面に表示
- 裏側でSnowflakeのテーブルに保存
Cortexでフィードバック生成
今回は、Cortex関数のSNOWFLAKE.CORTEX.AI_COMPLETEから openai-gpt-5-chat を使って、KPTに対するフィードバックを生成しました。
snowflake-arcticなど、もう少し安いモデルも試したかったのですが、日本語の出力がいまひとつでした。
（すべてを試したわけではないので、もっとコスパの良い選択肢もあるはず。）
料金は以下に記載されています。
https://www.snowflake.com/legal-files/CreditConsumptionTable.pdf
後述しますが、2026年1月現在の東京リージョンだと使えるモデルが少なめなので、クロスリージョン推論を可能にするためのパラメータ変更も必要です。
フィードバックのプロンプト
入力されたKPTに対して、前向きで具体的なフィードバックが返るようにしました。
(あくまでも社内アプリなので思い切ってギャル風にしてみましたが、どんなときも絵文字もりもりで返してくれて助かる。が、最近飽きてきた。)
以下のような関数を作って、KPT入力後に呼び出すような作りです。
def generate_feedback(keep_text: str, problem_text: str, try_text: str):
"""Cortex を呼び出してフィードバックを生成する。"""
prompt = f"""
あなたはユーザーに親身に寄り添うメンターです。
とにかく明るいギャル風の性格で、芯を突く発言が評判です。(絵文字などもたくさん使います。)
以下のKPT（Keep/Problem/Try）に対して、前向きで具体的なフィードバックを200〜500文字で提供してください。
```
Keep（良かったこと）:
{keep_text}
Problem（課題）:
{problem_text}
Try（次に試すこと）:
{try_text}
```
"""
sql = "SELECT TO_VARCHAR(SNOWFLAKE.CORTEX.AI_COMPLETE(?, ?)) as FEEDBACK"
result = session.sql(sql, params=[MODEL_NAME, prompt]).collect()
return result[0]["FEEDBACK"], MODEL_NAME
東京リージョンで必要になった設定：クロスリージョン推論
先述の通り、2026年1月時点だと東京リージョンで使えるモデルの選択肢が少なかったため、クロスリージョン推論 を有効化して、他のリージョンのモデルも使えるようにしました。
作ったもの
ざっと以下のような1つのテーブルとstreamlit_app.pyがあれば、最低限でKPT入力/AIフィードバック/過去分の閲覧ができました。
（絶対にもっと良い書き方はあると思うが、サクッと作りたかっただけだし！）
1) KPT保存用のテーブル
create or replace TABLE ST_APP_DB.KPT.KPT_ENTRY_ZENN (
ENTRY_ID VARCHAR(16777216) NOT NULL DEFAULT UUID_STRING(),
USER_EMAIL VARCHAR(16777216) NOT NULL,
ENTRY_DATE DATE NOT NULL,
KEEP_TEXT VARCHAR(16777216) NOT NULL,
PROBLEM_TEXT VARCHAR(16777216) NOT NULL,
TRY_TEXT VARCHAR(16777216) NOT NULL,
CREATED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP(),
UPDATED_AT TIMESTAMP_NTZ(9),
FEEDBACK_TEXT VARCHAR(16777216),
FEEDBACK_MODEL VARCHAR(16777216),
unique (USER_EMAIL, ENTRY_DATE),
primary key (ENTRY_ID)
)
COMMENT='このテーブルには、KPT法に則ったユーザーの振り返りテキストやAIによるフィードバックコメントが含まれています。レコードはユーザー/日付の単位で格納されています。';
※ SnowflakeだとUNIQUE制約は強制力がないので、unique (USER_EMAIL, ENTRY_DATE) は「意図のメモ」くらいの扱い。
2) streamlit_app.py（1ファイル）
そのまま書くと縦に長くなるので、適度にファイルの分割をしたほうが良いのは間違いない。
import streamlit as st
from snowflake.snowpark.context import get_active_session
from datetime import date
# ===== 設定 =====
KPT_TABLE = "ST_APP_DB.KPT.KPT_ENTRY_ZENN"
MODEL_NAME = "openai-gpt-5-chat"
session = get_active_session()
# ===== AI関連機能 =====
def generate_feedback(keep_text: str, problem_text: str, try_text: str):
"""Cortex を呼び出してフィードバックを生成する。"""
prompt = f"""
あなたはユーザーに親身に寄り添うメンターです。
とにかく明るいギャル風の性格で、芯を突く発言が評判です。(絵文字などもたくさん使います。)
以下のKPT（Keep/Problem/Try）に対して、前向きで具体的なフィードバックを200〜500文字で提供してください。
```
Keep（良かったこと）:
{keep_text}
Problem（課題）:
{problem_text}
Try（次に試すこと）:
{try_text}
```
"""
sql = "SELECT TO_VARCHAR(SNOWFLAKE.CORTEX.AI_COMPLETE(?, ?)) as FEEDBACK"
result = session.sql(sql, params=[MODEL_NAME, prompt]).collect()
return result[0]["FEEDBACK"], MODEL_NAME
# ===== データベース操作 =====
def save_kpt_entry(
user_email: str,
entry_date: date,
keep_text: str,
problem_text: str,
try_text: str,
feedback_text: str,
feedback_model: str,
) -> bool:
"""KPT エントリを Snowflake に MERGE で保存する（同日分は更新）。"""
try:
merge_sql = f"""
MERGE INTO {KPT_TABLE} AS target
USING (
SELECT
? as USER_EMAIL,
? as ENTRY_DATE,
? as KEEP_TEXT,
? as PROBLEM_TEXT,
? as TRY_TEXT,
? as FEEDBACK_TEXT,
? as FEEDBACK_MODEL
) AS source
ON target.USER_EMAIL = source.USER_EMAIL
AND target.ENTRY_DATE = source.ENTRY_DATE
WHEN MATCHED THEN
UPDATE SET
KEEP_TEXT = source.KEEP_TEXT,
PROBLEM_TEXT = source.PROBLEM_TEXT,
TRY_TEXT = source.TRY_TEXT,
UPDATED_AT = CURRENT_TIMESTAMP(),
FEEDBACK_TEXT = source.FEEDBACK_TEXT,
FEEDBACK_MODEL = source.FEEDBACK_MODEL
WHEN NOT MATCHED THEN
INSERT (
USER_EMAIL, ENTRY_DATE, KEEP_TEXT, PROBLEM_TEXT, TRY_TEXT,
FEEDBACK_TEXT, FEEDBACK_MODEL
)
VALUES (
source.USER_EMAIL, source.ENTRY_DATE, source.KEEP_TEXT,
source.PROBLEM_TEXT, source.TRY_TEXT,
source.FEEDBACK_TEXT, source.FEEDBACK_MODEL
)
"""
session.sql(
merge_sql,
params=[
user_email, entry_date, keep_text, problem_text, try_text,
feedback_text, feedback_model
],
).collect()
return True
except Exception as e:
st.error(f"保存エラー: {e}")
return False
# ===== UI =====
st.title("📝 KPT記録アプリ")
st.markdown("Keep / Problem / Tryを記録して、AIフィードバックを受け取ろう！")
tab1, tab2 = st.tabs(["✏️ KPT入力", "📊 履歴"])
with tab1:
col1, col2 = st.columns([2, 1])
with col1:
user_email = st.text_input("メールアドレス",value=st.user.email,disabled=True)
with col2:
entry_date = st.date_input("日付", value=date.today())
with st.form("kpt_form"):
keep_text = st.text_area("📗 Keep（続けたいこと） *")
problem_text = st.text_area("📕 Problem（問題・課題） *")
try_text = st.text_area("📘 Try（次に試すこと） *")
submitted = st.form_submit_button("💾 保存してAIフィードバックを受け取る")
if submitted:
if not keep_text or not problem_text or not try_text:
st.warning("⚠️ 全ての KPT 項目を入力してください")
st.stop()
with st.spinner("🤖 AIがフィードバックを生成しています..."):
try:
feedback_text, feedback_model = generate_feedback(keep_text, problem_text, try_text)
except Exception as e:
st.error(f"AI生成エラー: {e}")
st.stop()
success = save_kpt_entry(
user_email=user_email,
entry_date=entry_date,
keep_text=keep_text,
problem_text=problem_text,
try_text=try_text,
feedback_text=feedback_text,
feedback_model=feedback_model
)
if success:
st.success("✅ 保存が完了しました！")
st.markdown("### 🤖 AIフィードバック")
st.info(feedback_text)
with tab2:
st.markdown("### 📊 履歴")
st.caption("日付ごとにメールアドレスを表示します。メールをクリックすると、その日付ブロックの直下に詳細を表示します。")
# ------------------------------
# 日付範囲（デフォルト：今日を含む直近7日）
# ------------------------------
today = date.today()
default_from = date.fromordinal(today.toordinal() - 6)
col1, col2 = st.columns([1, 1])
with col1:
date_from = st.date_input("開始日", value=default_from, key="history_date_from")
with col2:
date_to = st.date_input("終了日", value=today, key="history_date_to")
if date_from > date_to:
st.warning("⚠️ 開始日は終了日以前にしてください。")
st.stop()
# ------------------------------
# 一覧取得（ENTRY_DATE×USER_EMAIL）
# ------------------------------
try:
list_sql = f"""
SELECT
ENTRY_DATE,
USER_EMAIL
FROM {KPT_TABLE}
WHERE ENTRY_DATE BETWEEN ? AND ?
GROUP BY ENTRY_DATE, USER_EMAIL
ORDER BY ENTRY_DATE DESC, USER_EMAIL ASC
"""
rows = session.sql(list_sql, params=[date_from, date_to]).collect()
except Exception as e:
st.error(f"履歴取得エラー: {e}")
rows = []
if not rows:
st.info("📭 指定した期間の履歴がありません。")
st.stop()
# ------------------------------
# 日付ごとにまとめる（メール昇順）
# ------------------------------
by_date = {}
for r in rows:
d = r["ENTRY_DATE"]
email = r["USER_EMAIL"]
by_date.setdefault(d, []).append(email)
for d in by_date:
by_date[d] = sorted(list(set(by_date[d])))
# ------------------------------
# 選択状態（クリックした日付ブロック直下に表示するため）
# ------------------------------
if "history_selected_date" not in st.session_state:
st.session_state["history_selected_date"] = None
if "history_selected_email" not in st.session_state:
st.session_state["history_selected_email"] = None
# ------------------------------
# UI：日付縦並び → その下にメールボタン
# ------------------------------
for d in sorted(by_date.keys(), reverse=True):
st.markdown("---")
st.subheader(f"📅 {d}")
emails = by_date[d]
if not emails:
st.write("（この日の投稿はありません）")
continue
# ボタンを縦に並べる
for i, email in enumerate(emails):
btn_key = f"history_btn_{d}_{email}"
if st.button(email, key=btn_key):
st.session_state["history_selected_date"] = d
st.session_state["history_selected_email"] = email
# ------------------------------
# 選択された場合は「この日付ブロック直下」に詳細表示
# ------------------------------
selected_date = st.session_state.get("history_selected_date")
selected_email = st.session_state.get("history_selected_email")
if selected_date == d and selected_email:
with st.container(border=True):
st.markdown(f"#### 🧾 詳細 | {selected_email}")
try:
detail_sql = f"""
SELECT
USER_EMAIL,
ENTRY_DATE,
KEEP_TEXT,
PROBLEM_TEXT,
TRY_TEXT,
FEEDBACK_TEXT
FROM {KPT_TABLE}
WHERE USER_EMAIL = ?
AND ENTRY_DATE = ?
"""
detail = session.sql(detail_sql, params=[selected_email, selected_date]).collect()
except Exception as e:
st.error(f"詳細取得エラー: {e}")
detail = []
if not detail:
st.warning("⚠️ 詳細データが見つかりませんでした（削除された可能性があります）。")
continue
# SQLの取得結果の1件目を取得する
d0 = detail[0]
st.markdown("##### 📗 Keep")
with st.container(border=True):
st.text(d0["KEEP_TEXT"])
st.markdown("##### 📕 Problem")
with st.container(border=True):
st.text(d0["PROBLEM_TEXT"])
st.markdown("##### 📘 Try")
with st.container(border=True):
st.text(d0["TRY_TEXT"])
st.markdown("##### 🤖 AIフィードバック")
with st.container(border=True):
st.text(d0["FEEDBACK_TEXT"])
画面イメージ
入力画面
こんな感じで、KPTを入力したらAIからフィードバックが返ってきます。
履歴閲覧画面
日毎に各ユーザーのボタンが出てくるので、クリックすると下側にKPTの内容が表示されます。
(人数が少ないため、この単純なUIでも使い物にはなるが、対象者を絞り込む機能なども合った方が良さそう)
触ってみて感じたこと
ポジティブ面
Streamlit自体、最低限動くものを作る程度なら、そこまで困らずサクッと作れる印象です。
(Github CopilotやClaude Codeのようなコーディングエージェントを使えば、ある程度作ってくれますし。)
また、実際に触ってみて良いなと思ったのは以下のような点です。
-
アプリのホスティング環境や権限・認証設定周りを気にしなくて良いこと
- 対象ユーザーがSnowflakeユーザーであれば、動くものを、すぐに、 提供できる
-
シームレスにAIモデルを利用できること
- SQLのパラメータ選択だけで、様々なモデルを試せる気軽さ
- トークンの設定や管理などが不要なこと
- CORTEX関数のクレジット消費状況を見られるビューもあるらしい
つまり、全部Snowflake内で完結することが最大の魅力だと感じました。
様々な企業の現場では、別のサービスの利用が発生すると、社内申請などが必要となり、導入に対する心理的なハードルが上がってしまうこともありますが、既にSnowflakeを使っているなら、「すぐに始める」状態になれるのが改めて強いなと。
ネガティブ面
- アプリの起動が遅い
- 利用頻度が低いアプリだと、停止しているウェアハウスを起動する待ち時間が発生するため、数秒間の待ち時間が発生します。
- この数秒が割とストレス。。。
- アプリ起動中にウェアハウスのコストがかかり続ける
- 利用頻度や利用時間が長いアプリだと、結構コストが膨らみそう
- そういう場合は、Container Runtimeにすればいいんだろうけど、それでもある程度のコストはかかるのでは
- (アプリを使っている間、ぼんやり「お金がかかっているんだなー」という雑念が入り込む)
- 利用頻度や利用時間が長いアプリだと、結構コストが膨らみそう
おわりに
ひとまず、Streamlit in SnowflakeのキャッチアップのためにKPTアプリを作ってみたけど、コードさえ書けばクイックに始められるという魅力を実感できました！
一方で、この程度のシンプルなAIアプリであれば、Difyのようなノーコード開発でも十分な気がしました。
次はDifyで同じようなアプリも作ってみて、「各個人への週次レポート」や、「リーダー向けのチームサマリー提供」といった応用的な活用も試してみようと思います！
株式会社youthful daysのテックブログです。 データエンジニアリングを中心に、日々の学びや取り組みを発信します（ときどき農業やメンバーの興味・関心の話も）。
Discussion