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

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

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

https://qiita.com/minorun365/items/dd05a3e4938482ac6055

https://qiita.com/moritalous/items/6a9ed8db3efb00cca28e

## 事前準備

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

```shell
uv add strands-agents streamlit nest-asyncio
```

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

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

```python: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"])
```

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

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

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

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

```python: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`で管理します。

```python: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`を指定します。

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



```python
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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/bafee471-9edf-4067-a545-bc69235f9580.png)

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

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

```shell
uv add strands-agents-tools
```

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

```mermaid
flowchart LR
    A[ユーザーの入力（プロンプト）] --> Loop

    subgraph Loop["ループ処理"]
        direction TB
        B["生成AIがツール実行要否を判断"] --> C["ツール実行を要求"]
        C --> D["ツール実行"]
        D --> E["ツール実行結果を返却"]
        E --> B
    end
    
    Loop --> F[最終回答]
```

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

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

```python
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にも対応しています。

```python
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はすでにドキュメントに記載があります。はやい！

https://langfuse.com/docs/integrations/strands-agents


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

```python
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](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/2d2b47de-50d5-4ed6-b3d1-0cae575d2a51.png)


[ここ](https://github.com/strands-agents/sdk-python/pull/37)でプルリク上げてますので、マージされるのをお待ち下さい。