前回、OCI で AI チャットを作りました。
あれはあれで、良いんですが、もうちょっと頑張って 履歴保存が出来るようにしてみました。
履歴保存するには、ユーザーを管理しないと出来ないんで、googleさんに認証してもらうことにしました。
別に、streamlit が対応してりゃ、他でも出来ると思います。
但し、ユーザーを一意識別する方法は、調べないとダメですよ。
ということで、なんちゃって 履歴保存付き AI チャットです。
履歴保存には、OCIの NoSQL を使いました。
しかしこれ、NoSQLなのに、がっつり SQL使ってるんですけど、どういうこと。
まー、良いんですけど、Deleteとかって、何故か一括で削除出来ないとか、仕様がよくわからんです。
ということで、これが完成ソースだー
import streamlit as st
import oci
from oci.generative_ai import GenerativeAiClient
from oci.generative_ai_inference import GenerativeAiInferenceClient
from oci.generative_ai_inference.models import (
ChatDetails,
CohereChatRequest,
OnDemandServingMode
)
import uuid
import datetime
import pytz
import hashlib
# テーマの取得
theme = "dark" if st.config.get_option("theme.base") == "dark" else "light"
# モバイル表示の問題を修正
# テーマに応じたCSSを適用
st.markdown(f"""
<style>
@media (max-width: 800px) {{
.stChatInput {{
position: fixed;
bottom: 0;
left: 0;
width: 100%;
padding: 10px;
z-index: 1000;
transition: background-color 0.3s ease;
}}
.stChatInput textarea {{
width: 100%;
box-sizing: border-box;
}}
/* Lightモードのスタイル */
.{theme}-mode .stChatInput {{
background-color: #ffffff;
border-top: 1px solid #ccc;
box-shadow: 0 -2px 5px rgba(0, 0, 0, 0.1);
}}
/* Darkモードのスタイル */
.dark-mode .stChatInput {{
background-color: #1e1e1e;
border-top: 1px solid #444;
box-shadow: 0 -2px 5px rgba(0, 0, 0, 0.5);
}}
.dark-mode .stChatInput textarea {{
color: #ffffff;
background-color: #1e1e1e;
}}
}}
</style>
""", unsafe_allow_html=True)
# --- OCI NoSQL Database の設定 ---
NOSQL_TABLE_NAME = "ChatHistory" # NoSQLテーブル名
# --- OCI設定ファイルのロード ---
config = oci.config.from_file("~/.oci/config", "DEFAULT")
COMPARTMENT_ID = "ocid1.compartment.oc1..aaaaaaaamgmw22hogecwqnunirb3urhoger4ihdgoilkdjkv2sabokaq5svc"
# NoSQLクライアントの初期化
nosqlcl : oci.nosql.nosql_client.NosqlClient = oci.nosql.nosql_client.NosqlClient(config)
# Generative AI クライアントの初期化
DEFAULT_MODEL = "cohere.command-a-03-2025"
client = GenerativeAiInferenceClient(config=config)
generative_ai_client = GenerativeAiClient(config)
jst_timezone = pytz.timezone('Asia/Tokyo')
st.markdown("""
<style>
@media (max-width: 800px) {
.stChatInput {
position: fixed;
bottom: 0;
left: 0;
width: 100%;
background-color: white;
padding: 10px;
z-index: 1000;
}
}
</style>
""", unsafe_allow_html=True)
#日時変換->JST
def parseDateTime( tm ) :
return datetime.datetime.fromisoformat(tm.replace('Z', '+00:00')).astimezone(jst_timezone)
# セッションID生成
def generate_unique_session_id() -> str:
"""一意性を保ちつつ、より短いセッションIDを生成する。"""
return hashlib.md5(str(uuid.uuid4()).encode('utf-8')).hexdigest()[:12]
# 索引作成
def create_session_id_index_if_not_exists():
print(f"NoSQL インデックス 'idx_session_id' を確認・作成中...")
try:
index_name = "idx_session_id"
create_index_details = oci.nosql.models.CreateIndexDetails(
name=index_name,
compartment_id=COMPARTMENT_ID,
keys=[
oci.nosql.models.IndexKey(column_name="user_id"), # user_idも指定してクエリと合わせる
oci.nosql.models.IndexKey(column_name="session_id")
],
is_if_not_exists = True
)
response: oci.response.Response = nosqlcl.create_index(table_name_or_id=NOSQL_TABLE_NAME,create_index_details=create_index_details)
print(f"インデックス作成リクエスト送信済み。ワークリクエストID: {response.request_id}")
oci.wait_until(
nosqlcl,
nosqlcl.get_index(table_name_or_id=NOSQL_TABLE_NAME, index_name=index_name),
'lifecycle_state',
'ACTIVE'
)
print(f"インデックス '{index_name}' が ACTIVE 状態になりました。")
except oci.exceptions.ServiceError as e:
if e.code == 'IndexAlreadyExists':
print(f"インデックス '{index_name}' は既に存在します。")
else:
print(f"インデックス作成中にエラーが発生しました: {e}")
print(f"詳細エラーメッセージ: {e.message}")
except Exception as e:
print(f"予期せぬエラー: {e}")
# テーブル作成
def create_chat_history_table_if_not_exists():
"""チャット履歴テーブルが存在しない場合に作成する"""
print(f"NoSQL テーブル '{NOSQL_TABLE_NAME}' を確認・作成中...")
try:
try:
print(f"NoSQL テーブル '{NOSQL_TABLE_NAME}' を確認中...")
# テーブルが既に存在するか確認
response: oci.response.Response = nosqlcl.get_table(table_name_or_id=NOSQL_TABLE_NAME, compartment_id=COMPARTMENT_ID)
tbl : oci.nosql.models.Table = response.data
if tbl.lifecycle_state == oci.nosql.models.Table.LIFECYCLE_STATE_ACTIVE:
print(f"テーブル '{NOSQL_TABLE_NAME}' は既に存在します。")
return
except oci.exceptions.ServiceError as e:
if e.message.startswith('Table not found') == False:
print(f"テーブル確認中にエラーが発生しました: {e.message}")
return
print(f"NoSQL テーブル '{NOSQL_TABLE_NAME}' を作成中...")
# TableLimitsを定義 (プロビジョニング容量の例)
table_limits = oci.nosql.models.TableLimits(
max_read_units=10,
max_write_units=5,
max_storage_in_g_bs=1,
capacity_mode = oci.nosql.models.TableLimits.CAPACITY_MODE_PROVISIONED
)
ddl_statement = f"""
CREATE TABLE {NOSQL_TABLE_NAME} (
user_id STRING,
session_id STRING,
message_timestamp TIMESTAMP(3),
role STRING,
message STRING,
PRIMARY KEY (SHARD(user_id), session_id, message_timestamp)
) USING TTL 90 days
"""
create_table_details = oci.nosql.models.CreateTableDetails(
name=NOSQL_TABLE_NAME,
compartment_id=COMPARTMENT_ID,
ddl_statement=ddl_statement,
table_limits=table_limits
)
response: oci.response.Response = nosqlcl.create_table(create_table_details)
print(f"テーブル作成リクエスト送信済み。ワークリクエストID: {response.request_id}")
oci.wait_until(
nosqlcl,
nosqlcl.get_table(table_name_or_id=NOSQL_TABLE_NAME, compartment_id=COMPARTMENT_ID),
'lifecycle_state',
'ACTIVE'
)
print(f"テーブル '{NOSQL_TABLE_NAME}' が ACTIVE 状態になりました。")
create_session_id_index_if_not_exists()
except oci.exceptions.ServiceError as e:
if e.code == 'TableAlreadyExists':
print(f"テーブル '{NOSQL_TABLE_NAME}' は既に存在します。")
else:
print(f"テーブル作成中にエラーが発生しました: {e}")
print(f"詳細エラーメッセージ: {e.message}")
except Exception as e:
print(f"予期せぬエラー: {e}")
# チャット履歴保存
def save_chat_message(user_id: str, session_id: str, role: str, message: str):
"""チャットメッセージをNoSQL Databaseに保存する"""
try:
timestamp = datetime.datetime.now(pytz.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
row_data = {
'user_id': user_id,
'session_id': session_id,
'message_timestamp': timestamp,
'role': role,
'message': message
}
put_row_details = oci.nosql.models.UpdateRowDetails(
compartment_id=COMPARTMENT_ID,
value=row_data
)
nosqlcl.update_row(
table_name_or_id=NOSQL_TABLE_NAME,
update_row_details=put_row_details
)
except Exception as e:
print(f"NoSQL: メッセージ保存中にエラーが発生しました: {e}")
# チャット履歴ロード (特定のセッションIDを指定)
def load_chat_history_for_session(user_id: str, session_id: str):
"""NoSQL Databaseから特定のセッションのチャット履歴を読み込む"""
print(f"NoSQL: ユーザー {user_id}, セッション {session_id} の履歴を読み込み中...")
history = []
try:
query_statement = f"""
SELECT role, message, message_timestamp
FROM {NOSQL_TABLE_NAME}
WHERE user_id = '{user_id}' AND session_id = '{session_id}'
"""
query_details = oci.nosql.models.QueryDetails(
compartment_id=COMPARTMENT_ID,
statement=query_statement
)
response: oci.response.Response = nosqlcl.query(query_details)
result: oci.nosql.models.QueryResultCollection = response.data
for item in result.items:
st_role = "assistant" if item['role'] == "CHATBOT" else "user"
history.append({"role": st_role, "message": item['message']})
print(f"NoSQL: {len(history)} 件の履歴を読み込みました。")
except Exception as e:
print(f"NoSQL: 履歴読み込み中にエラーが発生しました: {e}")
return history
# ユーザーの全セッションIDを取得する新しい関数
def get_user_session_ids(user_id: str):
"""指定されたユーザーのすべてのチャットセッションIDをNoSQL Databaseから取得する"""
print(f"NoSQL: ユーザー {user_id} のセッションIDを検索中...")
session_ids = dict()
try:
# DISTINCTキーワードを使用して、重複しないsession_idを取得
query_statement = f"""
SELECT session_id, min(message_timestamp) as timestamp
FROM {NOSQL_TABLE_NAME}
WHERE user_id = '{user_id}'
GROUP BY session_id
"""
query_details = oci.nosql.models.QueryDetails(
compartment_id=COMPARTMENT_ID,
statement=query_statement
)
response: oci.response.Response = nosqlcl.query(query_details)
result: oci.nosql.models.QueryResultCollection = response.data
for item in result.items:
s_id = item['session_id']
t_stamp = item['timestamp']
session_ids[s_id] = t_stamp
print(f"NoSQL: {len(session_ids)} 件のセッションIDを検出しました。")
except Exception as e:
print(f"NoSQL: セッションID取得中にエラーが発生しました: {e}")
return session_ids
# ユーザーの指定セッションを削除する
def delete_user_session(user_id: str, session_id: str):
"""指定されたユーザーの特定のチャットセッションをNoSQL Databaseから削除する"""
print(f"NoSQL: ユーザー {user_id}, セッション {session_id} を削除中...")
try:
query_statement = f"""
SELECT role, message, message_timestamp
FROM {NOSQL_TABLE_NAME}
WHERE user_id = '{user_id}' AND session_id = '{session_id}'
"""
query_details = oci.nosql.models.QueryDetails(
compartment_id=COMPARTMENT_ID,
statement=query_statement
)
response: oci.response.Response = nosqlcl.query(query_details)
result: oci.nosql.models.QueryResultCollection = response.data
for item in result.items:
response: oci.response.Response = nosqlcl.delete_row(
compartment_id=COMPARTMENT_ID,
table_name_or_id=NOSQL_TABLE_NAME,
key=[f"user_id:{user_id}",f"session_id:{session_id}", f"message_timestamp:{item['message_timestamp']}"]
)
drr:oci.nosql.models.DeleteRowResult = response.data
print(f"NoSQL: セッション {session_id} の削除が完了しました。")
except Exception as e:
print(f"NoSQL: セッション削除中にエラーが発生しました: {e}")
# 許可確認する
def isContain(oid) :
return True
#モデル一覧
available_models = []
ret:oci.response.Response = generative_ai_client.list_models( compartment_id=COMPARTMENT_ID)
models:oci.generative_ai.models.ModelCollection = ret.data
model:oci.generative_ai.models.Model
for model in models.items:
if( "CHAT" in model.capabilities ):
available_models.append(model.display_name)
#タイトル
st.title("OCI AI Chat")
if not st.user.is_logged_in:
st.title("ログインしてください")
if st.button("Googleでログイン"):
st.login("google")
st.stop()
else:
oid = st.user.get("sub")
if 'nosql_table_checked' not in st.session_state:
create_chat_history_table_if_not_exists()
st.session_state.nosql_table_checked = True
if 'current_chat_session_id' not in st.session_state:
st.session_state.current_chat_session_id = None
if 'messages_loaded_for_session' not in st.session_state:
st.session_state.messages_loaded_for_session = None
# ログアウトボタン
if st.button("ログアウト"):
st.logout()
# 利用可能権限チェック
if( isContain(oid) == False ) :
st.write("許可されていません")
else :
st.sidebar.header(f"Login: {st.user.name}")
selected_model = st.sidebar.selectbox("使用するモデルを選択", available_models, index=available_models.index(DEFAULT_MODEL) if DEFAULT_MODEL in available_models else 0)
# セッションIDの管理と選択
# ユーザーの全セッションIDを取得
all_session_ids: dict = get_user_session_ids(oid)
# 新しいセッションを開始するためのオプションを追加
NEWCHAT = "新しいチャットを開始"
options : dict = dict()
options["-1"] = NEWCHAT
for s_id,timestamp in all_session_ids.items():
options[s_id] = parseDateTime(timestamp)
# サイドバーでセッションを選択
selected_session_option = st.sidebar.selectbox(
"過去チャットを選択",
options,
index=0,
format_func = lambda s_id: f"{options[s_id]}",
key="session_select_box"
)
print(f"{selected_model},{selected_session_option},{st.session_state.current_chat_session_id}")
# 新しいセッションIDが既存のものと異なる場合のみリセット
if selected_session_option == "-1": # 仮のIDを設定して区別
if st.session_state.messages_loaded_for_session is None and st.session_state.current_chat_session_id is not None:
#新規で継続中
st.session_state.messages = load_chat_history_for_session(oid, st.session_state.current_chat_session_id)
else :
st.session_state.current_chat_session_id = generate_unique_session_id()
st.session_state.messages = []
st.session_state.messages_loaded_for_session = None
else:
print(f"履歴ロード {options[selected_session_option]}")
# 選択された既存のセッションIDをロード
if st.session_state.current_chat_session_id != selected_session_option:
st.session_state.current_chat_session_id = selected_session_option
st.session_state.messages = load_chat_history_for_session(oid, st.session_state.current_chat_session_id)
st.session_state.messages_loaded_for_session = selected_session_option
# セッションのリセットボタン
if st.session_state.messages_loaded_for_session is None and st.session_state.current_chat_session_id is not None:
if st.sidebar.button("リセット"):
st.session_state.current_chat_session_id = None
st.session_state.messages = []
st.session_state.messages_loaded_for_session = None
st.rerun()
# 選択されたセッション履歴を削除
if selected_session_option != "-1":
if st.sidebar.button("削除"):
delete_user_session(oid, st.session_state.current_chat_session_id)
st.session_state.current_chat_session_id = None
st.session_state.messages = []
st.session_state.messages_loaded_for_session = None
st.rerun()
# チャット履歴表示
for message in st.session_state.messages:
role = "assistant" if message["role"] == "assistant" else "user"
with st.chat_message(role):
st.markdown(message["message"])
# チャット 入力待ち
if prompt := st.chat_input("ここにメッセージを入力してください..."):
# セッション チャット履歴追加
st.session_state.messages.append({"role": "user", "message": prompt})
# DB チャット履歴追加
save_chat_message(oid, st.session_state.current_chat_session_id, "USER", prompt)
with st.chat_message("user"):
st.markdown(prompt)
with st.chat_message("assistant"):
with st.spinner("思考中..."):
chat_history = []
for message in st.session_state.messages:
if message["role"] == "user":
chat_history.append({"role": "USER", "message": message["message"]})
elif message["role"] == "assistant":
chat_history.append({"role": "CHATBOT", "message": message["message"]})
chat_request = CohereChatRequest(
message=prompt,
chat_history=chat_history[:-1] if chat_history else None,
max_tokens=3000,
temperature=0.7,
is_echo=True,
is_stream=False
)
serving_mode = OnDemandServingMode(model_id=selected_model)
chat_details = ChatDetails(
compartment_id=COMPARTMENT_ID,
chat_request=chat_request,
serving_mode=serving_mode
)
response = client.chat(chat_details)
bot_reply = response.data.chat_response.text
if bot_reply:
# セッション チャット履歴追加
st.session_state.messages.append({"role": "assistant", "message": bot_reply})
# DB チャット履歴追加
save_chat_message(oid, st.session_state.current_chat_session_id, "CHATBOT", bot_reply)
# 出力
st.markdown(bot_reply)
認証SSOは、GCP で、OAuth 2.0 クライアント ID を取得して 良い感じの 承認済みのリダイレクト URI を指定すればOkです。
./streamlit/secrets.toml
後は、ChatGPT とか、Gemeni に聞けばわかります。
こんな感じ Darkモードですね
