ゼロから始めるAIシステム開発 #06 「LangChainとStreamlitを使った生成AIアプリ開発」

フレームワークを使用して生成AIアプリを開発

入門書3章の後半で、実際に生成AIアプリを実装してみるハンズオン。
Amazon Bedrock 生成AIアプリ開発入門 ［AWS深掘りガイド］
https://amzn.asia/d/co7MB5S
なかなか難しいパートだったので一週空いてしまった。

使用するフレームワークについて

• LangChain
  最も有名でスタンダード。シンプルなコードで生成AIアプリを開発できるので試作検証におすすめ。
• Streamlit
  Pythonの数行のコードでチャットインターフェースを実装できるため概念実証によく使われる。

1.LangChainの実装

1_langchain.py

from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage, SystemMessage
chat = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs={"max_tokens":1000},
)
messages = [
    SystemMessage(content="あなたのタスクはユーザーの質問に明確に答えることです。"),
    HumanMessage(content="空が青いのはなぜですか？"),
]
response = chat.invoke(messages)
print(response.content)

• Chatbedrockを生成
• メッセージを定義
• モデル呼び出し

2.デバッグオプションでプロンプトのログ取得

2_langchain-debug.py

from langchain.globals import set_debug
from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage, SystemMessage
set_debug(True)
chat = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs={"max_tokens":1000},
)
messages = [
    SystemMessage(content="あなたのタスクはユーザーの質問に明確に答えることです。"),
    HumanMessage(content="空が青いのはなぜですか？"),
]
response = chat.invoke(messages)
print(response.content)

• デバッグを有効化（以下は1と同じ）

3.ストリーム出力

3_langchain-streaming.py

from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage, SystemMessage
chat = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs={"max_tokens":1000},
    streaming=True,
)
messages = [
    SystemMessage(content="あなたのタスクはユーザーの質問に明確に答えることです。"),
    HumanMessage(content="空が青いのはなぜですか？"),
]
for chunk in chat.stream(messages):
    print(chunk.content,end="",flush=True)
print("")

• 最後のモデル呼び出しをストリーム形式にする

（↑1とちょっと違う結果になっている）

4.Streamlitとの統合

入門書ではStreamlitのアプリ画面確認にCloud9の機能を使っているが、現状ではサービスが終了しているのでSageMakerで代替している。（Pinggyという外部サービスをさらに利用）

4_streamlit.py

import streamlit as st
from langchain_aws import ChatBedrock
from langchain_core.messages import HumanMessage, SystemMessage
st.title("Bedrock チャット")
chat = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs={"max_tokens":1000},
    streaming=True,
)
messages = [
    SystemMessage(content="あなたのタスクはユーザーの質問に明確に答えることです。"),
]
if prompt := st.chat_input("なんでも聞いてください"):
    messages.append(HumanMessage(content=prompt))
    with st.chat_message("user"):
        st.markdown(prompt)
    with st.chat_message("assistant"):
        st.write_stream(chat.stream(messages))

• st.titleでタイトルを作成
• st.chat_inputでチャット入力欄を定義
• st.chat_messageでアバターのアイコンが変わる
• st.markdownで入力した結果を表示
• st.write_streamでモデル呼び出しと結果表示
• 変数promptで入力内容を受け取り

Streamlitではテキスト欄の入力などのイベントが発生したときに毎回 Pythonスクリプトの処理が再実行される。 この仕組みで変数が毎回初期化されてしまうため、↑のスクリプトでは過去の質問を参照することができない。

5.やりとりを継続する

…ので、過去の質問を保持して継続したやりとりをするためのスクリプトが以下になる。

5_streamlit-session.py

import streamlit as st
from langchain_aws import ChatBedrock
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
st.title("Bedrock チャット")
chat = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs={"max_tokens":1000},
    streaming=True,
)
if "messages" not in st.session_state:
    st.session_state.messages = [
        SystemMessage(content="あなたのタスクはユーザーの質問に明確に答えることです。"),
    ]
for message in st.session_state.messages:
    if message.type != "system":
        with st.chat_message(message.type):
            st.markdown(message.content)
if prompt := st.chat_input("なんでも聞いてください"):
    st.session_state.messages.append(HumanMessage(content=prompt))
    with st.chat_message("user"):
        st.markdown(prompt)
    with st.chat_message("assistant"):
        response = st.write_stream(chat.stream(st.session_state.messages))
    st.session_state.messages.append(AIMessage(content=response))

• st.session_stateを使用すると値を保持できる
• ifでst.session_stateに内容がないとき（初回）のみに内容をセットする

チャット履歴を永続化する

過去の質問を保持できるようになったが、これでもブラウザをリロードすると履歴が初期化されてしまうため、データベースに保存して永続化する処理を追加する。
データベースにはAWSのDynamoDBを使用。

6_streamlit-dynamodb.py

import streamlit as st
from langchain_aws import ChatBedrock
from langchain_community.chat_message_histories import DynamoDBChatMessageHistory
from langchain_core.messages import HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
st.title("Bedrock チャット")
if "session_id" not in st.session_state:
    st.session_state.session_id = "session_id"
if "history" not in st.session_state:
    st.session_state.history = DynamoDBChatMessageHistory(
        table_name="BedrockChatSessionTable", session_id=st.session_state.session_id
    )
if "chain" not in st.session_state:
    prompt = ChatPromptTemplate.from_messages(
        [
            ("system","あなたのタスクはユーザーの質問に明確に答えることです。"),
            MessagesPlaceholder(variable_name="messages"),
            MessagesPlaceholder(variable_name="human_message"),
        ]
    )
    chat = ChatBedrock(
        model_id="anthropic.claude-3-sonnet-20240229-v1:0",
        model_kwargs={"max_tokens":1000},
        streaming=True,
    )
    chain = prompt | chat
    st.session_state.chain = chain
if st.button("履歴クリア"):
    st.session_state.history.clear()
for message in st.session_state.history.messages:
    with st.chat_message(message.type):
        st.markdown(message.content)
if prompt := st.chat_input("なんでも聞いてください"):
    with st.chat_message("user"):
        st.markdown(prompt)
    with st.chat_message("assistant"):
        response = st.write_stream(
            st.session_state.chain.stream(
                {
                    "messages":st.session_state.history.messages,
                    "human_message":[HumanMessage(content=prompt)],
                },
                config={"configurable":{"session_id": st.session_state.session_id}},
            )
        )
    st.session_state.history.add_user_message(prompt)
    st.session_state.history.add_ai_message(response)

• DynamoDBChatMessageHistoryをインスタンス化してst.session_state.historyに代入
• Chain（LangChainの処理の呼び出しを連続的につなげたもの）を定義。今回はpromptとchatを繋ぐ
• メッセージをst.session_state.historyから取得し画面表示
• 履歴クリアボタンを設置
• 最後に履歴を追加する

感想

3章ハンズオンはもう一つAWS Lambda上で動作するアプリのパートがあるのだがとりあえずここで区切り。正直Pythonが全く分からないので機序の理解が難しくなってきた。ちょっとだけでも先にPythonを勉強したほうが話が早いかもしれない。