0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@jungyeounjae(yj j)

Streamlit × Agents SDK — Runner.run_streamedでリアルタイム応答を表示する

0
Posted at

Streamlit × Agents SDK — Runner.run_streamedでリアルタイム応答を表示する

前回までで Streamlit の UI 骨格（chat_message / session_state）と、Agents SDK 単体でのエージェント実行（ノートブック）を別々に見てきました。

今回は main.py で両者を接続 し、Streamlit 上で LLM のストリーミング回答 を表示します。新しいポイントは Runner.run_streamed と、2種類の ストリームイベント の使い分けです。

※ session_state の基本や st.chat_message の説明は前回記事を参照。

全体構成

st.chat_input（ユーザー入力）
        ↓
asyncio.run(run_agent(message))
        ↓
Runner.run_streamed(agent, message, session=session)
        ↓
stream.stream_events() を async for で処理
        ↓
st.empty() に delta を追記 → リアルタイム表示
        ↓
SQLiteSession が会話を DB に保存
レイヤー 担当
Streamlit 入力・表示・rerun
asyncio.run Streamlit（同期）と Agents SDK（非同期）の橋渡し
Runner.run_streamed エージェント実行 + ストリーミング
SQLiteSession エージェント向け会話履歴（前回ノートブックと同じ）

コード全体

import asyncio

import streamlit as st
from agents import Agent, ItemHelpers, Runner, SQLiteSession
from dotenv import load_dotenv

load_dotenv()

if "agent" not in st.session_state:
    st.session_state["agent"] = Agent(
        name="ChatGPT Clone Agent",
        instructions="""
        あなたはChatGPTのようなAIアシスタントです。
        """,
    )

agent = st.session_state["agent"]

if "session" not in st.session_state:
    st.session_state["session"] = SQLiteSession(
        "chat-history",
        "chat-gpt-clone-memory.db",
    )

session = st.session_state["session"]


async def run_agent(message: str) -> str:
    stream = Runner.run_streamed(agent, message, session=session)
    response = ""

    with st.chat_message("ai"):
        placeholder = st.empty()
        async for event in stream.stream_events():
            if event.type == "raw_response_event":
                if event.data.type == "response.output_text.delta":
                    response += event.data.delta
                    placeholder.write(response)
            elif event.type == "run_item_stream_event":
                if event.item.type == "message_output_item":
                    response = ItemHelpers.text_message_output(event.item)
                    placeholder.write(response)

    return response


prompt = st.chat_input("メッセージを入力してください")

if prompt:
    with st.chat_message("human"):
        st.write(prompt)
    asyncio.run(run_agent(prompt))

with st.sidebar:
    reset = st.button("Reset memory")
    if reset:
        asyncio.run(session.clear_session())
        st.rerun()
    st.write(asyncio.run(session.get_items()))

起動:

uv run streamlit run main.py

.env に OPENAI_API_KEY が必要です。load_dotenv() で読み込みます。

2026-07-2722.52.50-ezgif.com-video-to-gif-converter.gif

Runner.run_streamed — 引数の順番に注意

# ❌ session を input の位置に渡してしまう
stream = Runner.run_streamed(agent, session)

# ✅ 2番目はユーザーメッセージ、session はキーワード引数
stream = Runner.run_streamed(agent, message, session=session)

シグネチャは run_streamed(agent, input, session=session) です。

間違えると 'SQLiteSession' object has no attribute 'copy' のようなエラーになります。SDK は 2番目引数を ユーザー入力（文字列） として扱うためです。

asyncio.run — Streamlit と async の接続

Streamlit のスクリプトは 同期 で上から下へ実行されます。Agents SDK の stream_events() は async for が必要です。

asyncio.run(run_agent(prompt))
Streamlit Agents SDK
実行モデル 同期（rerun） 非同期（async/await）
橋渡し asyncio.run(コルーチン) async def run_agent

run_agent を async 関数にし、呼び出し側で asyncio.run する — これが定番パターンです。

ストリームイベント — 2種類の使い分け

async for event in stream.stream_events():
    if event.type == "raw_response_event":
        if event.data.type == "response.output_text.delta":
            response += event.data.delta
            placeholder.write(response)
    elif event.type == "run_item_stream_event":
        if event.item.type == "message_output_item":
            response = ItemHelpers.text_message_output(event.item)
            placeholder.write(response)

raw_response_event — リアルタイム

OpenAI API から届く テキストの断片（delta） です。

"こ" → "こん" → "こんに" → "こんにち" → ...

response += event.data.delta で蓄積し、placeholder.write(response) で画面更新。ChatGPT のように 文字が少しずつ現れる UX はここで作ります。

run_item_stream_event — 最終テキスト

Agents SDK が整理した 実行結果 です。message_output_item は 完成した AI メッセージ を表します。

イベント タイミング 用途
raw_response_event + delta 生成中 リアルタイム表示
run_item_stream_event + message_output_item 完了時 最終テキストの確定

delta だけでも動きますが、両方処理しておくと 表示の抜けやズレ を防ぎやすくなります。

st.empty() — 1つの吹き出しの中で更新

with st.chat_message("ai"):
    placeholder = st.empty()
    placeholder.write(response)  # 同じ場所を何度も上書き

st.write(response) をループ内で毎回呼ぶと、AI 吹き出しが増殖 する恐れがあります。

st.empty() で 1スロット を確保し、そこに delta を追記していく — ストリーミング UI の定番です。

session_state に Agent と SQLiteSession を置く理由

if "agent" not in st.session_state:
    st.session_state["agent"] = Agent(...)

if "session" not in st.session_state:
    st.session_state["session"] = SQLiteSession(...)

rerun のたびに Agent や DB セッションを 作り直すと、接続の無駄や状態の不整合が起き得ます。

session_state のキー 中身 rerun 後
agent Agent インスタンス 再利用
session SQLiteSession 再利用（DB ファイルはそのまま）

前回記事の messages リストが UI 用の記憶 なら、ここでは エージェント実行に必要なオブジェクト を rerun 間で保持しています。

サイドバー — DB の中身とリセット

st.write(asyncio.run(session.get_items()))  # DB に保存された会話履歴

SQLiteSession に保存された履歴をそのまま表示できます。UI の吹き出しが rerun で消えても、DB 側には会話が残っている ことを確認するデバッグ用パネルです。

asyncio.run(session.clear_session())
st.rerun()

Reset で DB の会話を消去し、即 rerun します。

まとめ

  • Runner.run_streamed(agent, message, session=session) — 第2引数は必ずユーザーメッセージ
  • asyncio.run — Streamlit から async エージェント処理を呼ぶ
  • raw_response_event + delta — リアルタイムストリーミング
  • run_item_stream_event + message_output_item — 最終テキストの確定
  • st.empty() — 1つの AI 吹き出し内で delta を追記
  • session_state — Agent / SQLiteSession を rerun 間で保持
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?