Bedrock Webアプリのサンプル6本（チャット、RAG、Agent）-Amazon Bedrock APIで始めるLLM超入門⑦

Last updated at 2024-03-02Posted at 2023-10-17

これまで作った以下の6本をStreamlitを使ってWebアプリ化し、デモしやすくします。

boto3からのBedrock呼出し
boto3からのBedrock呼出し（ストリーミング）
LangChainからのBedrock呼出し
LangChainからのチャットボット（履歴付き）
LangChain Agent機能を使ったWeb検索／要約機能
LangChain Agent機能を使ったkendra検索RAG

ローカル環境にWebサーバー機能を持つのでインフラの構築やデプロイは不要です。
元となった各プログラムの解説は過去の記事をご参照ください。

2024/2/29
LangChainを最新バージョンにすると色々動かないので、最新バージョンで動かす場合は以下のエントリをご参照ください

動作イメージ

花椒やサンショウ、ラードや豚肉の脂身


米もスパイスも入ってないですね

何故かラミレス

合ってるのか合ってないのか難易度高いですが最近のニュースではありそうです

素のClaude2はBedrockを知らないので合ってます！

Streamlitとは

公式サイト（英語）をWeb検索要約君に説明してもらうと以下との事です。

ローカル環境の構築

ターミナル

pip install -U streamlit

他のモジュールや環境構築等は実行するプログラムによって異なるので過去の記事をご参照ください

実行方法

ターミナル

python -m streamlit run xxxxxxx.py

これでブラウザが立ち上がると思いますが、実行後に表示される以下のlocal URLを起動しても実行できると思います（ポート番号はデフォルトでは8501のようです）。

ターミナル

  You can now view your Streamlit app in your browser.

  Local URL: http://localhost:8501

プログラム

コードの麗しさには目を瞑って、ほぼ元のプログラムの前後をStreamlit用の処理で挟んだだけのつもりです。入力値を取得する変数を変えたぐらいです。
1つのアプリはプログラム1本だけで動きますので、該当のプログラムを保存してStreamlitから起動すれば動きます。

元の処理を活かした為、最大トークン数等のパラメータの有無もそのままなので、そのあたりは適宜調整してください。プログラムによって指定していたりしていなかったり長かったり短かったりします。またプロンプトやコードの書き方も、元プログラムを書いた時の気分やコピペ元で違いがある可能性があります

1_st_bedrock.py

import boto3
import json

# Streamlit用の処理
# ボタンを押されたら元の処理を呼ぶ
import streamlit as st

st.title("シンプルなBedrock呼出し")
input_text = st.text_input("このテキストをBedrockに送信します")
send_button = st.button("送信")

if send_button:
    # ここから元の処理(プロンプトの入力値はStreamlitから取得)
    bedrock = boto3.client(service_name='bedrock-runtime')
    body = json.dumps({
        "prompt": "\n\nHuman:" + input_text + "\n\nAssistant:",
        "max_tokens_to_sample": 300,
        "temperature": 0.1,
        "top_p": 0.9,
    })
    modelId = 'anthropic.claude-v2'
    accept = 'application/json'
    contentType = 'application/json'

    response = bedrock.invoke_model(body=body, modelId=modelId, accept=accept, contentType=contentType)
    response_body = json.loads(response.get('body').read())
    response_text = response_body.get('completion')

    # 結果をStreamlitに出力
    st.write(response_text)

2_st_bedrock_stream.py

import boto3
import json

# Streamlit用の処理
# ボタンを押されたら元の処理を呼ぶ
import streamlit as st

st.title("BedrockのStream呼出し")
input_text = st.text_input("このテキストをBedrockに送信します")
send_button = st.button("送信")

if send_button:
    result_area = st.empty()
    text = ''

    # ここから元の処理(プロンプトの入力値はStreamlitから取得)
    # Stream処理を見たいので出力トークン数を少し大きくした（1000）
    bedrock = boto3.client(service_name='bedrock-runtime')
    body = json.dumps({
        'prompt': '\n\nHuman:' + input_text + '\n\nAssistant:',
        'max_tokens_to_sample': 1000
    })                    
    response = bedrock.invoke_model_with_response_stream(modelId='anthropic.claude-v2', body=body)
    stream = response.get('body')
    if stream:
        for event in stream:
            chunk = event.get('chunk')
            if chunk:
                # 取得したチャンクをappendしてStreamlitに出力
                text += json.loads(chunk.get('bytes').decode())["completion"]
                result_area.write(text)

3_st_langchain.py

from langchain.llms import Bedrock

# Streamlit用の処理
# ボタンを押されたら元の処理を呼ぶ
import streamlit as st

st.title("LangChainからのBedrock呼出し")
input_text = st.text_input("このテキストをBedrockに送信します")
send_button = st.button("送信")

if send_button:

    # ここから元の処理(プロンプトの入力値はStreamlitから取得)
    llm = Bedrock(model_id="anthropic.claude-v2")
    answer = llm.predict(input_text)

    # 結果をStreamlitに出力
    st.write(answer)

↑
これ短さが極まってますね。9ステップでBedrockで生成するWebアプリが作れます。
抽象化され過ぎてて何が何だか分かりません。

4_st_chat.py

from langchain.llms import Bedrock
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory
from langchain.memory.chat_message_histories import DynamoDBChatMessageHistory

# セッションID（この値をチャット履歴のキーとしてDynamoDBから取得／格納する）
session_id = "12345"

# Streamlit用の処理
# ボタンを押されたら元の処理を呼ぶ
import streamlit as st

st.title("LangChainでChat")

# チャット履歴を初期化する
if "messages" not in st.session_state:
    # 辞書形式で定義
    st.session_state["messages"] = []

# これまでのチャット履歴を全て表示する 
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])
    
if input_text := st.chat_input("このテキストをBedrockに送信します"):

    # ユーザの入力を表示する
    with st.chat_message("user"):
        st.write(input_text)
    # ユーザの入力をチャット履歴に追加する
    st.session_state.messages.append({"role": "user", "content": input_text})

    # ここから元の処理(プロンプトの入力値はStreamlitから取得)

    # promptの定義
    prompt = PromptTemplate(
        input_variables=["chat_history","Query"],
        template="""あなたは人間と会話をするAIアシスタントです。

    ChatHistory: {chat_history}
    \nHuman: {Query}
    \nAssistant:"""
    )

    # チャット履歴の定義（DynamoDBの"SessionTable"を使用。使用されるキーは"SessionId"）
    message_history = DynamoDBChatMessageHistory(table_name="SessionTable", session_id=session_id)
    memory = ConversationBufferMemory(
        memory_key="chat_history", chat_memory=message_history, return_messages=True
    )

    # LLMの定義
    llm = Bedrock(
        model_id="anthropic.claude-v2"
    )

    # LLMChainの定義
    llm_chain = LLMChain(
        llm=llm,
        prompt=prompt,
        verbose=False, #最終的なPromptの内容を出力するパラメータ
        memory=memory,
    )

    # LLMChainの実行
    answer = llm_chain.predict(Query=input_text)

    # ChatBotの返答を表示する 
    with st.chat_message("assistant"):
        st.write(answer)
    # ChatBotの返答をチャット履歴に追加する
    st.session_state.messages.append({"role": "assistant", "content": answer})

5_st_Agent.py

from langchain.llms import Bedrock
from langchain.chains import LLMChain
from langchain.agents import XMLAgent
from langchain.agents.agent import AgentExecutor
from langchain.agents import Tool
from langchain.tools import DuckDuckGoSearchRun

# ここは元の処理(ツール定義)

# Webページ読み込み用関数。渡されたURLの本文を返却する
from langchain.document_loaders import WebBaseLoader
def web_page_reader(url: str) -> str:
    loader = WebBaseLoader(url)
    content = loader.load()[0].page_content
    return content

# Web検索用ツール
search = DuckDuckGoSearchRun()

# 使用可能なツールと説明
tools = [
    Tool(
        name="duckduckgo-search",
        func=search.run,
        description="このツールはWeb上の最新情報を検索します。引数で検索キーワードを受け取ります。最新情報が必要ない場合はこのツールは使用しません。",
    ),
    Tool(
        name = "WebBaseLoader",
        func=web_page_reader,
        description="このツールは引数でURLを渡された場合に内容をテキストで返却します。引数にはURLの文字列のみを受け付けます。"
    )
]

# Streamlit用の処理
# ボタンを押されたら元の処理を呼ぶ
import streamlit as st

st.title("Web検索要約君")
input_text = st.text_input("Web検索をして答えます。URLを渡された場合は内容を答えます。気まぐれで検索せずに答えます。")
send_button = st.button("送信")

if send_button:

    # ここから元の処理(プロンプトの入力値はStreamlitから取得)
    llm = Bedrock(
        model_id="anthropic.claude-v2",
        model_kwargs={"max_tokens_to_sample": 2000} # 最大トークン数は大きめに
    )

    # Agentの定義
    chain = LLMChain(
        llm=llm,
        prompt=XMLAgent.get_default_prompt(),
        output_parser=XMLAgent.get_default_output_parser()
    )
    agent = XMLAgent(tools=tools, llm_chain=chain)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False, handle_parsing_errors=True)

    # Agentの実行
    answer = agent_executor.invoke({"input": "特に言語の指定が無い場合はあなたは質問に対して日本語で回答します。" + input_text})

    # 結果をStreamlitに出力
    st.write(answer['output'])

6_st_RAG.py

from langchain.llms import Bedrock
from langchain.chains import LLMChain
from langchain.agents import XMLAgent
from langchain.agents.agent import AgentExecutor
from langchain.agents import Tool

# ここは元の処理(ツール定義)

# Amazon Kendraから情報を取得する
# 日本語で"登録されている"ドキュメントを20件(top_k=20)検索して取得結果を全て結合する
from langchain.retrievers import AmazonKendraRetriever
kendra_index_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" #各自のKendra Index IDに書き換えてください
def get_retrieval_result(query) -> str:
    attribute_filter = {"EqualsTo": {"Key": "_language_code","Value": {"StringValue": "ja"}}}
    retriever = AmazonKendraRetriever(index_id=kendra_index_id,attribute_filter=attribute_filter,top_k=20)
    docs = retriever.get_relevant_documents(query=query)
    context=""
    for doc in docs:
        context += doc.page_content
    return context

# 使用可能なツールと説明
tools = [
    Tool(
        name = "KendraSearch",
        func=get_retrieval_result,
        description=""""
            このツールは最新のWeb情報を検索するツールです。引数は検索キーワードです。検索キーワードの例は<examples>の通りです。
            <examples>
                <question>Bedrockについて説明してください</question><検索キーワード>Bedrock</検索キーワード>
                <question>Kendraについて英語で教えてください</question><検索キーワード>Kendra</検索キーワード>
                <question>LangChainはBedrockで使用する事が出来ますか？</question><検索キーワード>LangChain Bedrock</検索キーワード>                
                <question>Amplifyは無料で使用する事が出来ますか？</question><検索キーワード>Amplify</検索キーワード>
                <question>アマゾンとは何か</question><検索キーワード>アマゾン</検索キーワード>
            </examples>"""
    )
]

# Streamlit用の処理
# ボタンを押されたら元の処理を呼ぶ
import streamlit as st

st.title("KendRAG(LangChain Agent版)")
input_text = st.text_input("LLMに対する指示を取り除いてからKendraを検索して答えます")
send_button = st.button("送信")

if send_button:

    # ここから元の処理(プロンプトの入力値はStreamlitから取得)

    # LLMの定義
    llm = Bedrock(
        model_id="anthropic.claude-v2",
        model_kwargs={"max_tokens_to_sample": 1000}
    )

    # Agentの定義
    chain = LLMChain(
        llm=llm,
        prompt=XMLAgent.get_default_prompt(),
        output_parser=XMLAgent.get_default_output_parser()
    )
    agent = XMLAgent(tools=tools, llm_chain=chain)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=False, handle_parsing_errors=True)

    # Agentの実行
    answer = agent_executor.invoke({"input": "特に言語の指定が無い場合はあなたは質問に対して日本語で回答します。<question>" + input_text + "</question>\
        もし<question>の内容が<observation>に含まれない場合は、「検索結果にありません」と答えてください\n\nAssistant:"})

    # 結果をStreamlitに出力
    st.write(answer['output'])

こんな感じでBedrock、Claude2、LangChainを手元で好きに触れる環境が出来たかなと思います。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up