BedrockのCode interpretationを詳細解説（マネコン、SDK、Streamlit化してトレースと生成コードも出力）

AWS Summit New Yorkのタイミングで追加されたBedrockの新機能の一つ、「Code interpretation」をガッツリDeep Diveします。

その他のアップデートはこちらをご参照ください。

BedrockのCode interpretationとは

いわゆるCode Interpreterです。

公式ドキュメントでは、以下のように表現されています。

Code interpretationにより、エージェントは安全なテスト環境でアプリケーション コードを生成、実行、およびトラブルシューティングできます。Code interpretationを使用すると、エージェントの基盤モデルを使用して、基本的な機能を実装するためのコードを生成し、生成 AI アプリケーションの構築に集中できます。

Bedrockの機能というよりは、Agents for Amazon Bedrockの一部という位置づけです。

マネジメントコンソールでやってみる

Agents for Amazon Bedrockにオプションが追加されています。「Code Interpreter」の設定を有効にするだけです。

24/7/11時点で、ドキュメントでは「Code interpretation」、マネジメントコンソールでは「Code Interpreter」と表記が揺れています。

チャットメッセージを入力する際に、ファイルの添付を行います。

「添付ファイルをグラフにして」のように依頼すると、この通り、グラフにして表示してくれます。

いい！！！

SDKで呼び出す

Boto3は1.34.143以降でCode interpretationに対応しています。
リクエスト送信時にファイルを添付します。useCaseとしてCODE_INTERPRETERを指定します。

import boto3
import uuid

client = boto3.client("bedrock-agent-runtime")

with open("Data.xlsx", mode="rb") as f:
    source_file = f.read()

response = client.invoke_agent(
    agentId="ENGSPYXEUI",
    agentAliasId="TSTALIASID",
    sessionId=str(uuid.uuid4()),
    inputText="添付ファイルをグラフにしてください。グラフにする際はX軸をDateとし、昇順にソートしたうえでグラフ化してください",
    sessionState={
        "files": [
            {
                "name": "Data.xlsx",
                "source": {
                    "byteContent": {
                        "data": source_file,
                        "mediaType": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                    },
                    "sourceType": "BYTE_CONTENT",
                },
                "useCase": "CODE_INTERPRETER",
            }
        ]
    },
)

レスポンスはストリーミング形式です。「files」の中に生成したファイルが格納され、「chunk」の中に生成したテキストが格納されます。

for event in response.get("completion"):
    if "files" in event:
        files = event["files"]["files"]
        for file in files:
            with open(file["name"], mode="wb") as f:
                f.write(file["bytes"])

    if "chunk" in event:
        chunk = event["chunk"]
        print(chunk["bytes"].decode())

添付のExcelファイルのデータを使って、X軸をDateとし、Y軸をValueとするグラフを作成しました。ご要望通り、Dateを昇順にソートした上でグラフ化しています。生成されたグラフは graph.png という名前で保存されています。

トレース情報付きでStreamlitにする

「invoke_agent」APIには「enableTrace」というパラメーターがあり、「True」にすることで、マネジメントコンソールでテスト実行した際の右側のエリアのようにトレース情報を取得可能です。

せっかくなので、StreamlitでGUIをつけたアプリにしました。

Streamlitのソースコードはこちら（折りたたみ表示）

# Pyhton外部モジュールのインポート
import uuid

import boto3
import streamlit as st

# タイトル
st.title("Bedrock Code Interpreter")

# Bedrock Agent Runtime クライアント
if "client" not in st.session_state:
    st.session_state.client = boto3.client("bedrock-agent-runtime")
client = st.session_state.client

# セッションID
if "session_id" not in st.session_state:
    st.session_state.session_id = str(uuid.uuid4())
session_id = st.session_state.session_id

# メッセージ
if "messages" not in st.session_state:
    st.session_state.messages = []
messages = st.session_state.messages

# ファイルの添付
uploaded_file = st.file_uploader("files", accept_multiple_files=True)


# アップロードファイルをCode Interpreterのフォーマットに変換
def get_files():
    files = []
    for file in uploaded_file:
        files.append(
            {
                "name": file.name,
                "source": {
                    "byteContent": {"data": file.getvalue(), "mediaType": file.type},
                    "sourceType": "BYTE_CONTENT",
                },
                "useCase": "CODE_INTERPRETER",
            }
        )
    return files


# チャット入力欄を定義
if prompt := st.chat_input("何でも聞いてください。"):
    # ユーザーの入力をメッセージに追加
    messages.append({"role": "human", "text": prompt})

    # ユーザーの入力を画面表示
    with st.chat_message("user"):
        st.markdown(prompt)

    response = client.invoke_agent(
        agentId="ENGSPYXEUI",
        agentAliasId="TSTALIASID",
        sessionId=session_id,
        enableTrace=True,
        inputText=prompt,
        sessionState={"files": get_files()},
    )

    with st.chat_message("assistant"):
        for event in response.get("completion"):
            if "trace" in event:
                if "orchestrationTrace" in event["trace"]["trace"]:
                    orchestrationTrace = event["trace"]["trace"]["orchestrationTrace"]

                    if "modelInvocationInput" in orchestrationTrace:
                        with st.expander("modelInvocationInput", expanded=False):
                            st.write(orchestrationTrace)

                    if "rationale" in orchestrationTrace:
                        with st.expander("rationale（根拠）", expanded=True):
                            st.write(orchestrationTrace)

                    if "invocationInput" in orchestrationTrace:
                        if (
                            "codeInterpreterInvocationInput"
                            in orchestrationTrace["invocationInput"]
                            and "code"
                            in orchestrationTrace["invocationInput"][
                                "codeInterpreterInvocationInput"
                            ]
                        ):
                            code = orchestrationTrace["invocationInput"][
                                "codeInterpreterInvocationInput"
                            ]["code"]

                            with st.expander("code", expanded=False):
                                st.write(orchestrationTrace)

                            with st.expander("Python Code", expanded=True):
                                st.markdown(f"```\n{code}\n```")
                        else:
                            with st.expander("invocationInput", expanded=False):
                                st.write(orchestrationTrace)

                    if "observation" in orchestrationTrace:
                        with st.expander("observation（観察）", expanded=True):
                            st.write(orchestrationTrace)

            if "files" in event:
                files = event["files"]["files"]
                for file in files:

                    with open(file["name"], mode="wb") as f:
                        f.write(file["bytes"])

                    st.image(file["bytes"], caption=file["name"])

            if "chunk" in event:
                chunk = event["chunk"]
                answer = chunk["bytes"].decode()

                st.write(answer)
                messages.append({"role": "assistant", "text": answer})

とりあえず動きを先に見てみましょう。
使用しているモデルはClaude 3 Sonnetです。

トレースを出力しながら、画像と最終回答を生成する様子が確認できました。
トレースの内容を追いかけてみましょう。

トレースにも複数種類ありますが、今回は「OrchestrationTrace」のみを対象とします。

詳細はドキュメントをご確認ください。

• modelInvocationInput
  生成AIに渡すインプット情報

• rationale
  インプットに対して、エージェント（生成AI）が考えた結果
  「コードを書いてみましょう」と考えたので、次のステップでコードを生成します。

  

• codeInterpreterInvocationInput
  コードを生成する部分
  ここにPythonのコードが含まれます

  

  他のプロンプトに混ざってではなく、コード単体で返却される作りのようです。そのため、この部分を抽出することで、コードを流用することも可能です。

  

  頑張ればClaudeのArtifactsみたいなことができるかも？？

• observation
  これまでの手順を受けて、ユーザーのリクエストに答えられたかを確認します

  

無事、画像も生成されています。

このあと、再度modelInvocationInput、rationale、observationと一巡した後、最終回答が出力されました。
実行するたびに、処理の流れは変わるようでした。

使い道がすごく広がりそうです！