Amazon Bedrock+ Anthropic Claude 3 Sonnetでストリーミングテキストを生成する

はじめに

Amazon BedrockでAnthropic Claude 3 Sonnetが利用可能となったので､AWSのドキュメントを参考にストリーミングテキストを生成するサンプルコードを作成してみました。

参考情報

デモ動画

環境構築

以下のリソースを作成･削除･変更できる権限をもつAWSユーザーを利用すること

• Amazon Bedrock
  • Bedrockの画面でAnthropic Claude 3 Sonneのモデルアクセス有効化が必要です｡

開発環境構築

作業環境のOSバージョン

Windows 11上のWSLでUbuntu 23.04を動かしています。

$ cat /etc/os-release | grep PRETTY_NAME
PRETTY_NAME="Ubuntu 23.04"

Python環境

$ python3 --version
Python 3.12.0

$ python3 -m venv .venv
source .venv/bin/activate

$ pip3 install --upgrade pip
$ pip3 --version
pip 24 from /home/xxx/.venv/lib/python3.12/site-packages/pip (python 3.12)

$ pip install boto3

サンプルコード

Anthropic Claude Messages API の Streaming multimodal prompt with InvokeModelWithResponseStream をもとにしました。ここではimageは使用せずtextのみ扱っています。

実行

$ python3 ./stream.py

実行環境にもよるかもしれませんが、最初の文が出力されるまで15秒ほどかかりました。

stream.py

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
#
"""
Shows how to generate a message with Anthropic Claude (on demand).
"""
import json
import logging

import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


def generate_message(bedrock_runtime, model_id, system_prompt, messages, max_tokens):
    """
    Streams the response from a multimodal prompt.
    Args:
        bedrock_runtime: The Amazon Bedrock boto3 client.
        model_id (str): The model ID to use.
        system_prompt (str) : The system prompt text.
        messages (JSON) : The messages to send to the model.
        max_tokens (int) : The maximum  number of tokens to generate.
    Returns:
        None.
    """

    body = json.dumps(
        {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": max_tokens,
            "temperature": 1,  # default value is 1
            "top_p": 0.999,    # default value is 0.999
            "system": system_prompt,
            "messages": messages
        }
    )

    response = bedrock_runtime.invoke_model_with_response_stream(
        body=body,
        modelId=model_id)

    for event in response.get("body"):
        chunk = json.loads(event["chunk"]["bytes"])

        if chunk['type'] == 'message_delta':
            print(f"\nStop reason: {chunk['delta']['stop_reason']}")
            print(f"Stop sequence: {chunk['delta']['stop_sequence']}")
            print(f"Output tokens: {chunk['usage']['output_tokens']}")
        if chunk['type'] == 'content_block_delta':
            if chunk['delta']['type'] == 'text_delta':
                print(chunk['delta']['text'], end="")


def main():
    """
    Entrypoint for Anthropic Claude message example.
    """

    try:

        bedrock_runtime = boto3.client(service_name='bedrock-runtime')

        model_id = 'anthropic.claude-3-sonnet-20240229-v1:0'
        system_prompt = """
あなたはチャットでユーザを支援するAIアシスタントです｡

* Let's think step by step
* Take a deep breath
* 回答のルールが<rules></rules>にあります｡ ルールを理解してください｡ルールは必ず守ってください｡例外はありません｡
* 「〜って何？」「〜とは？」「〜を説明して」というような概要を聞く質問については、「〜の概要」と読み替えてください｡

<rules>
 * 日本語で質問された場合、必ず日本語で回答してください｡
 * 質問に具体性がなく回答できない場合は、質問の仕方をアドバイスしてください｡､
</rules>
"""

        user_prompt = """
火星で人類が暮らす未来についてエッセイを日本語で2000語で書いてください。
エッセイには以下の内容を含めてください。

- 火星で生活するための課題や解決策
- 火星での生活のメリットとデメリット
- 火星での生活の様子
- 火星での生活が地球での生活とどう違うのか

最後に､あなたの考えを述べてください。
"""

        max_tokens = 5000
        user_message = {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": user_prompt
                }
            ]
        }
        messages = [user_message]

        generate_message(
            bedrock_runtime, model_id, system_prompt, messages, max_tokens)

    except ClientError as err:
        message = err.response["Error"]["Message"]
        logger.error("A client error occurred: %s", message)
        print("A client error occured: " + format(message))


if __name__ == "__main__":
    main()