社内WikiやWorksのデータを活用してAIエージェントを作れないか頑張ってみた

第1章　何作ろう

RAGを簡易的に作る方法は…？

Knowledge Bases for Amazon Bedrockを使えばいいのか！

画像出典：https://aws.amazon.com/jp/blogs/news/knowledge-bases-for-amazon-bedrock-rag-patent/

第2章 千里の道もローカルから

社内Wikiで使用している「Growi」

環境準備

Amazon Linux2023にdocker・docker-composeをインストール

sudo dnf update
sudo dnf install -y docker
sudo systemctl start docker
sudo gpasswd -a $(whoami) docker
sudo service docker restart
sudo systemctl enable docker
sudo curl -L "https://github.com/docker/compose/releases/download/v2.29.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose
sudo ln -s /usr/local/bin/docker-compose /usr/bin/docker-compose

git・growi-docker-composeをインストール

sudo dnf install git -y
git clone https://github.com/weseek/growi-docker-compose.git growi

dockerコンテナを立ち上げる

docker-compose up -d

適当な架空の会社の社内情報を作成してみる

就業規則

社内ルール

どちらの内容もイケイケな企業って感じがしますね。（適当）

Growiの記事をmarkdownファイルとして出力する

手順1 Growiのデータエクスポート機能を使用してアーカイブデータを作成する

手順2 作成されたzipファイルを展開してMarkdownファイルをエクスポートする

Markdownファイルのエクスポート

import json
import os
import re

with open('pages.json', 'r') as f:
    pages = json.load(f)

with open('revisions.json', 'r') as f:
    revisions = json.load(f)

pages_dict = {page['_id']: page['path'] for page in pages}
revisions_dict = {revision['pageId']: revision['body'] for revision in revisions}

pattern = re.compile(r'/Documents')

for page_id, path in pages_dict.items():
    if page_id in revisions_dict:
        Documents配下以外のファイルは作成しない
        if not pattern.search(path):
            continue
        ファイルパスを作成
        path = path.replace(' ', '_')
        file_path = './output' + path + '.md'
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        body = revisions_dict[page_id]
        
        ファイルを書き込む
        with open(file_path, 'w') as f:
            f.write(body)

それでは作成したファイルをS3の任意のバケットにアップロードしましょう。

第3章 文章さよなら、ベクトルこんにちは

Pinecone…？松ぼっくり…？

手順1 ナレッジベースの詳細を入力

手順2 データソースを設定

手順3 埋め込みモデルを選択肢、ベクトルストアを設定する

ナレッジベースは果たして機能するか…？

いざ参る…！

念のため…

第4章 もう全部任せた！

画像出典：https://dev.classmethod.jp/articles/introduction-2024-agents-for-amazon-bedrock/

エージェントに依頼するタスク

「Q&Aまとめ」Worksアプリの作成

また、エージェントがWorksアプリに対して検索をかける、つまりAPIを利用するためにAPIキーとアクセストークンを発行し、「AWS Secrets Manager」でそれらを登録しておきます。

サスケWorksでAPIを利用する際にはAPIドキュメントを参考ください。

(※後ほど説明しますが、エージェントには必要に応じてWorksアプリを検索するよう「アクショングループ」でlambda関数を設定します。その作成したアクショングループが登録したキーたちを利用できるようリソースポリシーを編集する必要がありますが詳細については割愛させていただきます。)

AIエージェントの作成

エージェントの詳細

アクショングループの設定

実際に実行されるlambda関数

import json
import requests
import boto3
from botocore.exceptions import ClientError


def lambda_handler(event, context):
    agent = event['agent']
    actionGroup = event['actionGroup']
    function = event['function']
    parameters = event.get('parameters', [])
    
    result = get_info_from_worksapp(parameters)

    responseBody =  {
        "TEXT": {
            "body": json.dumps(result, ensure_ascii=False)
        }
    }

    action_response = {
        'actionGroup': actionGroup,
        'function': function,
        'functionResponse': {
            'responseBody': responseBody
        }

    }

    function_response = {'response': action_response, 'messageVersion': event['messageVersion']}
    print("Response: {}".format(function_response))

    return function_response


def get_info_from_worksapp(parameters):
    
    service = get_value(parameters, 'service')
    type_question = get_value(parameters, 'type_question')
    
    WorksのAPIをたたく準備
    url = "https://api.works.app/v1/XXXX/records"
    api_key, token = get_secret()
    
    APIリクエストの設定
    headers = {
        "x-token": token,
        "x-api-key": api_key,
    }

    payload = {
        "searchParams": {
            "service": service,
            "sort": "dt_time",
            "sortType": "DESC",
            "type_question": type_question
        }
    }
    
    try:
        POSTリクエストを送信
        response = requests.post(url, json=payload, headers=headers)
    except requests.exceptions.RequestException as e:
        print(f"エラーが発生しました: {e}")

    
    レスポンスの整形
    res_data = response.json()
    records = res_data.get("records", [])
    
    新しい辞書を格納するリスト
    result = []
    
    ループ処理で 'question' と 'answer' の値を取り出して新しい辞書を作成
    for record in records:
        question = record.get("question").replace("\n", "")
        answer = record.get("answer").replace("\n", "")
        
        新しい辞書を作成してリストに追加
        result.append({
            "question": question,
            "answer": answer
        })

    return result
    
    
def get_secret():
    secret_name = "XXXXXX"  シークレット名を指定
    region_name = "XXXXXX"    シークレットが保存されているリージョン

    boto3クライアントを作成
    client = boto3.client('secretsmanager', region_name=region_name)

    try:
        Secrets Managerからシークレットを取得
        response = client.get_secret_value(SecretId=secret_name)
    except ClientError as e:
        
        raise Exception(f"Failed to retrieve secret: {e}")
    
    
    if 'SecretString' in response:
        secret = response['SecretString']
    else:
        secret = base64.b64decode(response['SecretBinary'])

    secret_dict = json.loads(secret)
    
    APIキーとアクセストークンを取得
    api_key = secret_dict['XXXXXX']
    access_token = secret_dict['XXXXXX']
    
    return api_key, access_token

def get_value(parameters, key, default=None):
    return next((p['value'] for p in parameters if p['name'] == key), default)

いざ実行…！

「rationale」の「text」の値に「これは一般的な知識に関する質問であり、ナレッジベースや特定のツールを使用する必要はありません。私の持っている情報を基に、北海道の代表的な観光スポットをいくつか紹介しましょう。」とあるので、確かに既に学習済みの知識から回答していることがわかりました。

次は先ほど作成したナレッジベースの情報をもとに回答してもらうことを想定した質問を投げてみましょう。今回は就業規則に関する質問にしてみます。

質問：TechFusion Inc.のフレックスタイム制度について教えてください。
回答：TechFusion Inc.のフレックスタイム制度について、以下のように説明いたします： 1. コアタイムは10:00〜15:00に設定されています。 2. 従業員は月間160時間の勤務を自主的に管理します。 3. 標準労働時間は1日8時間、週40時間です。 4. 従業員は一定の範囲内で自由に勤務時間を調整できます。 また、フレックスタイム制度と併せてリモートワークも推奨されており、以下のガイドラインがあります： - 勤務開始時にオンライン状態を共有 - Slackでの連絡をこまめに確認 これらの制度により、従業員の柔軟な働き方とワークライフバランスの向上がサポートされています。

コアタイムなど社内Wikiに記載されていた情報と一緒なのでこちらも大丈夫そうですが、「オーケストレーションとナレッジベース」を確認してみましょう。

見てみると「ナレッジベースから詳細な情報が得られた」と記述があるので期待通りにナレッジベースを利用してくれましたね。

最後にWorksアプリを検索する必要がある質問を投げかけてみたいと思います。
アプリには事前に「SUBLINEの料金体系とSUBLINEの同一スペース（同契約）同士の通話において着信料は発生しない」という情報を登録してあります。

質問：以下の問い合わせに対する回答を、サポート担当出茂が回答する形で作成してください。 問い合わせ内容 SUBLINEにおいて同一スペース同士の通話では着信料は発生しないのでしょうか？

回答：お客様各位 SUBLINEサポートの出茂でございます。 お問い合わせいただき、誠にありがとうございます。 ご質問の「SUBLINEにおいて同一スペース同士の通話では着信料は発生しないのでしょうか？」について、ご回答させていただきます。 結論から申し上げますと、同一スペース内での通話では着信料は発生いたしません。 詳細は以下の通りです： 1. 通常、SUBLINEでの着信料は1分につき2.2円（税込）発生いたします。 2. ただし、同一スペース内での通話に限り、この着信料は発生いたしません。 これにより、同じスペース内のメンバー同士で通話をする際は、着信料を気にすることなくコミュニケーションを取っていただけます。 なお、以下の点にご注意ください： - 通話録音や留守電設定で録音機能をオンにした場合は、別途録音料（2.2円/分、税込）が発生いたします。 - 転送設定を行った場合は、転送元として料金が発生いたします。 ご不明な点やさらなるご質問がございましたら、お気軽にお問い合わせください。SUBLINEを快適にご利用いただけるよう、精一杯サポートさせていただきます。 今後ともSUBLINEをよろしくお願いいたします。 SUBLINEサポート 出茂

こちらからの質問に回答したうえで追加で与えた「サポート担当出茂が回答する形で回答を作成してください」というオーダーにも対応してくれました。

今回もトレースを見てみると質問内容がどういうものであるかを判断したうえで、作成しlambda関数を実行してくれていることがわかります。

気持ちばかりではありますが…

import uuid
import json
import boto3
import streamlit as st

Agentの定義
agent_id: str = "XXXXXX"
agent_alias_id: str = "XXXXXX"

セッションIDの生成と保存
if "session_id" not in st.session_state:
    st.session_state.session_id = str(uuid.uuid4())
session_id = st.session_state.session_id
client = boto3.client("bedrock-agent-runtime")

if "history" not in st.session_state:
    st.session_state.history = []
st.title("AIエージェント")

for history in st.session_state.history:
    with st.chat_message(history["role"]):
        st.markdown(history["content"])


def invoke_bedrock_agent(input_text, agent_session_state=None):
    invoke_params = {
        "inputText": input_text,
        "agentId": agent_id,
        "agentAliasId": agent_alias_id,
        "sessionId": session_id,
        "enableTrace": True,
    }

    return client.invoke_agent(**invoke_params)

Bedrock Agentからのレスポンスを処理する関数
def process_agent_response(response):

    full_response = ""

    for event in response.get("completion", []):
        if "chunk" in event:
            chunk = event["chunk"]["bytes"].decode("utf-8")
            full_response += chunk

    return full_response

ユーザー入力の処理
if user_input := st.chat_input("入力してください"):
    ユーザー入力をチャット履歴に追加
    st.session_state.history.append({"role": "user", "content": user_input})
    with st.chat_message("user"):
        st.markdown(user_input)

    with st.chat_message("assistant"):
        response_placeholder = st.empty()
        full_response = ""

        Bedrock Agentを呼び出し、レスポンスを処理
        response = invoke_bedrock_agent(user_input)
        full_response = process_agent_response(response)

        response_placeholder.markdown(full_response)
        st.session_state.history.append({"role": "assistant", "content": full_response})

streamlit run sample.py

第5章 さいごに

※参考情報(Bedrockを利用することでかかった費用)