【生成AI初心者大歓迎】1時間でRAG on AWSを作る！Slack AI チャットボット実践ハンズオン！

Last updated at 2024-11-21Posted at 2024-11-20

本資料は、社内で生成AIハンズオンを実施する際に作成したものです。
より多くの方にも参考にしていただけるよう、一般公開しています。

このハンズオンには、以下の3つの特徴があります：

AWS初心者やプログラミング未経験者でも取り組めるよう、ほぼGUIベースの操作のみで完結
作ったもの社内でそのまま活用できるよう、AWS上だけでなくSlackとの連携まで実施
多くの方が時間を取りやすいよう、所要時間を1時間に設定

↓ここからが本編です。

1. はじめに

1.1 このハンズオンの目的

ChatGPTの登場以降、生成AIは私たちの仕事や生活に大きな影響を与えています。特にIT業界においては、もはや避けて通れないテーマとなっています。

このような状況を踏まえ、本ハンズオンでは以下を目指します

生成AIに関する知見を、実際に手を動かしながら身に付けていただき、お客様への提案や施策の企画などへ活用していただくこと
ハンズオン内で作成したシステムを、そのまま実務でも活用していただき、業務効率化につなげていただくこと

1.2 生成AIの仕組みと特徴

まず最初に、生成AIがなぜ質問への回答や文章作成といった高度な言語タスクを実行できるのかを説明します。

簡単に言うと、生成AIは人間の脳の仕組みをコンピューター上で再現しているため、人間のような言語処理が可能なのです。

具体的には、私たちの脳の中にある神経細胞（ニューロン）のつながり（ネットワーク）を数式で表現した「ニューラルネットワーク」という仕組みを使っています。

https://jitera.com/ja/insights/53089

生成AIは、このニューラルネットワークを使って膨大な量の文章（Wikipedia数千個分※）を読み込んで学習します。

この学習には、家庭用の高性能パソコン（ゲーミングPC）を数十万台分集めたような、とてつもない計算能力が必要です。

このような大量の学習によって、生成AIは人間の言葉をよく理解できるようになり、質問に答えたり、文章をまとめたり、レポートを書いたり、プログラミングのコードを作ったりと、様々な言語の作業を正確にこなせるようになります。

※Llama 3の情報を基に、Wikipedia英語版（約30億単語）とゲーミングPCでよく使われるRTX 4090の性能とを比較し推定
https://medium.com/axinc/llama3%E3%81%AE%E8%AB%96%E6%96%87%E3%82%92%E8%AA%AD%E3%82%80-fd232b658424

そこで本ハンズオンでは、この優れた技術を実際の仕事で役立てるため、AWSの生成AIサービスとSlackを組み合わせた、実用的なチャットボットの作り方を学んでいきます。

2. AWSで生成AIを触ってみよう

まずは実際に生成AIを操作しながら、その特徴をより深く理解していきましょう。

そのために、AWSが提供する生成AI向けのマネージドサービス「Amazon Bedrock」を使用します。
Amazon Bedrockは、様々な生成AIモデルを簡単に利用できます。

2.1 環境準備

まず、AWSマネジメントコンソールにログインし、リージョンをオレゴン(us-west-2)に変更します。

その後、検索バーで「Bedrock」と入力し、「Amazon Bedrock」を選択します。

2.2 モデルアクセスの有効化

モデルアクセスの設定をするには、まず左側メニューから「モデルアクセス」を選択してください。
その後、モデルアクセスを初めて設定される方は「特定のモデルを有効にする」を、以前に他のモデルアクセスを設定したことがある方は「モデルアクセスを変更」をクリックしてください。

「Claude 3.5 Haiku」と「Titan Text Embeddings V2」選択し、「次へ」→「送信」で有効化します。

利用規約に同意することになりますのでご注意ください。

Claudeは、Anthropic社が開発した生成AIで、高性能な大規模言語モデル（LLM）です。
GPT-4oよりも多くのベンチマークで性能を上回っているのが特徴です。

https://www.anthropic.com/news/3-5-models-and-computer-use

2.3 モデルの動作確認

実際に有効化したClaude 3.5 haikuと対話してみましょう。
「プレイグラウンド」→「Chat/Text」を選択します。

Claude 3.5 haikuを選択します。

これで質問できるようになりました。
以下のような質問をしてみましょう

Pythonでcsvからjsonに変換するスクリプトを書いてください

質問に対して回答してくれていることが分かります。

次に、以下のような質問をしてみましょう

あなたはいつまでの情報を学習してますか？
また、HULFT Squareはどこの会社が提供しているサービスですか？

2022年1月までの情報しか学習していないため、回答が間違っていることが分かります。

HULFT Squareは、2023年2月にリリースされた、株式会社セゾンテクノロジーが提供する、システム間のデータ連携を実現するクラウドサービス（iPaaS）です。

このことから分かるように、LLMには事前学習していない、最新の情報や公開されていない情報については、回答できないことが分かります。

これを克服するために「RAG（Retrieval-Augmented Generation）」という手法が開発されています。
次章では、RAGを使ってLLMが答えられない情報も回答できるようにしていきます。

3. RAGの基礎と概要

3.1 RAGとは

RAGとは、LLMと外部の知識データベースを組み合わせて使用する手法です。

これにより、LLMが学習していない情報にも回答できるようになったり、誤った情報の生成（ハルシネーション）を減らすこともできます。

RAGの仕組みをより深く理解するために、まずは基本的な処理の流れを見ていきましょう。

3.2 RAGの処理フロー

RAGの基本的な処理の流れは以下のステップで構成されています：

①ドキュメントの前処理：必要な文書を検索可能な形式に加工し、データベースに保存します。
②質問：ユーザーから質問を受け取ります。
③検索：質問内容に関連する情報をデータベースから探し出します。
④取得：検索結果から必要な部分を取り出します。
⑤回答生成：抽出した情報をLLMに渡して、回答を生成します。

この仕組みにより、LLMは事前学習では得られなかった情報でも回答することができます。

それでは、この処理の技術的な詳細について、さらに掘り下げて説明していきましょう。

3.3 RAGの技術的な仕組み

ドキュメントの前処理

まず、ドキュメントの前処理では、長い文書を「チャンク」と呼ばれる小さな単位に分割します。

次に、各チャンクを「埋め込みモデル」というAIを使って数値データ（ベクトル）に変換し、ベクターDBに保存します。

質問~回答生成処理

ユーザーからの質問も同様に数値データに変換され、ベクトル検索によって質問の意味に最も近いチャンクを見つけ出します。

最後に、見つかったチャンクの情報をLLMに提供して回答を生成します。

ここまでRAGの基本的な概念と仕組みについて説明してきました。

それでは実際に、AWS上でRAGを構築していきましょう。

4. Amazon Bedrock Knowledge Basesによる実装

4.1 システムアーキテクチャ

AWSの「Amazon Bedrock Knowledge Bases」というサービスは、RAGを構成する下記アーキテクチャを簡単に構築することができます。

システムの全体像が把握できたところで、実際の環境構築に進んでいきましょう。

4.2 環境構築

4.2.1 データの準備

RAGに読み込ませるデータのダウンロード

今回はRAGに読み込ませるドキュメントとしてHULFT SquareのHPを使用します。

下記URLにアクセスし
https://www.hulft.com/service/hulft-square
Ctrl+S でHTMLとして保存しておきます。

S3バケットの作成

AWSコンソールで「S3」を検索します。

「バケットを作成」を選択します。

バケット名を設定し、その他デフォルトのままバケットを作成します。

S3バケットが作成できたら、次はダウンロードしたファイルをアップロードしていきましょう。

ダウンロードしたファイルのアップロード

作成したS3バケットを選択します。

「アップロード」をクリックします。

「ファイルを追加」を選択し、先ほどダウンロードしたファイルを選択し、アップロードします。

これで読み込ませるデータの準備は完了しました。次は、このデータを活用するためのKnowledge Basesを作成していきましょう。

4.2.2 Knowledge Basesの構築

Amazon Bedrockコンソールに移動し、左メニューから「ナレッジベース」を選択し、「ナレッジベースを作成」をクリックします。

分かりやすいよう、ナレッジベース名に先ほどのS3バケットの名前を追加します。

その他はデフォルト設定で次へ進みます。

「S3 URIの参照」を選択し、

先ほど作成したS3バケットを選択し、次のページへ進みます。

埋め込みモデルは「titan-text-embedding-v2」を選択し、ベクトルデータベースは「新しいベクトルストアをクイック作成」を選択（デフォルト）し、次へ進みます。

最後に設定内容を確認し、作成します。

作成されるまで数分待機します。

作成失敗した場合は、もう一度作成を試みます。

作成成功したら、Slackとの連携時に使用するためナレッジベースIDを保管しておきます。

「データソース」→「同期」を選択し完了するまで待機します。

これでAmazon Bedrock Knowledge Basesを作成し、データ同期まで完了しました。それでは、実際に作成したシステムの動作を確認していきましょう。

4.3 動作検証

「ナレッジベースをテスト」から「モデルを選択」を選択し、

モデルは先ほどと同様Claude 3.5 Haikuを選択します。

それでは、以下の質問を試してみましょう：

HULFT Squareはどこの会社が提供しているサービスですか？

先ほどは間違っていた質問でも、今回は回答が得られていることが分かります。

このように、Amazon Bedrock Knowledge Basesを用いることで、簡単にRAGを実装でき、LLMが学習していない情報からでも、正確な回答が得られるようになりました。

5. Slackとの連携

これまでAmazon Bedrock Knowledge Basesを使ってRAGシステムを構築し、テストコンソールで動作確認を行いました。

本章では、このシステムを実際の業務で活用できるよう、Slackと連携させる実装を行います。

5.1 システムアーキテクチャ

Slackと連携したシステムの全体像は以下のようになります。

ユーザーがSlackでボットにメンションを送ると、そのメッセージはAmazon API Gatewayを経由してAWS Lambdaに届きます。

Lambda関数は受け取ったメッセージを元に、先ほど作成したKnowledge Basesに問い合わせを行い、その回答をSlackに返信します。

5.2 Slack Appの作成

まず、Slackにチャットボットをインストールするための準備として、Slack Appを作成します。

Slack APIにアクセスし、

「Create New App」をクリックします。

「From Scratch」を選択し、

App Nameを設定し、導入したいSlackワークスペースを選択し、「Create App」を選択し作成します。

次に「OAuth & Permissions」セクションに移動し、「Bot Token Scopes」で以下の権限を追加し、

- app_mentions:read
- chat:write
- channels:read

「Install to Workspace」をクリックします。

確認画面で「許可する」を選択します。

OAuth TokensのBot User OAuth Tokenと、Basic informationにあるSigning Secretは後で使用するので保管してください。

5.3 AWS側の環境構築

AWSコンソールで「Lambda」と検索し、クリックします。

そして「関数の作成」をクリックします。

「設計図の使用」を選択し、設計図名は「Handle Slack slash commands」、関数名は任意に設定します。
実行ロールは「基本的な Lambda アクセス権限で新しいロールを作成」のままとします。

API Gatewayトリガーでは、インテントは「新規APIを作成」、APIタイプは「HTTP API」、セキュリティは「オープン」に設定します。

環境変数には以下を設定し、

KNOWLEDGE_BASE_ID: [作成したKnowledge BaseのID]
MODEL_ARN: arn:aws:bedrock:us-west-2::foundation-model/anthropic.claude-3-5-haiku-20241022-v1:0
SLACK_BOT_TOKEN: [先ほど作成したSlack AppのSlackのBot User OAuth Token]
SLACK_SIGNING_SECRET: [先ほど作成したSlack AppのSigning Secret]

「関数の作成」を選択します。

Lambda関数が正常に作成されることを確認します。

そして、Lambda関数のコードに、下記コードを全文コピペし、「Deploy」を選択します。

import json
import os
import urllib.request
import urllib.parse
import boto3
import logging
import hashlib
import hmac
import time

# ロギングの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 環境変数
SLACK_BOT_TOKEN = os.environ['SLACK_BOT_TOKEN']
SLACK_SIGNING_SECRET = os.environ['SLACK_SIGNING_SECRET']
KNOWLEDGE_BASE_ID = os.environ['KNOWLEDGE_BASE_ID']
MODEL_ARN = os.environ['MODEL_ARN']

# Bedrockクライアントの初期化
bedrock_agent_runtime = boto3.client('bedrock-agent-runtime')

def verify_slack_request(headers, body):
    """
    Slackリクエストの検証を行う関数
    """
    # タイムスタンプの取得と検証
    timestamp = headers.get('x-slack-request-timestamp', '')
    if not timestamp:
        logger.warning("No timestamp in request")
        return False

    # 現在時刻との差分を確認（5分以上経過したリクエストは拒否）
    current_time = int(time.time())
    if abs(current_time - int(timestamp)) > 300:
        logger.warning("Request is too old")
        return False

    # 署名の検証
    slack_signature = headers.get('x-slack-signature', '')
    if not slack_signature:
        logger.warning("No signature in request")
        return False

    # 署名の計算と比較
    sig_basestring = f'v0:{timestamp}:{body}'
    calculated_signature = 'v0=' + hmac.new(
        SLACK_SIGNING_SECRET.encode(),
        sig_basestring.encode(),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(calculated_signature, slack_signature):
        logger.warning("Invalid signature")
        return False

    return True

def query_knowledge_base(query_text):
    """
    Knowledge Baseに問い合わせを行う関数
    """
    try:
        response = bedrock_agent_runtime.retrieve_and_generate(
            input={
                'text': query_text
            },
            retrieveAndGenerateConfiguration={
                'type': 'KNOWLEDGE_BASE',
                'knowledgeBaseConfiguration': {
                    'knowledgeBaseId': KNOWLEDGE_BASE_ID,
                    'modelArn': MODEL_ARN,
                    'retrievalConfiguration': {
                        'vectorSearchConfiguration': {
                            'numberOfResults': 5
                        }
                    }
                }
            }
        )
        return response['output']['text']
    except Exception as e:
        logger.error(f"Error querying knowledge base: {str(e)}")
        return "申し訳ありません。問い合わせ中にエラーが発生しました。"

def send_slack_message(channel_id, message, thread_ts=None):
    """
    Slackにメッセージを送信する関数
    """
    headers = {
        'Authorization': f'Bearer {SLACK_BOT_TOKEN}',
        'Content-Type': 'application/json'
    }
    payload = {
        'channel': channel_id,
        'text': message
    }
    if thread_ts:
        payload['thread_ts'] = thread_ts
    
    try:
        data = json.dumps(payload).encode('utf-8')
        req = urllib.request.Request(
            'https://slack.com/api/chat.postMessage',
            data=data,
            headers=headers,
            method='POST'
        )
        with urllib.request.urlopen(req) as response:
            return json.loads(response.read().decode('utf-8'))
    except Exception as e:
        logger.error(f"Error sending message: {str(e)}")
        return {'ok': False, 'error': str(e)}

def lambda_handler(event, context):
    try:
        # リクエストの詳細をログに記録
        logger.info(f"Received event: {json.dumps(event)}")
        
        # ヘッダーの準備（大文字小文字の違いに対応）
        headers = {k.lower(): v for k, v in event.get('headers', {}).items()}
        body = event.get('body', '')

        # 重複リクエストのチェック
        retry_num = headers.get('x-slack-retry-num')
        if retry_num:
            logger.info(f"Detected x-slack-retry-num: {retry_num}. Exiting to avoid processing a retry from Slack.")
            return {
                "statusCode": 200,
                "body": json.dumps({"message": "Request identified as a retry, thus ignored."})
            }

        # リクエストの検証
        if not verify_slack_request(headers, body):
            return {
                'statusCode': 401,
                'body': json.dumps({'error': 'Invalid request signature'})
            }

        if not body:
            return {
                'statusCode': 400,
                'headers': {'Content-Type': 'application/json'},
                'body': json.dumps({'error': 'No body provided'})
            }

        # Base64エンコードされている場合はデコード
        if event.get('isBase64Encoded', False):
            import base64
            body = base64.b64decode(body).decode('utf-8')
        
        slack_event = json.loads(body)

        # Slack URLの検証チャレンジ
        if 'challenge' in slack_event:
            return {
                'statusCode': 200,
                'headers': {'Content-Type': 'text/plain'},
                'body': slack_event['challenge']
            }

        # メンションイベントの処理
        if 'event' in slack_event:
            event_data = slack_event['event']
            if event_data.get('type') == 'app_mention':
                channel_id = event_data['channel']
                thread_ts = event_data.get('thread_ts', event_data['ts'])
                
                # メッセージテキストから不要な部分を削除
                message_text = event_data['text']
                # メンションを除去（<@USERID>形式）
                message_text = ' '.join(word for word in message_text.split() if not word.startswith('<@'))
                
                # 処理中メッセージを送信
                processing_message = "お問い合わせを処理中です。少々お待ちください..."
                send_slack_message(channel_id, processing_message, thread_ts)
                
                # Knowledge Baseに問い合わせ
                response_text = query_knowledge_base(message_text)
                
                # 最終的な回答を送信
                response = send_slack_message(channel_id, response_text, thread_ts)
                
                if not response.get('ok', False):
                    raise Exception(f"Failed to send message: {response.get('error', 'Unknown error')}")

        return {
            'statusCode': 200,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({'message': 'Success'})
        }

    except Exception as e:
        logger.error(f"Error: {str(e)}")
        return {
            'statusCode': 500,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({'error': str(e)})
        }