Amazon Bedrock ガードレールでゲーム内のキャラクターを彩ろう！

from aws_cdk import (
    Duration,
    Stack,
    aws_lambda as _lambda,
    # api gateway
    aws_apigateway as _apigw,
    aws_iam as iam,
    aws_logs as logs,
    # aws_sqs as sqs,
)
from constructs import Construct

class CdkStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Lambda 実行ロールを定義
        lambda_role = iam.Role(
            self, "LambdaExecutionRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSLambdaBasicExecutionRole")
            ],
            inline_policies={
                "BedrockAccess": iam.PolicyDocument(
                    statements=[
                        iam.PolicyStatement(
                            actions=[
                                "bedrock:InvokeModel"
                            ],
                            resources=["*"]  # You might want to restrict this to specific model ARNs
                        )
                    ]
                )
            }
        )

        # Lambda を定義
        lambda_function = _lambda.Function(
            self, "Lambda",
            runtime=_lambda.Runtime.PYTHON_3_9,
            code=_lambda.Code.from_asset("lambda"),
            handler="index.handler",
            role=lambda_role,
            environment={
                "BEDROCK_MODEL_ID": "us.amazon.nova-premier-v1:0",  # Replace with your custom model ID
            },
            timeout=Duration.seconds(30)  # Increased timeout for model inference
        )

        # API Gateway のログ用 IAM ロールを定義
        api_logging_role = iam.Role(
            self, "ApiGatewayLoggingRole",
            assumed_by=iam.ServicePrincipal("apigateway.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonAPIGatewayPushToCloudWatchLogs")
            ]
        )

        # API Gateway を定義
        api = _apigw.RestApi(
            self, "Apigw",
            deploy_options=_apigw.StageOptions(
                stage_name="prod",
                logging_level=_apigw.MethodLoggingLevel.INFO,  # ログレベルを INFO に設定
                access_log_destination=_apigw.LogGroupLogDestination(
                    logs.LogGroup(self, "ApiGatewayAccessLogs", retention=logs.RetentionDays.ONE_WEEK)
                ),
                access_log_format=_apigw.AccessLogFormat.clf()  # Common Log Format を使用
            ),
            cloud_watch_role=True  # CloudWatch ロールを有効化
        )
        
        # alpaca リソースを作成
        alpaca = api.root.add_resource("alpaca")
        
        # GET メソッドを追加し、text パラメータを設定
        alpaca.add_method(
            "GET",
            _apigw.LambdaIntegration(lambda_function),
            request_parameters={
                "method.request.querystring.text": True  # Required parameter
            }
        )

import json
import boto3
import os

# Initialize Bedrock client
bedrock = boto3.client('bedrock-runtime')

# Get the model ID from environment variable or use a default
MODEL_ID = os.environ.get('BEDROCK_MODEL_ID', 'us.amazon.nova-premier-v1:0')  # Cross-region inference profile
GUARDRAIL_ID = os.environ.get('GUARDRAIL_ID', 'GUARDRAIL_ID')  # Guardrail ID
GUARDRAIL_VERSION = os.environ.get('GUARDRAIL_VERSION', '1')  # Guardrail version

def handler(event, context):
    print(f"Lambda function started. Event: {json.dumps(event)}")
    print(f"Context: {context}")
    
    try:
        print("Step 1: Getting query string parameters")
        # Get the query string parameters, defaulting to empty dict if None
        query_params = event.get('queryStringParameters') or {}
        print(f"Query params: {query_params}")
        
        # Get the text parameter
        input_text = query_params.get('text')
        print(f"Text parameter: {input_text}")
        
        # Since text is required in API Gateway config, return 400 if missing
        if input_text is None:
            print("Text parameter is missing, returning 400")
            return {
                'statusCode': 400,
                'headers': {
                    'Content-Type': 'application/json'
                },
                'body': json.dumps({
                    'error': 'Missing required query parameter: input_text'
                })
            }
        
        predefined_instruction = "あなたは賢明なアルパカのキャラクターです。理工学分野の博士号を持っており、常に冷静に質問に回答します。回答は 2-3 文で、「です」「ます」は使いません。全ての文の語尾は必ず「パカ」とし、質問では語尾が「パカ？」になります。また、口癖として、「思う」「考える」などの動詞の代わりに「おもんパカる」を頻繁に使います。「おもんパカる」を使う場合も語尾は「パカ」になります。\n\n"

        text = predefined_instruction + input_text
        
        print("Step 2: Preparing Bedrock request")
        # Prepare the request body for Bedrock (Nova Premier uses Messages API)
        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "text": text
                    }
                ]
            }
        ]
        
        print("Step 3: Calling Bedrock")
        # Call Bedrock with Guardrails
        converse_params = {
            'modelId': MODEL_ID,
            'messages': messages
        }
        
        response = bedrock.converse(**converse_params)
        print(f"Bedrock response received: {type(response)}")
        
        print("Step 4: Processing streaming response")
        # Process the Converse API response
        model_response = ""
        if 'output' in response and 'message' in response['output']:
            content = response['output']['message']['content']
            for item in content:
                if 'text' in item:
                    model_response += item['text']
        
        print(f"Complete model response: {model_response}")
        
        print("Step 5: Returning success response")
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps({
                'text': model_response
            })
        }
        
    except Exception as e:
        print(f"ERROR: Exception occurred: {str(e)}")
        print(f"ERROR: Exception type: {type(e)}")
        import traceback
        print(f"ERROR: Traceback: {traceback.format_exc()}")
        return {
            'statusCode': 500,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps({
                'error': str(e)
            })
        }

import json
import boto3
import os

# Initialize Bedrock client
bedrock = boto3.client('bedrock-runtime')

# Get the model ID from environment variable or use a default
MODEL_ID = os.environ.get('BEDROCK_MODEL_ID', 'us.amazon.nova-premier-v1:0')  # Cross-region inference profile
GUARDRAIL_ID = os.environ.get('GUARDRAIL_ID', 'GUARDRAIL_ID')  # Guardrail ID
GUARDRAIL_VERSION = os.environ.get('GUARDRAIL_VERSION', '1')  # Guardrail version

def handler(event, context):
    print(f"Lambda function started. Event: {json.dumps(event)}")
    print(f"Context: {context}")
    
    try:
        print("Step 1: Getting query string parameters")
        # Get the query string parameters, defaulting to empty dict if None
        query_params = event.get('queryStringParameters') or {}
        print(f"Query params: {query_params}")
        
        # Get the text parameter
        input_text = query_params.get('text')
        print(f"Text parameter: {input_text}")
        
        # Since text is required in API Gateway config, return 400 if missing
        if input_text is None:
            print("Text parameter is missing, returning 400")
            return {
                'statusCode': 400,
                'headers': {
                    'Content-Type': 'application/json'
                },
                'body': json.dumps({
                    'error': 'Missing required query parameter: input_text'
                })
            }
        
        predefined_instruction = "あなたは賢明なアルパカのキャラクターです。理工学分野の博士号を持っており、常に冷静に質問に回答します。回答は 2-3 文で、「です」「ます」は使いません。全ての文の語尾は必ず「パカ」とし、質問では語尾が「パカ？」になります。また、口癖として、「思う」「考える」などの動詞の代わりに「おもんパカる」を頻繁に使います。「おもんパカる」を使う場合も語尾は「パカ」になります。\n\n"

        text = predefined_instruction + input_text
        
        print("Step 2: Preparing Bedrock request")
        # Prepare the request body for Bedrock (Nova Premier uses Messages API)
        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "text": text
                    }
                ]
            }
        ]
        
        print("Step 3: Calling Bedrock")
        # Call Bedrock with Guardrails
        converse_params = {
            'modelId': MODEL_ID,
            'messages': messages
        }
        
        # Add Guardrails if configured
        if GUARDRAIL_ID:
            answer = bedrock.apply_guardrail(
                guardrailIdentifier=GUARDRAIL_ID,
                guardrailVersion=GUARDRAIL_VERSION,
                source='INPUT',
                content=[
                    {
                        'text': {
                            'text': input_text,
                            'qualifiers': [
                                'guard_content'
                            ]
                        }
                    },
                ]
            )

            print(f"Using Guardrail: {GUARDRAIL_ID} version {GUARDRAIL_VERSION}")

            if len(answer['outputs']) > 0:
                return {
                    'statusCode': 200,
                    'headers': {
                        'Content-Type': 'application/json'
                    },
                    'body': json.dumps({
                        'text': answer['outputs'][0]['text']
                    })
                }
        
        response = bedrock.converse(**converse_params)
        print(f"Bedrock response received: {type(response)}")
        
        print("Step 4: Processing streaming response")
        # Process the Converse API response
        model_response = ""
        if 'output' in response and 'message' in response['output']:
            content = response['output']['message']['content']
            for item in content:
                if 'text' in item:
                    model_response += item['text']
        
        print(f"Complete model response: {model_response}")
        
        print("Step 5: Returning success response")
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps({
                'text': model_response
            })
        }
        
    except Exception as e:
        print(f"ERROR: Exception occurred: {str(e)}")
        print(f"ERROR: Exception type: {type(e)}")
        import traceback
        print(f"ERROR: Traceback: {traceback.format_exc()}")
        return {
            'statusCode': 500,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps({
                'error': str(e)
            })
        }

import json
import boto3
import os

# Initialize Bedrock client
bedrock = boto3.client('bedrock-runtime')

# Get the model ID from environment variable or use a default
MODEL_ID = os.environ.get('BEDROCK_MODEL_ID', 'us.amazon.nova-premier-v1:0')  # Cross-region inference profile
GUARDRAIL_ID = os.environ.get('GUARDRAIL_ID', 'GUARDRAIL_ID')  # Guardrail ID
GUARDRAIL_VERSION = os.environ.get('GUARDRAIL_VERSION', '1')  # Guardrail version

def handler(event, context):
    print(f"Lambda function started. Event: {json.dumps(event)}")
    print(f"Context: {context}")
    
    try:
        print("Step 1: Getting query string parameters")
        # Get the query string parameters, defaulting to empty dict if None
        query_params = event.get('queryStringParameters') or {}
        print(f"Query params: {query_params}")
        
        # Get the text parameter
        input_text = query_params.get('text')
        print(f"Text parameter: {input_text}")
        
        # Since text is required in API Gateway config, return 400 if missing
        if input_text is None:
            print("Text parameter is missing, returning 400")
            return {
                'statusCode': 400,
                'headers': {
                    'Content-Type': 'application/json'
                },
                'body': json.dumps({
                    'error': 'Missing required query parameter: input_text'
                })
            }
        
        predefined_instruction = "あなたは賢明なアルパカのキャラクターです。理工学分野の博士号を持っており、常に冷静に質問に回答します。回答は 2-3 文で、「です」「ます」は使いません。全ての文の語尾は必ず「パカ」とし、質問では語尾が「パカ？」になります。また、口癖として、「思う」「考える」などの動詞の代わりに「おもんパカる」を頻繁に使います。「おもんパカる」を使う場合も語尾は「パカ」になります。\n\n"

        text = predefined_instruction + input_text
        
        print("Step 2: Preparing Bedrock request")
        # Prepare the request body for Bedrock (Nova Premier uses Messages API)
        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "text": text
                    }
                ]
            }
        ]
        
        print("Step 3: Calling Bedrock")
        # Call Bedrock with Guardrails
        converse_params = {
            'modelId': MODEL_ID,
            'messages': messages
        }
        
        # Add Guardrails if configured
        if GUARDRAIL_ID:
            converse_params['guardrailConfig'] = {
                'guardrailIdentifier': GUARDRAIL_ID,
                'guardrailVersion': GUARDRAIL_VERSION
            }
            print(f"Using Guardrail: {GUARDRAIL_ID} version {GUARDRAIL_VERSION}")
        
        response = bedrock.converse(**converse_params)
        print(f"Bedrock response received: {type(response)}")
        
        print("Step 4: Processing streaming response")
        # Process the Converse API response
        model_response = ""
        if 'output' in response and 'message' in response['output']:
            content = response['output']['message']['content']
            for item in content:
                if 'text' in item:
                    model_response += item['text']
        
        print(f"Complete model response: {model_response}")
        
        print("Step 5: Returning success response")
        return {
            'statusCode': 200,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps({
                'text': model_response
            })
        }
        
    except Exception as e:
        print(f"ERROR: Exception occurred: {str(e)}")
        print(f"ERROR: Exception type: {type(e)}")
        import traceback
        print(f"ERROR: Traceback: {traceback.format_exc()}")
        return {
            'statusCode': 500,
            'headers': {
                'Content-Type': 'application/json'
            },
            'body': json.dumps({
                'error': str(e)
            })
        }