🐻‍❄️

Snowflake Arctic × Streamlit で会話履歴を保持したチャットAppを実装してみた

2024/05/19に公開

Streamlit

はじめに

この記事では、
Snowflake Arcticを利用したチャットアプリ上での会話履歴の保持の方法ををまとめます。
やり方は色々あると思うので、参考程度に。
会話履歴を保持する場合としない場合の挙動は以下です。

会話履歴あり

会話履歴なし

前提

2024年5月の執筆時点では、Arcticはまだ一部リージョンしか使用できません。
そのためオレゴン(aws_us_west_2)にSnowflakeアカウントを作成し、ユーザーやロールの初期設定は終えているものとします。
この辺りはsakatokuさんの記事をご覧ください。

参考)利用可能リージョン

Streamlit実装

現時点ではsnowparkのcortex.compleate関数だと履歴を保持した会話ができなかったので、
session.sqlで実装しています。(get_response関数の部分)

# Streamlit
import streamlit as st

# Snowflake and Snowpark
import snowflake.connector
import snowflake.snowpark as snowpark

# General libraries
import ast

def init():
    st.set_page_config(page_title="Arctic", page_icon=":snowflake:", layout="wide", initial_sidebar_state="collapsed")
    st.title("Arctic Chat App")

# Snowflake接続用の関数
@st.cache_resource(ttl=7200)
def connect_snowflake():
    # Snowflakeの接続情報はStreamlitのシークレット(.streamlit/secret.toml)に保存しておく
    connection = snowflake.connector.connect(
        user=st.secrets["Snowflake"]["user"],
        password=st.secrets["Snowflake"]["password"],
        account=st.secrets["Snowflake"]["account"],
        role=st.secrets["Snowflake"]["role"],
        warehouse=st.secrets["Snowflake"]["warehouse"])

    session = snowpark.Session.builder.configs({"connection": connection}).create()
    return session 

# Arctic実行用の関数
def get_response(session, messages):
    response = session.sql(f'''
    SELECT SNOWFLAKE.CORTEX.COMPLETE('snowflake-arctic',
        {messages},
        {{
            'temperature': 0.3,
            'top_p': 0.9
        }});
        ''').to_pandas().iloc[0,0]
    # レスポンスを辞書型に変換
    response = ast.literal_eval(response)
    response = response["choices"][0]["messages"]
    return response

# Main function
def main():
    init()
    session = connect_snowflake()

    # 履歴保持用の変数を準備
    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"])  

    # 会話受付
    prompt = st.chat_input("何か入力してね！")

    #入力後の動作
    if prompt:
        st.session_state.messages.append({"role": "user", "content": prompt})
        with st.chat_message("user"):
            st.markdown(prompt)
        
        messages=[]
        for m in st.session_state.messages:
              messages.append({"role": m["role"], "content": m["content"]})

        # Arctic実行
        response = get_response(session, messages)
        st.session_state.messages.append({"role": "assistant", "content": response})
        with st.chat_message("assistant"):
            st.markdown(response)

if __name__ == '__main__':
    main()

終わりに

履歴を保持するとトークン数が増えるので、履歴を保持せずに過去の会話の流れを要約して渡す方法もよいかもしれません。
そのあたりは今後も継続して勉強していこうと思います💪