Streamlitを使って生成AIの試行錯誤ができるWebアプリを作った話
風音屋アドバイザーの渡部徹太郎(@fetarodc) です。 このブログでは、Streamlitを使って、様々な形式の入力を受け付けて、Pythonのプログラムに渡すWebアプリの具体的な作り方を解説します。
このブログで得られる知見
- Streamlitを用いた、様々な形式の入力を受け付けてPythonのプログラムに渡すWebアプリの、具体的な作り方
- 状態を保持するsession_stateを使って、「ファイルパスを指定して中身を読み込むボタン」を作る方法
- Webアプリ上で入力データを編集し、Pythonプログラムに渡す方法。特に編集可能なテーブルデータ入力フォームの作り方
- 画面の入力をもとにPythonプログラムを実行したときに、実行結果をリアルタイムにWeb画面に出す方法
Webアプリ開発の目的
今回作ったのは、OpenAIのChatGPTに対する入力を色々変えながら実行し、 どういう入力が一番良い性能かを試行錯誤できるツールです。
ツールを作るに当たり以下のような要件がありました。
要件1:様々な形式の入力を受付られること
このツールはWeb画面から以下の入力を受け取って、ChatGPTのAPIに渡して動作させます。
- 会話データ(CSVデータ)
- 生成AIのシステムプロンプト(テキストデータ)
- ChatGPTのモデル(文字列の選択肢)
- ChatGPTのパラメータ(JSON)
要件2:入力がWeb画面上で直接編集できること
利用者が試行錯誤をするツールであるため、上記の4つの入力をWeb画面上で直接編集できることが必要でした。 特に、CSVデータはテーブルとして表示し、セルの値を直接編集できることが求められました。
要件3:実行結果がWeb画面にリアルタイムで表示されること
上記の入力を下に、ChatGPTのAPIに渡して実行するのですが、 実行には時間がかかったり、エラーが発生することがあるため、 実行結果がWeb画面にリアルタイムで表示されることが必要でした。
作ったもの
Web画面
コード
main.py
import json
import traceback
import logging
import pandas as pd
import streamlit as st
from pathlib import Path
from gpt_caller import GptCaller
# streamlitのsession_stateのキー
SESSION_KEY_SYSTEM_PROMPT_TEXT = "system_prompt_text"
SESSION_KEY_TALK_DF = "talk_df"
SESSION_KEY_TALK_EDITED_DF = "talk_edited_df"
# ---------------------------------------------------------------------------------------------------------------------
# 入力
# ---------------------------------------------------------------------------------------------------------------------
st.subheader("システムプロンプト")
system_prompt_url = st.text_input(label="システムプロンプトのパス", value="",placeholder="/path/to/system_prompt.txt")
if st.button("読み込み", key="read_system_prompt_button", type="primary"):
try:
st.session_state[SESSION_KEY_SYSTEM_PROMPT_TEXT] = Path(system_prompt_url).read_text()
except Exception as e:
st.error(e)
# 読み込まれたら内容を表示
if SESSION_KEY_SYSTEM_PROMPT_TEXT in st.session_state:
system_prompt = st.text_area(
label="実行されるプロンプト(直接編集できます)",
value=st.session_state.get(SESSION_KEY_SYSTEM_PROMPT_TEXT, ""),
height=st.session_state.get(SESSION_KEY_SYSTEM_PROMPT_TEXT, "").count("\n") * 25,
)
else:
system_prompt = ""
# -------------------------------------------
st.subheader("会話データ")
st.write("ファイルパスを指定してください")
talk_path = st.text_input(label="会話データのパス", value="", placeholder="/path/to/talk.csv")
if st.button("読み込み", key="read_talk_button", type="primary"):
try:
st.session_state[SESSION_KEY_TALK_DF] = pd.read_csv(talk_path)
except Exception as e:
st.error(e)
# 読み込まれたら内容を表示
if SESSION_KEY_TALK_DF in st.session_state:
# st.data_editorを用いることで、編集可能なテーブルデータ(DataFrame)を表示できる
# 編集したものはsession_stateに保存しておく
st.session_state[SESSION_KEY_TALK_EDITED_DF]: pd.DataFrame = st.data_editor(st.session_state[SESSION_KEY_TALK_DF])
# -------------------------------------------
st.subheader("GPTモデル")
gpt_model_name = st.selectbox(
label="GPTモデル",
options=["gpt-3.5-turbo", "gpt-4o", "gpt-4o-turbo"],
index=2,
)
# -------------------------------------------
st.subheader("GPTパラメータ")
gpt_parameters_json = st.text_area(
label="GPTパラメータ(JSON)",
value=json.dumps({"temperature": 0.5, "frequency_penalty": 0.1})
)
# ---------------------------------------------------------------------------------------------------------------------
# 実行
# ---------------------------------------------------------------------------------------------------------------------
if st.button(key="run", label="実行", type="primary"):
try:
gpt_parameters = json.loads(gpt_parameters_json)
except Exception as e:
st.error("GPTパラメータが不正なJSONです: " + str(e))
st.stop()
if system_prompt == "":
st.error("システムプロンプトが読み込まれていません")
st.stop()
if st.session_state[SESSION_KEY_TALK_DF].empty:
st.error("会話データが読み込まれていません")
st.stop()
st.write("実行開始")
st.warning(
"実行中は画面の入力値を更新しないでください。更新すると実行ログが表示されなくなると共に、実行停止するか裏で実行継続するか、判別不能な不安定な状態になります")
st.write("↓↓実行ログ↓↓")
# streamlitの画面にログを表示するためのハンドラ
class StreamlitLogHandler(logging.Handler):
msgs = ""
def __init__(self, widget_update_func):
super().__init__()
self.widget_update_func = widget_update_func
self.setFormatter(logging.Formatter("[%(levelname)s] %(message)s"))
def emit(self, record):
msg = self.format(record)
self.msgs = self.msgs + msg + "\n"
self.widget_update_func(self.msgs)
# ロガーの作成
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# 標準出力に印字するログハンドラを登録
stream_handler = logging.StreamHandler()
log_formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%dT%H:%M:%S%z")
stream_handler.setFormatter(log_formatter)
logger.addHandler(stream_handler)
# Streamlitに出力するログハンドラを登録
logger.addHandler(StreamlitLogHandler(st.empty().text))
try:
# GPT
score = GptCaller.run(
system_prompt=system_prompt,
takl_df=st.session_state[SESSION_KEY_TALK_EDITED_DF],
gpt_model_name=gpt_model_name,
gpt_parameters=gpt_parameters,
logger=logger,
)
st.write(f"実行結果 score={score}")
except Exception as e:
st.error(f"エラーが発生しました:{e}")
st.text_area(label="エラー内容", value=traceback.format_exc())
gpt_caller.py
import logging
import pandas as pd
class GptCaller:
@staticmethod
def run(
system_prompt: str,
takl_df: pd.DataFrame,
gpt_model_name: str,
gpt_parameters: dict,
logger: logging.Logger
):
logger.info("入力を表示")
logger.info(f"{system_prompt=}")
logger.info(f"{takl_df=}")
logger.info(f"{gpt_model_name=}")
logger.info(f"{gpt_parameters=}")
logger.info("GPT呼び出し開始")
logger.info(f"GPTの結果 = xxxxxxxx")
logger.info(f"GPTの結果を評価")
score = 0.82
logger.info(f"GPTの評価結果 = {score}")
return score
特徴的な部分の説明
画面変化しても値を保持する st.session_state
を使って読み込みボタンを実現
最初に、システムプロンプトを入力する部分ですが、ここはファイルのパスを指定して読み込みボタンを押すと、 内容が読み込まれて画面に表示されます。
ポイントとしては、ファイルの内容を表示する部分のコードを、ボタンのif文の外側に出している点です。 ソースコード上でいうとこの部分です。
if st.button("読み込み", key="read_system_prompt_button", type="primary"):
try:
st.session_state[SESSION_KEY_SYSTEM_PROMPT_TEXT] = Path(system_prompt_url).read_text()
except Exception as e:
st.error(e)
# 読み込まれたら内容を表示
if SESSION_KEY_SYSTEM_PROMPT_TEXT in st.session_state:
system_prompt = st.text_area(
label="実行されるプロンプト(直接編集できます)",
value=st.session_state.get(SESSION_KEY_SYSTEM_PROMPT_TEXT, ""),
height=st.session_state.get(SESSION_KEY_SYSTEM_PROMPT_TEXT, "").count("\n") * 25,
)
else:
system_prompt = ""
もしボタンのif文の中に表示するコードを書いてしまうと、他のボタンをクリックされたときに、 このif文を通らないため、ファイルの内容を表示するコードが実行されません。 なぜこうなるかというと、Streamlitは、仕様上、画面変化イベントがあった場合プログラムの先頭から再実行するためです。 画面変化イベントは、ボタンのクリックや、テキストエリアの編集、選択リストの選択などあらゆるところで発生します。
これを回避するためには、ボタンが押されたときに、Streamlitが用意している st.session_state
にファイルの内容を保存しておきます。
st.session_state
は画面変化に関わらず、値を保持してくれるので、画面変化が起きてもファイルの内容を保存してくれます。
これを用いて、ボタンのif文の外側で表示することで、画面変化イベントがどのように発生しても、常にファイルの値を画面に表示できます。
編集可能なテーブル st.data_editor
を使って読み込みボタンを実現
会話データの読み込みの部分では、CSVを読み込んで、それを画面に表示しています。
パスを入力し読み込みボタンを押すと、テーブルが表示されます。ポイントは、このテーブルが編集可能であるという点です。以下のようにセル単位で編集できます。
これは、読み込んだCSVをpandasのDataFrameに変換し、st.session_state[SESSION_KEY_TALK_DF]
に格納。
その後、Streamlitの st.data_editor
に渡すことで、編集可能なテーブルを表示しています。
st.data_editor
の戻り値は session_state
に保存しておくことで、他の画面変化があっても値が保持されるようにしています。
if st.button("読み込み", key="read_talk_button", type="primary"):
try:
st.session_state[SESSION_KEY_TALK_DF] = pd.read_csv(talk_path)
except Exception as e:
st.error(e)
# 読み込まれたら内容を表示
if SESSION_KEY_TALK_DF in st.session_state:
# st.data_editorを用いることで、編集可能なテーブルデータ(DataFrame)を表示できる
# 編集したものはsession_stateに保存しておく
st.session_state[SESSION_KEY_TALK_EDITED_DF]: pd.DataFrame = st.data_editor(st.session_state[SESSION_KEY_TALK_DF])
リストの選択は st.selectbox
でサクッと
GPTのモデルは st.selectbox
を使って、選択肢を選ぶようにしています。簡単に実現できました。
gpt_model_name = st.selectbox(
label="GPTモデル",
options=["gpt-3.5-turbo", "gpt-4o", "gpt-4o-turbo"],
index=2,
)
JSONの入力は、良い部品がなかったため、st.text_area
を使いました
GPTのパラメータはJSON形式なので、st.json
を使って表示することも考えましたが、
st.json
は編集ができないため、JSONの文字列を st.text_area
で表示することで編集可能にしました。
gpt_parameters_json = st.text_area(
label="GPTパラメータ(JSON)",
value=json.dumps({"temperature": 0.5, "frequency_penalty": 0.1}))
いつかeditableなJSON部品ができることを祈っています。
実行結果のリアルタイム表示は独自の Logging Handler を使って実現
入力が揃ったら実行ボタンで実行しますが、実行ボタンを押したあとのプログラムの標準出力をStreamlitの画面に表示するようにしました。
これにはひと工夫必要です。
プログラム内でprint文を使っても、Streamlitの画面に表示されず、StreamlitのWebサーバを起動しているコンソールにしか出力されません。
そこで、PythonのLoggingモジュールにあるカスタムハンドラを使って、標準出力をStreamlitの画面に表示するようにしました。
Loggingモジュールにおけるハンドラとは、出力先を指定する部品のことです。 例えばStreamHandlerであれば標準出力に、FileHandlerであればファイルに出力します。 今回は、StreamlitHandlerという独自のハンドラを作成し、Streamlitの画面に出力するようにしました。
まず、以下の様にStreamlitHandlerを作成します。
class StreamlitLogHandler(logging.Handler):
msgs = ""
def __init__(self, widget_update_func):
super().__init__()
self.widget_update_func = widget_update_func
self.setFormatter(logging.Formatter("[%(levelname)s] %(message)s"))
def emit(self, record):
msg = self.format(record)
self.msgs = self.msgs + msg + "\n"
self.widget_update_func(self.msgs)
logging.Handlerクラスを継承して、独自のハンドラを作成します。
これを以下のようにしてloggerにセットします。
logger.addHandler(StreamlitLogHandler(st.empty().text))
このようにすることで、loggerに出力された内容が、Streamlitの画面に表示されるようになります。
ポイントは、 st.empty().text
を渡している点です。
st.empty()
はStreamlitの空のウィジェットであり、そのウィジェットにテキストを維持する関数である text
を引数に渡します。
渡された関数は、loggerのハンドラ内で呼び出され、出力された内容がStreamlitの画面に表示されます。
なお、この方法は2025年5月時点では公式ドキュメントで案内されている方法ではありませんが、公式のディスカッションの場では議論されている方法になります。
感想
Streamlitを使うことで、HTML、JavaScript、CSSの機能を使わずに、様々な入力を受け取って実行結果を表示するWeb画面が作れました。 普段Pythonで生活しているデータサイエンティストが、ちょっとしたWebアプリを作りたいと思ったら、Streamlitは最適解だと感じました。
ただ、画面上で様々なUIイベントが起こるWebアプリを作ろうと思うと、Streamlitの「画面が変わるたびにプログラムの先頭から実行される」という仕様が厳しい制約になってきます。
今回のように、入力データをロードしたり編集してプログラムを実行するぐらいのUIの変化であればStreamlitで十分ですが、 それ以上に複雑なUIイベント、例えば入力項目を動的に増やすようなWebアプリを作る場合は、従来のJavaScriptベースのWebアプリフレームワークを使うべきだと思いました。