Python ( boto3 ) からAgents for Amazon Bedrockを呼び出す

前回作成したAgentをPythonから呼び出してみます。

エイリアスの作成

APIのI/Fを見るとagentAliasIdが必要なようなので、マネコンの該当AgentからCreate Aliasでエイリアスを作成します。

参考資料

プログラム

マネコンからAgentのIDと、Alias IDをメモしておきます。
※作成時に指定したAlias nameではないので注意

上記の参考資料を見ながらプログラムを書きます。

agentforbedrock.py

import uuid
import boto3

# Agentへの入力テキスト
input_text:str = "Bedrockで使用可能なモデルは？"

# Agentの定義
agent_id:str = 'XXXXXXXXXX' # ご自身のAgentのID
agent_alias_id:str = 'XXXXXXXXXX' # ご自身のAgentのAlias ID
session_id:str = str(uuid.uuid1()) # 乱数

# Clientの定義
client = boto3.client("bedrock-agent-runtime")

# Agentの実行
response = client.invoke_agent(
    inputText=input_text,
    agentId=agent_id,
    agentAliasId=agent_alias_id,
    sessionId=session_id,
    enableTrace=False
)

# Agent実行結果の取得
event_stream = response['completion']
for event in event_stream:        
    if 'chunk' in event:
        data = event['chunk']['bytes'].decode("utf-8")
        print(data)

実行結果

Bedrockで使用可能なモデルには、Amazonが提供するTitanテキストモデルやTitan画像モデルなどと、
Anthropic、Cohere、Hugging Faceなどのサードパーティモデルプロバイダーのモデルがあります。

相変わらず正確性が微妙なRAGになっていますが、ともあれAgent経由でKnowledge baseの検索が行えました。
boto3的にはAgentを呼び出しているのみなので、Agentの中の処理（今回はKnowledge baseのRAG)やLLMは隠蔽されています。

Streamlitでガワを被せるとこんな感じです。

agentforbedrock2.py

import uuid
import boto3

# Agentの定義
agent_id:str = 'XXXXXXXXXX' # ご自身のAgentのID
agent_alias_id:str = 'XXXXXXXXXX' # ご自身のAgentのAlias ID
session_id:str = str(uuid.uuid1()) # 乱数

# Clientの定義
client = boto3.client("bedrock-agent-runtime")

# Streamlitでガワを被せる
import streamlit as st

st.title("Bedrock AgentからのKnowledge base呼出し")
input_text = st.text_input("このテキストをAgentに送信します")
send_button = st.button("送信")


if send_button:
    result_area = st.empty()
    text = ''

    # Agentの実行
    response = client.invoke_agent(
        inputText=input_text,
        agentId=agent_id,
        agentAliasId=agent_alias_id,
        sessionId=session_id,
        enableTrace=False
    )

    # Agent実行結果の取得
    event_stream = response['completion']
    for event in event_stream:        
        if 'chunk' in event:
            text += event['chunk']['bytes'].decode("utf-8")
            result_area.write(text)