chakokuのブログ(rev4)

テック・コミック・DTM・・・ごくまれにチャリ

某学校様向け研修教材;マイコンだって生成AI(仮)

背景:ある学校様向けにマイコンプログラミングの研修を行うことになり教材を検討中・・・Lチカから始めて、I2C、シリアル接続ときて、WebAPIに進み、生成AIの活用の流れで授業を進めたい
取り組み:マイコンから直接生成AIのAPIに繋ぐのは認証等が大変なので、AWSでWebAPIを立てて、それ経由でAmazon Bedrockに接続する
課題:Bedrockがよく分かっていないので試作する
結論:県民あるあるまでは仕上いないが、Token付きのWebAPIでhello worldを回答してくれるところまでは実装
詳細:

まずは生成AIに聞いてみる。以下質問

AWSの生成AIを使って、簡単なクイズを作成してもらいたいと思います。
この場合、AWSのどのようなAIを使うのが良いでしょうか? 
またコストを抑えたいので、小規模言語モデルで十分です

生成AIの回答

上記より、Nova Microで良かろうと。できたらGUIでコチコチやらず、Infra as Codeで構築したい。でまた質問

Infra as a Codeとして、Codeで構築させたいのですが可能ですか?
今回の構成(API Gateway → Lambda → Bedrock)をInfrastructure as Code(IaC)で
構築することは可能です。以下に、AWS CDK(Python版)を使った最小構成の例を紹介します:

bedrock_quiz_api/
├── app.py
├── requirements.txt
└── bedrock_quiz_api/
    ├── __init__.py
    ├── bedrock_quiz_api_stack.py
    └── lambda/
        └── handler.py

CDK Stack

from aws_cdk import (
    Stack,
    aws_lambda as _lambda,
    aws_apigatewayv2 as apigw,
    aws_apigatewayv2_integrations as integrations,
    aws_iam as iam,
)
from constructs import Construct

class BedrockQuizApiStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)

        # Lambda関数
        fn = _lambda.Function(
            self, "QuizLambda",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="handler.lambda_handler",
            code=_lambda.Code.from_asset("bedrock_quiz_api/lambda")
        )

        # IAMロールにBedrockアクセス許可
        fn.add_to_role_policy(iam.PolicyStatement(
            actions=["bedrock:InvokeModel"],
            resources=["*"]
        ))

        # HTTP API Gateway
        api = apigw.HttpApi(
            self, "QuizApi",
            default_integration=integrations.HttpLambdaIntegration("QuizIntegration", fn)
        )

        # エンドポイント出力
        self.api_url = api.url

Lambda関数

import json
import boto3

def lambda_handler(event, context):
    client = boto3.client("bedrock-runtime", region_name="ap-northeast-1")
    prompt = "日本の首都はどこですか?選択肢と正解を含めてクイズ形式で答えてください。"
    
    response = client.invoke_model(
        modelId="amazon.titan-text-micro-v1",
        contentType="application/json",
        accept="application/json",
        body=json.dumps({"inputText": prompt})
    )
    
    result = json.loads(response['body'].read())
    return {
        "statusCode": 200,
        "body": json.dumps(result)
    }

デプロイ手順

cd bedrock_quiz_api
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cdk bootstrap
cdk deploy

CDK以外にTerraformもあるよと言ってきたので、どっちを使うのが良いかを質問。回答は以下

PythonでBedrockを呼び出すLambdaを中心に構成するなら、CDKが自然で効率的です。 
ただし、インフラ構成を明快に共有・管理したい場合はTerraformも非常に有力です。
もし「CDKでまず構築 → Terraformで再構成して比較」するような検証も面白いと思います。

すんなりIaCで構築できるとは到底思えないが、おぼろげにやること理解したので、なるべくコードで構築してみる。あるいは、最初は手作業で試作して、最後はIaCで検証か。 毎月AWS代を払っているが、いろいろアカウント作ったし、どこのIDが生きているのかあいまい。
セキュリティの事なので詳しくは書けないが、、、自分自身はアカウント曖昧だが、とある誰かが覚えてくれていて、これですか?と言ってきたので、それを使ってみた。なお、2FAにはしているので、、2FAも入れる。なんとかコンソールに入れた。
使おうとしているNova Micro

よく分からないが、利用申請が必要らしい


Nova Microに県民あるあるを出題してください。と依頼してみた

県民あるあるで、A/Bの2択のクイズを考えています。各県で、あるあるの出題例はありますか?
正解も一緒に出題してください

出題例は以下

Q: 北海道の人は冬に何をよくしますか?
A) スキーをする / B) 温泉に行く
正解: A) スキーをする
北海道はスキーの聖地として知られており、冬のスポーツとしてスキーが人気です。

Q: 青森県の人はりんごを何によく使いますか?
A) ジュースにする / B) 焼酎の原料にする
正解: B) 焼酎の原料にする
青森県はりんごの生産量が日本一で、そのりんごを原料とした焼酎が有名です。

質問に深みは無いがまぁ問としてはできている。
戦国武将あるあるを作ってみてもらった

Q: 織田信長はどんなことをよくしていましたか?
A) 茶道を楽しむ/B) 火攻をよく使った

正解: B) 火攻をよく使った
織田信長は火攻を戦略的に使用することで、多くの城を攻略しました。

織田信長は茶道もやっていたと思うのだが、、、まあここはお遊びという感じか・・CopilotとかChatGPTだともっと深みのあるあるあるを言ってくるのだろうか。
Copilotに作ってもらった県民あるある

大阪府民がエスカレーターに乗るとき、どちら側に立つのが一般的?
A: 左側/B: 右側

愛知県民がよく食べる独特な味噌はどれ?
A: 赤味噌/B: 白味噌

沖縄県民が親しみを持つ飲み物として有名なものは?
A: さんぴん茶/B: 抹茶

Copilitの方が面白いような気が・・・ これはモデルのサイズの違いなのか、たまたまなのか。。
boto3で呼び出すコードは以下となるらしい

import boto3
import json

client = boto3.client("bedrock-runtime", region_name="us-east-1")

payload = {
    "inputText": "大阪府民がよく使う方言は?"
}

response = client.invoke_model(
    modelId="amazon.nova-micro-v1:0",
    contentType="application/json",
    accept="application/json",
    body=json.dumps(payload)
)

result = json.loads(response['body'].read())
print(result)

AWSが提示するサンプルは以下(
https://docs.aws.amazon.com/bedrock/latest/userguide/bedrock-runtime_example_bedrock-runtime_Converse_AmazonNovaText_section.html
)

# Use the Conversation API to send a text message to Amazon Nova.

import boto3
from botocore.exceptions import ClientError

# Create a Bedrock Runtime client in the AWS Region you want to use.
client = boto3.client("bedrock-runtime", region_name="us-east-1")

# Set the model ID, e.g., Amazon Nova Lite.
model_id = "amazon.nova-lite-v1:0"

# Start a conversation with the user message.
user_message = "Describe the purpose of a 'hello world' program in one line."
conversation = [
    {
        "role": "user",
        "content": [{"text": user_message}],
    }
]

try:
    # Send the message to the model, using a basic inference configuration.
    response = client.converse(
        modelId=model_id,
        messages=conversation,
        inferenceConfig={"maxTokens": 512, "temperature": 0.5, "topP": 0.9},
    )

    # Extract and print the response text.
    response_text = response["output"]["message"]["content"][0]["text"]
    print(response_text)

except (ClientError, Exception) as e:
    print(f"ERROR: Can't invoke '{model_id}'. Reason: {e}")
    exit(1)

Nova Microはap‑northeast‑1でも使えることを期待
以下のコードでBedrockのNova Microを呼び出してみた

#!/usr/bin/python3

#export AWS_ACCESS_KEY_ID=AKIAxxxxxxxxxxxx
#export AWS_SECRET_ACCESS_KEY=xxxxxxxxxxxxxxxxxxxx
#export AWS_DEFAULT_REGION=ap-northeast-1

REGION = 'ap-northeast-1'
MODEL_ID = 'amazon.nova-micro-v1:0'

import boto3
from botocore.exceptions import ClientError

client = boto3.client("bedrock-runtime", region_name = REGION)
user_message = "Describe the purpose of a 'hello world' program in one line."
conversation = [
    {
        "role": "user",
        "content": [{"text": user_message}],
    }
]

try:
    response = client.converse(
        modelId = MODEL_ID,
        messages = conversation,
        inferenceConfig = {"maxTokens": 512, "temperature": 0.5, "topP": 0.9},
    )
    response_text = response["output"]["message"]["content"][0]["text"]
    print(response_text)
except (ClientError, Exception) as e:
    print(f"ERROR: Can't invoke '{MODEL_ID}'. Reason: {e}")
    exit(1)

すると以下のExceptionが発生

$  ./nova_conversation.py
ERROR: Can't invoke 'amazon.nova-micro-v1:0'. Reason: An error occurred (ValidationException) 
when calling the Converse operation: Invocation of model ID amazon.nova-micro-v1:0 
with on-demand throughput isn’t supported. Retry your request with the ID 
or ARN of an inference profile that contains this model.

Nova Microをon demand throughputで使うのがNGなのか??
Copilitの回答は以下

このメッセージは以下を示しています:
* amazon.nova-micro-v1:0 は 直接呼び出し(on-demand)では使えない
* 推論プロファイル(Inference Profile)を使って呼び出す必要がある

Nova Micro用のInferance Profileが提供されている。言語モデルを直接呼び出さず、Inferance Profileから使いなさいということか。

Inference ProfileのARNは以下。これをBotoで指定しろと。。

arn:aws:bedrock:ap-northeast-1:36**********4:inference-profile/apac.amazon.nova-micro-v1:0

INFERENCEでいろいろトラブったが、AsiaPacific用のモデル(MODEL_ID = 'apac.amazon.nova-micro-v1:0')を指定することで、INFERENCE経由でなくても利用可能なようであった。
多分、リージョンを付けない言語モデルはUSかどこかで動いていて、INFERENCE経由でルーティングしてもらう必要があるのではなかろうか。

#!/usr/bin/python3

#export AWS_ACCESS_KEY_ID=AKIAxxxxxxxxxxxx
#export AWS_SECRET_ACCESS_KEY=xxxxxxxxxxxxxxxxxxxx
#export AWS_DEFAULT_REGION=ap-northeast-1

REGION = 'ap-northeast-1'
MODEL_ID = 'apac.amazon.nova-micro-v1:0'

import boto3
from botocore.exceptions import ClientError

client = boto3.client("bedrock-runtime", region_name = REGION)
user_message = "Describe the purpose of a 'hello world' program in one line."
conversation = [
    {
        "role": "user",
        "content": [{"text": user_message}],
    }
]

try:
    response = client.converse(
        modelId = MODEL_ID,
        messages = conversation,
        inferenceConfig = {
             "maxTokens": 512, 
             "temperature": 0.5, "topP": 0.9,
        }
    )
    response_text = response["output"]["message"]["content"][0]["text"]
    print(response_text)

except (ClientError, Exception) as e:
    print(f"ERROR: Can't invoke '{MODEL_ID}'. Reason: {e}")
    exit(1)

上記boto3でBedlockを呼び出すと以下が返却された。

The purpose of a 'hello world' program is to demonstrate the basic syntax 
and functionality of a new programming language or environment.

Boto3でAP(Tokyo)の環境で動いたのでこれで呼び出しは良しとして、これをLambdaで動かすようにして、WebAPIから叩けるようにすればよいと
IaCで作りたいと言っておきながら、トラブル多発なので、、GUIでシコシコ作って最後にIaCで清書するぐらいでないと最初からIaCで構築しようとしても、そもそもやり方間違っているのかIaCの書き方が間違っているのか切り分けも難しい。だから、、IaCって完成形が分かっているアーキテクチャでないとエラー無しに動かせるまでえらく時間がかかる気がするのでした。

まずはグサグサでもいいから動かしたいので、以下のコードをLambdaで定義した。そしてBedrockに対する権限を与えた。
test 機能を使ってLabmdaを呼び出したところ、いろんな言語によるhello worldのサンプルが返却された。

import json
import boto3
from botocore.exceptions import ClientError

REGION = 'ap-northeast-1'
MODEL_ID = 'apac.amazon.nova-micro-v1:0'

def taik_to_bedlock(query_text):

    response_text = None
    client = boto3.client("bedrock-runtime", region_name = REGION)
    conversation = [
    {
        "role": "user",
        "content": [{"text": query_text}],
    }
    ]

    try:
        response = client.converse(
            modelId = MODEL_ID,
            messages = conversation,
            inferenceConfig = {
              "maxTokens": 512, 
                 "temperature": 0.5, "topP": 0.9,
            }
        )
        response_text = response["output"]["message"]["content"][0]["text"]
        print(response_text)

    except (ClientError, Exception) as e:
        print(f"ERROR: Can't invoke '{MODEL_ID}'. Reason: {e}")
        exit(1)
    return response_text

def lambda_handler(event, context):
    response_text = taik_to_bedlock('hello world')
    return {
        'statusCode': 200,
        'body': json.dumps(response_text)
    }

Lambdaが動いたので、次はWebAPI化すると
API Gatewayを使うのが一般的らしい(昔試した記憶が・・・)
チュートリアル: API Gateway で Lambda を使用する - AWS Lambda

マニュアルとか読まず、過去にやった操作を思い出しながらGUIで適当に操作して、HTTP用のEndPointを作って、上記Lambdaにルーティングした。

今はテストなので、認証等を付けておらず、curlで呼び出すと以下の応答が来た。だからまぁ動いているのだろうと

$ curl https://t8******k.execute-api.ap-northeast-1.amazonaws.com/test/talk-nova
"Hello! If you're looking to create a simple \"Hello, World!\" program, here are examples in a few popular programming languages:\n\n### Python\n```python\nprint(\"Hello, World!\")\n```\n\n### JavaScript\n```javascript\nconsole.log(\"Hello, World!\");\n```\n\n### Java\n```java\npublic class HelloWorld {\n    public static void main(String[] args) {\n        System.out.println(\"Hello, World!\");\n    }\n}\n```\n\n### C\n```c\n#include <stdio.h>\n\nint main() {\n    printf(\"Hello, World!\\n\");\n    return 0;\n}\n```\n\n### C++\n```cpp\n#include <iostream>\n\nint main() {\n    std::cout << \"Hello, World!\" << std::endl;\n    return 0;\n}\n```\n\n### Ruby\n```ruby\nputs \"Hello, World!\"\n```\n\n### Go\n```go\npackage main\n\nimport \"fmt\"\n\nfunc main() {\n    fmt.Println(\"Hello, World!\")\n}\n```\n\nIf you need help with something specific or have any questions about these examples, feel free to ask!"

Token検証用Lambdaを作り

def lambda_handler(event, context):
    print("Received event:", event)

    #token = event.get("authorization")
    token = event["headers"]["authorization"]
    method_arn = event.get("routeArn")

    print(token)
    print(method_arn)
    if token == "Bearer token********":
        return generate_policy("user", "Allow", method_arn)
    elif token == "Bearer deny-token":
        return generate_policy("user", "Deny", method_arn)
    else:
        raise Exception("Unauthorized")

def generate_policy(principal_id, effect, resource):
    if effect not in ("Allow", "Deny"):
        raise ValueError("Effect must be 'Allow' or 'Deny'")

    return {
        "principalId": principal_id,
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Action": "execute-api:Invoke",
                    "Effect": effect,
                    "Resource": resource
                }
            ]
        }
    }

API Gatewayトークン検証を紐づける

Bearerトークンを付けて呼び出してみる

$ curl -X GET   -H "Authorization: Bearer token-******"   https://t833wzqcak.execute-api.ap-northeast-1.amazonaws.com/talk-nova
"Hello! If you're looking to create a simple \"Hello, World!\" program, here are examples in several popular programming languages:\n\n### Python\n```python\nprint(\"Hello, World!\")\n```\n\n### JavaScript\n```javascript\nconsole.log(\"Hello, World!\");\n```\n\n### Java\n```java\npublic class HelloWorld {\n    public static void main(String[] args) 

トークンを変えるとエラーになるのでトークンの検証が正しく行われていると判断
ただ、下記のように、event.get('authorization')でBearerTokenが取得できるはずが、authorizationには格納されておらず、WorkAroundとして、元から入っている、headers配下のauthorizationを取ってくるように修正

    #token = event.get("authorization")
    token = event["headers"]["authorization"]

今回試作した、生成AIを呼び出すWebAPIサービス(県民あるあるクイズ用)

残作業
hello worldでqueryしているので、query を以下に変更必要

  • 県民あるあるの問題を考えてください
  • 答えは、a/bの2択とする
  • 問と答えはカタカナで、最大14文字
  • 正解も付けて
  • JSON形式で出題して

県民アルアルを答えてくれるようにプロンプトを変えてみた。返ってきたクイズ案は以下

"{
 "Q" : "シズオカケンミンノヒミツニチャクメル",
 "A1": "Aノショウガモチゾウ",
 "A2": "Bノショウガモチゾウ"
}"

ちょっと意味分からないのだが。。しかも選択肢(a),(b)が同じという。。。
生成モデルをケチりすぎているのが問題だろうか(amazon.nova-microを使用)
言語モデルをClaude Sonnet 4にすると以下の回答。すぐに分かりそうなアルアルだけど、形式上は問題がない。

{
  "question": "シズオカケンミンガヨクイワレルコト?",
  "choices": {
    "A": "オチャガスキ",
    "B": "オンダンナノ"
  },
  "answer": "A",
  "explanation": "静岡県民あるあるとして「お茶が好き」と言われることがよく話題になります。静岡茶の産地として有名で、県民は日常的にお茶を飲む習慣があり、給茶機が学校にあることや、お茶の濃さにこだわりがあることなど、お茶に関するあるあるが数多く存在します。"
}