my_chat_app.py

import os
import random
from pathlib import Path
from typing import AsyncGenerator

import gradio as gr
from strands import Agent, tool
from strands.models.anthropic import AnthropicModel
from strands.session.file_session_manager import SESSION_PREFIX, FileSessionManager
from strands_tools import calculator, current_time

# セッションの保存ディレクトリ
SESSIONS_DIR = Path("./sessions")
SESSIONS_DIR.mkdir(exist_ok=True)


@tool
def roll_dice(num_dice: int) -> list[int]:
    """
    サイコロを転がして出た目の数を返します。

    Args:
        num_dice (int): サイコロの数

    Returns:
        list[int]: それぞれのサイコロの出た目の値
    """
    return [random.randint(1, 6) for _ in range(num_dice)]


def get_available_session_names() -> list[str]:
    """既存のセッション一覧を取得"""
    prefix_size = len(SESSION_PREFIX)
    session_names = []
    for path in SESSIONS_DIR.iterdir():
        if path.name.startswith(SESSION_PREFIX) and path.is_dir():
            session_names.append(path.name[prefix_size:])
    return session_names


def create_agent(system_prompt: str, session_id: str) -> tuple[Agent, list[dict]]:
    """エージェントを作成し、セッションIDに基づいてチャットの履歴を復元"""
    # モデルの定義
    # 今回はAnthoropicのAPIを利用 (APIキーが必要)
    model = AnthropicModel(
        model_id="claude-sonnet-4-0",
        max_tokens=8192,
        client_args={"api_key": os.environ["ANTHROPIC_API_KEY"]},
    )

    # ローカルファイルベースのセッション管理
    # "./sessions" の下にセッションIDのフォルダが作られて、チャット履歴が記録される
    session = FileSessionManager(session_id=session_id, storage_dir=str(SESSIONS_DIR))

    # エージェントを作成
    agent = Agent(
        model=model,
        session_manager=session,
        tools=[calculator, current_time, roll_dice],  # ツールを登録
        system_prompt=system_prompt,
    )

    # チャットの履歴から本文だけを取り出し
    # 元の履歴にはツール呼出しなどの情報も記録されており、GradioのチャットUIとは互換性がない
    # ここでつじつま合わせをする
    chat_history = []
    for message in agent.messages:
        # チャットの本文（"text" コンテンツの中身）だけ取り出す
        contents = message.get("content", [])
        text_items = (x["text"] for x in contents if x.get("text"))
        text = next(text_items, None)
        if text:
            chat_history.append({"role": message.get("role"), "content": text})

    return agent, chat_history


async def handle_chat(message, _history, agent: Agent) -> AsyncGenerator[str, None]:
    """引数のユーザーメッセージに対して返答する"""
    chunks = []
    async for event in agent.stream_async(message):
        if "data" in event:
            chunks.append(event["data"])
            yield "".join(chunks)


with gr.Blocks(fill_height=True) as demo:
    # 設定パネル
    with gr.Accordion("🛠️ 設定", open=False):
        system_prompt = gr.Textbox(
            value="あなたは有能なアシスタントです。",
            label="システムプロンプト",
        )
        session_id = gr.Dropdown(
            choices=get_available_session_names(),
            value="default",
            label="セッションID",
            allow_custom_value=True,
        )

    # エージェント格納用の入れ物
    agent_state = gr.State(None)
    # メインのチャットUI
    chat = gr.ChatInterface(
        handle_chat, type="messages", additional_inputs=[agent_state]
    )

    # エージェントを作成
    gr.on(
        # 起動時、設定変更時に呼び出し
        [chat.load, system_prompt.blur, session_id.blur],
        create_agent,
        inputs=[system_prompt, session_id],
        outputs=[agent_state, chat.chatbot_value],
    )

# Gradioアプリとして起動
demo.launch()