【Streamlit】st.write_stream()でデータストリームを表示する

こんにちは、JS2IIUです。
Streamlitで動的なデータを扱う際に便利なst.write_stream()について解説します。この関数は、ジェネレータ、イテラブル、またはストリームのようなシーケンスをアプリにストリーミング表示することができます。リアルタイムのデータ更新を表示するのに最適です。今回もよろしくお願いします。

シグネチャ
サンプルプログラム
参考になるWEBサイト

シグネチャ

Python

st.write_stream(data, *, key=None)

data: ジェネレータ、イテラブル、またはストリームのようなシーケンス。
key: ウィジェットの識別に使用するオプションのキー。

サンプルプログラム

1. 無限にカウントアップする数字を表示

Python

import streamlit as st
import time

def counter():
  i = 0
  while True:
    yield i
    i += 1
    time.sleep(1)

st.write_stream(counter())

解説:

counter()関数は無限に数字を生成するジェネレータです。yieldキーワードで数字を1つずつ返します。
time.sleep(1)で1秒ごとに数字を生成します。
st.write_stream(counter())でジェネレータをStreamlitアプリにストリーミング表示します。

2. ファイルから1行ずつ読み込んで表示

Python

import urllib.request

import streamlit as st
import time

def read_file(file_path):
    with open(file_path, 'r') as f:
        for line in f:
            for word in line.split():
                yield word + " "
                time.sleep(0.05)  # 表示間隔を調整

def read_web_file(url):
    with urllib.request.urlopen(url=url) as f:
        lines = f.read().decode('cp932').split('\n')
        for line in lines:
            for word in line.split():
                yield word + " "
                time.sleep(0.05)  # 表示間隔を調整

# タイトル
st.title("テキストファイルのストリーミング表示")

# ファイルの内容をストリーミング表示
st.write_stream(read_web_file('https://raw.githubusercontent.com/JS2IIU-MH/Streamlit_sample_data/refs/heads/main/data/lipsum.txt'))

解説:

read_file()関数はファイルから1行ずつ読み込むジェネレータです。
read_web_file()関数はWEB上のファイルを読みこみ、1行ずつ読みこむジェネレータです。
st.write_stream(read_file('data.txt'))でdata.txtファイルの内容を1行ずつストリーミング表示します。
WEB上のテキストファイルはこちらを使いました。

Streamlit_sample_data/data/lipsum.txt at main · JS2IIU-MH/Streamlit_sample_data

Sample data for blog post about Streamlit. Contribute to JS2IIU-MH/Streamlit_sample_data development by creating an account on GitHub.

3. 外部APIからデータを取得して表示

Python

import streamlit as st
import requests
import time

def fetch_data(url):
  while True:
    response = requests.get(url)
    data = response.json()
    yield data
    time.sleep(5)

st.write_stream(fetch_data('https://api.example.com/data'))

解説:

fetch_data()関数は外部APIからデータを定期的に取得するジェネレータです。
requests.get(url)でAPIにリクエストを送信し、response.json()でJSON形式のレスポンスをパースします。
time.sleep(5)で5秒ごとにAPIからデータを取得します。
st.write_stream(fetch_data('https://api.example.com/data'))でAPIから取得したデータをストリーミング表示します。

4. GPT-o4のチャットを表示する

Python

import streamlit as st
from openai import OpenAI

# OpenAIクライアントを生成
client = OpenAI(api_key="YOUR API KEY HERE")  # ここにあなたのAPIキーを入力してください

# セッションステートを初期化
if "messages" not in st.session_state:
    st.session_state.messages = []

# メッセージ履歴を表示
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# ユーザーからの入力を受け取る
if prompt := st.chat_input("What is up?"):
    # ユーザーのメッセージを履歴に追加
    st.session_state.messages.append({"role": "user", "content": prompt})
    # チャットメッセージとして表示
    with st.chat_message("user"):
        st.markdown(prompt)

    # OpenAI APIを使用して応答を生成
    def generate_response():
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                *st.session_state.messages,
            ]
        )
        response_content = response.choices[0].message.content

        # AIの応答を逐次返すジェネレーター
        for chunk in response_content.split():
            yield chunk + " "  # 各単語にスペースを追加して出力
            st.time.sleep(0.1)  # 擬似ストリーミング効果のための遅延

    # AI応答をストリーミング表示
    with st.chat_message("assistant"):
        st.write_stream(generate_response)

    # 最終的な応答内容をセッションステートに追加
    full_response = "".join(generate_response())
    st.session_state.messages.append({"role": "assistant", "content": full_response})

コード解説

ライブラリのインポート

Python

import streamlit as st
from openai import OpenAI

streamlit: PythonでインタラクティブなWebアプリケーションを作成するためのライブラリ。UIコンポーネントを簡単に追加できます。
OpenAI: OpenAI APIを利用して、GPTモデル（例：GPT-4）などのAI機能を操作するためのライブラリ。

OpenAIクライアントの生成

Python

client = OpenAI(api_key="YOUR API KEY HERE")  # ここにあなたのAPIキーを入力してください

OpenAIのAPIを利用するためのクライアントを初期化。
api_key: OpenAIアカウントで取得したAPIキーを指定することで、リクエストを認証。

セッションステートの初期化

Python

if "messages" not in st.session_state:
    st.session_state.messages = []

セッションステート: Streamlitが提供するデータ永続化の仕組み。ページを再読み込みしても値が保持される。
st.session_state.messages: チャットメッセージの履歴を格納するリスト。role（ユーザー/アシスタント）とcontent（メッセージ本文）を保持します。
初回実行時にmessagesが存在しない場合、空のリストで初期化。

メッセージ履歴の表示

Python

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

履歴表示:
- st.session_state.messagesに保存されているメッセージを順次表示。
- st.chat_message(role):
- role に応じて、メッセージがユーザー側かアシスタント側かを切り替えて表示。
st.markdown(content):
- メッセージ本文をMarkdown形式で表示。

ユーザーの入力を受け取る

Python

if prompt := st.chat_input("What is up?"):

st.chat_input: チャット入力用のUIコンポーネント。ユーザーがテキストを入力すると、その内容を取得。
prompt :=: Python 3.8以降の代入式（ウォルラス演算子）。条件式と代入を同時に実行。

ユーザーのメッセージを追加

Python

st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
    st.markdown(prompt)

セッションステートへの追加:
- ユーザーの入力メッセージを履歴に保存。
画面への表示:
- st.chat_message("user")を使用して、入力されたメッセージをユーザー側に表示。

OpenAI APIによる応答生成

Python

def generate_response():
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            *st.session_state.messages,
        ]
    )
    response_content = response.choices[0].message.content

    # AIの応答を逐次返すジェネレーター
    for chunk in response_content.split():
        yield chunk + " "  # 各単語にスペースを追加して出力
        st.time.sleep(0.1)  # 擬似ストリーミング効果のための遅延

関数generate_response:
- OpenAIのchat.completions.createメソッドを使用して、GPTモデルから応答を取得。
- model: 使用するモデル名（例：gpt-4o-mini）。
- messages: チャットの履歴をAPIに渡して文脈を保持。
- response_content: AIからの応答内容を取得。
ストリーミング:
- 応答内容を単語単位で分割し、ジェネレーターを使用して1つずつ返す。
- st.time.sleep(0.1): 0.1秒の遅延を挿入して、ストリーミング表示を模倣。

AIの応答をストリーミング表示

Python

with st.chat_message("assistant"):
    st.write_stream(generate_response)

st.write_stream:
- generate_responseが生成するデータをリアルタイムで画面に出力。
- AIの応答が一度に表示されるのではなく、逐次レンダリングされる。

応答を履歴に追加

Python

full_response = "".join(generate_response())
st.session_state.messages.append({"role": "assistant", "content": full_response})

最終応答の統合:
- generate_responseから返された単語を結合し、完全な応答を生成。
セッションステートへの保存:
- 完全なAIの応答をst.session_state.messagesに追加し、履歴を更新。

動作の全体概要

ユーザーがメッセージを入力すると、入力内容が履歴に保存され、画面に表示されます。
OpenAI APIを使用してAI応答を取得。
AIの応答をst.write_streamで逐次表示。
応答の履歴を保存して、チャットの継続性を維持。

このプログラムは、ユーザー入力とAIの応答をリアルタイムにやり取りするシンプルなチャットアプリケーションの実装例です。st.write_streamを利用して、より直感的でインタラクティブなUIを実現しています。

参考になるWEBサイト

– st.write_stream – Streamlit Docs

注記: st.write_stream()は、Streamlitのバージョン1.3.0以降で利用可能です。もし古いバージョンを使用している場合は、アップグレードしてください。

最後に、書籍のPRです。

最新のOpenAIのチャットAPIの使い方もしっかりと解説されている良書です。2024年11月初版発行、「LangChainとLangGraphによるRAG・AIエージェント[実践]入門」西見、吉田、大嶋著。

最後まで読んでいただきありがとうございます。73

月	火	水	木	金	土	日
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30	31