LoginSignup
22
27

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@moritalous

Streamlitでアプリを作って実感するStrands Agentsのいいところ

Posted at

Strands Agentsにもうちょっと踏み込んでみたいと思い、Streamlitのチャットを作りました。

この記事の前にこのあたりを先に読まれるとスムーズかと思います。

事前準備

ライブラリーをインストールしましょう。

uv add strands-agents streamlit nest-asyncio

1. 最低限の機能で実装する

ほとんどStrands Agentsの良さは発揮されてませんが、一番最低限なチャットはこうなります。

1_simple.py
import boto3
import streamlit as st
from strands import Agent
from strands.models import BedrockModel

st.title("Strands agent")


if prompt := st.chat_input():
    with st.chat_message("user"):
        st.write(prompt)

    agent = Agent(
        model=BedrockModel(
            model_id="us.amazon.nova-pro-v1:0",
            boto_session=boto3.Session(region_name="us-east-1"),
        ),
        callback_handler=None
    )

    response = agent(prompt=prompt)

    st.write(response.message["content"][0]["text"])

callback_handlerが未指定の場合は、デフォルトでPrintingCallbackHandlerというコールバックハンドラーが指定されます。（PrintingCallbackHandlerは標準出力にBedrockの返答を出力します。）
Streamlitで画面上に出力するので標準出力への出力を抑制するためにNoneを指定します。

2. ストリームでレスポンスを出力する

次はストリームで出力する方法です。これもほとんどConverse APIを使う場合と変わらないです。

Streamlitの非同期処理とバッティングするのか（？）、nest_asyncioが必要でした。

2_streaming.py
import asyncio

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        text = (
            event.get("event", {})
            .get("contentBlockDelta", {})
            .get("delta", {})
            .get("text", "")
        )

        yield text


async def main():
    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            callback_handler=None
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

3. 会話履歴を保持する

ここまでの実装では、毎回新規の会話が行われますので、過去の会話内容は反映されません。

エージェントの実行後、その会話のやり取りはagent.messagesで取得できるので、これをst.session_stateで管理します。

3_conversation_history.py
import asyncio

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel
from strands.types.content import Messages

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        text = (
            event.get("event", {})
            .get("contentBlockDelta", {})
            .get("delta", {})
            .get("text", "")
        )

        yield text


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            with st.chat_message(message["role"]):
                st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

4. 会話の最大数を管理する

このあたりからStrands Agentsの機能が活用されていきます。

会話の最大数を決めて、自動的に過去の会話を自動的に忘れることができます。

Agentのconversation_managerパラメーターにSlidingWindowConversationManagerを指定します。

window_sizeを奇数にすると、userメッセージだけ削除され、次のリクエストで「userから始まってないよエラー」になります。なので、偶数である必要がありそうです。

import asyncio

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.agent.conversation_manager import SlidingWindowConversationManager
from strands.models import BedrockModel
from strands.types.content import Messages

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        text = (
            event.get("event", {})
            .get("contentBlockDelta", {})
            .get("delta", {})
            .get("text", "")
        )

        yield text


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            with st.chat_message(message["role"]):
                st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
            conversation_manager=SlidingWindowConversationManager(window_size=4)
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

5. ツールを使う（ビルドインツール）

このあたりから「おっ！」と思う感じになってきます。

Strands Agentsには、豊富なビルドインツールが提供されています。

image.png

なかなか面白そうなツールが盛り沢山です。

Strands Agentsとは別のライブラリーとして提供されていますので、追加でインストールします。

uv add strands-agents-tools

ツールを呼び出す場合、いわゆる「イベントループ」の形でwhileループを組む必要があります。

Strands Agentsを使うと、 このイベントループ処理を時前で実装する必要はありません！ （すごい！）

もうこれだけでStrands Agentsの採用決定です。

import asyncio
import os

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel
from strands.types.content import Messages
from strands_tools import shell

nest_asyncio.apply()

os.environ["DEV"] = "true"

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        if "event" in event:
            text = (
                event.get("event", {})
                .get("contentBlockDelta", {})
                .get("delta", {})
                .get("text", "")
            )
            yield text

        elif "current_tool_use" in event:
            current_tool_use = event.get("current_tool_use", {})

            yield f"\n\n```\n🔧 Using tool: {current_tool_use}\n```\n\n"


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            if "text" in message["content"][0]:
                with st.chat_message(message["role"]):
                    st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
            tools=[shell],
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

6. ツールを使う（MCPツール）

Strands AgentsはMCPにも対応しています。

import asyncio

import boto3
import nest_asyncio
import streamlit as st
from mcp import StdioServerParameters, stdio_client
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient
from strands.types.content import Messages

nest_asyncio.apply()

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        if "event" in event:
            text = (
                event.get("event", {})
                .get("contentBlockDelta", {})
                .get("delta", {})
                .get("text", "")
            )
            yield text

        elif "current_tool_use" in event:
            current_tool_use = event.get("current_tool_use", {})

            yield f"\n\n```\n🔧 Using tool: {current_tool_use}\n```\n\n"


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            if "text" in message["content"][0]:
                with st.chat_message(message["role"]):
                    st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        stdio_mcp_client = MCPClient(
            lambda: stdio_client(
                StdioServerParameters(
                    command="uvx",
                    args=["awslabs.aws-documentation-mcp-server@latest"],
                    env={"FASTMCP_LOG_LEVEL": "ERROR"},
                )
            )
        )

        with stdio_mcp_client:
            tools = stdio_mcp_client.list_tools_sync()

            agent = Agent(
                model=BedrockModel(
                    model_id="us.amazon.nova-pro-v1:0",
                    boto_session=boto3.Session(region_name="us-east-1"),
                ),
                messages=st.session_state.messages if "messages" in st.session_state else [],
                callback_handler=None,
                tools=tools,
            )

            agent_stream = agent.stream_async(prompt=prompt)

            st.write_stream(streaming(agent_stream))

            st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

7. トレースもとれる

Strands Agentsではオブザーバビリティ機能も用意されています。トレース情報はOpenTelemetryの形式で対応しているとのことです。

みんな大好きLangfuseはすでにドキュメントに記載があります。はやい！

Arize Phoenixを使う場合は、以下のように環境変数``を設定するだけです。

import asyncio
import os

import boto3
import nest_asyncio
import streamlit as st
from strands import Agent
from strands.models import BedrockModel
from strands.types.content import Messages
from strands_tools import shell

nest_asyncio.apply()

os.environ["DEV"] = "true"
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "http://localhost:6006/v1/traces"

st.title("Strands agent")


async def streaming(stream):
    async for event in stream:
        if "event" in event:
            text = (
                event.get("event", {})
                .get("contentBlockDelta", {})
                .get("delta", {})
                .get("text", "")
            )
            yield text

        elif "current_tool_use" in event:
            current_tool_use = event.get("current_tool_use", {})

            yield f"\n\n```\n🔧 Using tool: {current_tool_use}\n```\n\n"


async def main():
    if "messages" in st.session_state:
        messages: Messages = st.session_state.messages
        for message in messages:
            if "text" in message["content"][0]:
                with st.chat_message(message["role"]):
                    st.write(message["content"][0]["text"])

    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

        agent = Agent(
            model=BedrockModel(
                model_id="us.amazon.nova-pro-v1:0",
                boto_session=boto3.Session(region_name="us-east-1"),
            ),
            messages=st.session_state.messages if "messages" in st.session_state else [],
            callback_handler=None,
            tools=[shell],
        )

        agent_stream = agent.stream_async(prompt=prompt)

        st.write_stream(streaming(agent_stream))

        st.session_state.messages = agent.messages


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

だがしかし、日本語は正しく表示されません。

localhost_6006_projects_UHJvamVjdDox_spans_7ef240efca1bde40f7348b89b798c424_selectedSpanNodeId=U3Bhbjo1OA==.png

ここでプルリク上げてますので、マージされるのをお待ち下さい。

22
27
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
22
27

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?