前回、OCI で AI チャットを作りました。
あれはあれで、良いんですが、もうちょっと頑張って 履歴保存が出来るようにしてみました。
履歴保存するには、ユーザーを管理しないと出来ないんで、googleさんに認証してもらうことにしました。
別に、streamlit が対応してりゃ、他でも出来ると思います。
但し、ユーザーを一意識別する方法は、調べないとダメですよ。
ということで、なんちゃって 履歴保存付き AI チャットです。
履歴保存には、OCIの NoSQL を使いました。
しかしこれ、NoSQLなのに、がっつり SQL使ってるんですけど、どういうこと。
まー、良いんですけど、Deleteとかって、何故か一括で削除出来ないとか、仕様がよくわからんです。
ということで、これが完成ソースだー
table_limits は、費用バランスを考慮して修正してね。
エクスポート、インポート機能を追加しました。
chatdb.py
import oci
import datetime
import time
import pytz
class chatdb:
tablename = "ChatHistory2"
childtablename = "messages"
nosqlcl = None
compartmentid = None
retry_count = 5
sleep_time = 5
def __init__(self, config, compid ):
self.nosqlcl : oci.nosql.nosql_client.NosqlClient = oci.nosql.nosql_client.NosqlClient(config)
self.compartmentid = compid
# 索引作成
def _create_table_index(self ):
print(f"NoSQL インデックス 'message' を確認・作成中...")
try:
index_name = "message"
create_index_details = oci.nosql.models.CreateIndexDetails(
name=index_name,
compartment_id=self.compartmentid,
keys=[
oci.nosql.models.IndexKey(column_name="user_id"),
oci.nosql.models.IndexKey(column_name="message_timestamp")
],
is_if_not_exists = True
)
response: oci.response.Response = self.nosqlcl.create_index(table_name_or_id=self.tablename ,create_index_details=create_index_details)
oci.wait_until(
self.nosqlcl,
self.nosqlcl.get_index(table_name_or_id=self.tablename, index_name=index_name, compartment_id=self.compartmentid),
'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_table(self ):
print(f"NoSQL テーブル '{self.tablename}' を確認・作成中...")
try:
try:
print(f"NoSQL テーブル '{self.tablename}' を確認中...")
# テーブルが既に存在するか確認
response: oci.response.Response = self.nosqlcl.get_table(table_name_or_id=self.tablename, compartment_id=self.compartmentid)
tbl : oci.nosql.models.Table = response.data
if tbl.lifecycle_state == oci.nosql.models.Table.LIFECYCLE_STATE_ACTIVE:
print(f"テーブル '{self.tablename}' は既に存在します。")
return
except oci.exceptions.ServiceError as e:
if e.message.startswith('Table not found') == False:
print(f"テーブル確認中にエラーが発生しました: {e.message}")
return
print(f"NoSQL テーブル '{self.tablename}' を作成中...")
# TableLimitsを定義 (プロビジョニング容量の例)
table_limits = oci.nosql.models.TableLimits(
max_read_units=8,
max_write_units=8,
max_storage_in_g_bs=1,
capacity_mode = oci.nosql.models.TableLimits.CAPACITY_MODE_PROVISIONED
)
ddl_statement = f"""
CREATE TABLE {self.tablename} (
user_id STRING,
session_id STRING,
message_timestamp TIMESTAMP(3),
title STRING,
PRIMARY KEY (SHARD(user_id), session_id)
) USING TTL 90 days
"""
create_table_details = oci.nosql.models.CreateTableDetails(
name=self.tablename,
compartment_id=self.compartmentid,
ddl_statement=ddl_statement,
table_limits=table_limits
)
response: oci.response.Response = self.nosqlcl.create_table(create_table_details)
print(f"テーブル作成リクエスト送信済み。ワークリクエストID: {response.request_id}")
oci.wait_until(
self.nosqlcl,
self.nosqlcl.get_table(table_name_or_id=self.tablename, compartment_id=self.compartmentid),
'lifecycle_state',
'ACTIVE'
)
print(f"テーブル '{self.tablename}' が ACTIVE 状態になりました。")
except oci.exceptions.ServiceError as e:
if e.code == 'TableAlreadyExists':
print(f"テーブル '{self.tablename}' は既に存在します。")
else:
print(f"テーブル作成中にエラーが発生しました: {e}")
print(f"詳細エラーメッセージ: {e.message}")
except Exception as e:
print(f"予期せぬエラー: {e}")
# 子テーブル作成
def _create_child_table( self):
fulltablename = self.tablename + "." + self.childtablename
print(f"NoSQL 子テーブル '{fulltablename}' を確認・作成中...")
try:
try:
print(f"NoSQL 子テーブル '{fulltablename}' を確認中...")
# テーブルが既に存在するか確認
response: oci.response.Response = self.nosqlcl.get_table(table_name_or_id=fulltablename, compartment_id=self.compartmentid)
tbl : oci.nosql.models.Table = response.data
if tbl.lifecycle_state == oci.nosql.models.Table.LIFECYCLE_STATE_ACTIVE:
print(f"子テーブル '{fulltablename}' は既に存在します。")
return
except oci.exceptions.ServiceError as e:
if e.message.startswith('Table not found') == False:
print(f"テーブル確認中にエラーが発生しました: {e.message}")
return
print(f"NoSQL 子テーブル '{fulltablename}' を作成中...")
ddl_statement = f"""
CREATE TABLE {fulltablename} (
message_timestamp TIMESTAMP(3),
role STRING,
message STRING,
PRIMARY KEY (message_timestamp)
) USING TTL 90 days
"""
create_table_details = oci.nosql.models.CreateTableDetails(
name=fulltablename,
compartment_id=self.compartmentid,
ddl_statement=ddl_statement
)
response: oci.response.Response = self.nosqlcl.create_table(create_table_details)
print(f"子テーブル作成リクエスト送信済み。ワークリクエストID: {response.request_id}")
oci.wait_until(
self.nosqlcl,
self.nosqlcl.get_table(table_name_or_id=fulltablename, compartment_id=self.compartmentid),
'lifecycle_state',
'ACTIVE'
)
print(f"子テーブル '{fulltablename}' が ACTIVE 状態になりました。")
except oci.exceptions.ServiceError as e:
if e.code == 'TableAlreadyExists':
print(f"子テーブル '{fulltablename}' は既に存在します。")
else:
print(f"子テーブル作成中にエラーが発生しました: {e}")
print(f"詳細エラーメッセージ: {e.message}")
except Exception as e:
print(f"予期せぬエラー: {e}")
# チャット履歴保存
def save_chat_message(self, user_id: str, session_id: str, role: str, message: str, title: str):
fulltablename = self.tablename + "." + self.childtablename
timestamp = datetime.datetime.now(pytz.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
# 親テーブル記録
for i in range(self.retry_count):
try:
row_data = {
'user_id': user_id,
'session_id': session_id,
'message_timestamp': timestamp,
'title': title
}
put_row_details = oci.nosql.models.UpdateRowDetails(
compartment_id=self.compartmentid,
value=row_data
)
self.nosqlcl.update_row(
table_name_or_id=self.tablename,
update_row_details=put_row_details
)
break # 成功したらループを抜ける
except oci.exceptions.ServiceError as e:
if e.status == 429:
print(f"NoSQL: 要求が多すぎるようだ")
time.sleep(self.sleep_time)
else:
break
except Exception as e:
print(f"NoSQL: メッセージ保存中にエラーが発生しました: {e}")
break # エラーが発生したらループを抜ける
# メッセージ記録
for i in range(self.retry_count):
try:
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=self.compartmentid,
value=row_data
)
self.nosqlcl.update_row(
table_name_or_id=fulltablename,
update_row_details=put_row_details
)
break # 成功したらループを抜ける
except oci.exceptions.ServiceError as e:
if e.status == 429:
print(f"NoSQL: 要求が多すぎるようだ")
time.sleep(self.sleep_time)
else:
break
except Exception as e:
print(f"NoSQL: メッセージ保存中にエラーが発生しました: {e}")
break # エラーが発生したらループを抜ける
# チャット履歴ロード (特定のセッションIDを指定)
def load_chat_history_for_session(self, user_id: str, session_id: str):
fulltablename = self.tablename + "." + self.childtablename
print(f"NoSQL: ユーザー {user_id}, セッション {session_id} の履歴を読み込み中...")
for i in range(self.retry_count):
history = []
try:
query_statement = f"""
SELECT role, message, message_timestamp
FROM {fulltablename}
WHERE user_id = '{user_id}' AND session_id = '{session_id}'
ORDER BY user_id ASC, session_id ASC, message_timestamp ASC
"""
query_details = oci.nosql.models.QueryDetails(
compartment_id=self.compartmentid,
statement=query_statement
)
response: oci.response.Response = self.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 oci.exceptions.ServiceError as e:
if e.status == 429:
print(f"NoSQL: 要求が多すぎるようだ")
time.sleep(self.sleep_time)
else:
break
except Exception as e:
print(f"NoSQL: 履歴読み込み中にエラーが発生しました: {e}")
break
return history
# ユーザーの指定セッションを削除する
def delete_user_session(self, user_id: str, session_id: str):
fulltablename = self.tablename + "." + self.childtablename
print(f"NoSQL: ユーザー {user_id}, セッション {session_id} を削除中...")
# メッセージ削除
for i in range(self.retry_count):
try:
query_statement = f"""
SELECT role, message, message_timestamp
FROM {fulltablename}
WHERE user_id = '{user_id}' AND session_id = '{session_id}'
"""
query_details = oci.nosql.models.QueryDetails(
compartment_id=self.compartmentid,
statement=query_statement
)
response: oci.response.Response = self.nosqlcl.query(query_details)
result: oci.nosql.models.QueryResultCollection = response.data
for item in result.items:
response: oci.response.Response = self.nosqlcl.delete_row(
compartment_id=self.compartmentid,
table_name_or_id=fulltablename,
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} の削除が完了しました。")
break # 成功したらループを抜ける
except oci.exceptions.ServiceError as e:
if e.status == 429:
print(f"NoSQL: 要求が多すぎるようだ")
time.sleep(self.sleep_time)
else:
break
except Exception as e:
print(f"NoSQL: セッション削除中にエラーが発生しました: {e}")
break # エラーが発生したらループを抜ける
# 親テーブル削除
for i in range(self.retry_count):
try:
query_statement = f"""
SELECT title
FROM {self.tablename}
WHERE user_id = '{user_id}' AND session_id = '{session_id}'
"""
query_details = oci.nosql.models.QueryDetails(
compartment_id=self.compartmentid,
statement=query_statement
)
response: oci.response.Response = self.nosqlcl.query(query_details)
result: oci.nosql.models.QueryResultCollection = response.data
for item in result.items:
response: oci.response.Response = self.nosqlcl.delete_row(
compartment_id=self.compartmentid,
table_name_or_id=self.tablename,
key=[f"user_id:{user_id}",f"session_id:{session_id}"]
)
drr:oci.nosql.models.DeleteRowResult = response.data
print(f"NoSQL: セッション {session_id} の削除が完了しました。")
break # 成功したらループを抜ける
except oci.exceptions.ServiceError as e:
if e.status == 429:
print(f"NoSQL: 要求が多すぎるようだ")
time.sleep(self.sleep_time)
else:
break
except Exception as e:
print(f"NoSQL: セッション削除中にエラーが発生しました: {e}")
break # エラーが発生したらループを抜ける
# ユーザーの全セッションIDを取得する新しい関数
def get_user_session_ids(self, user_id: str):
print(f"NoSQL: ユーザー {user_id} のセッションIDを検索中...")
for i in range(self.retry_count):
session_ids = []
try:
# DISTINCTキーワードを使用して、重複しないsession_idを取得
query_statement = f"""
SELECT session_id, message_timestamp, title
FROM {self.tablename}
WHERE user_id = '{user_id}'
ORDER BY user_id DESC, message_timestamp DESC
"""
query_details = oci.nosql.models.QueryDetails(
compartment_id=self.compartmentid,
statement=query_statement
)
response: oci.response.Response = self.nosqlcl.query(query_details)
result: oci.nosql.models.QueryResultCollection = response.data
for item in result.items:
s_id = item['session_id']
message_timestamp = item['message_timestamp']
title = item['title']
session_ids.append( [s_id, message_timestamp, title] )
print(f"NoSQL: {len(session_ids)} 件のセッションIDを検出しました。")
except oci.exceptions.ServiceError as e:
if e.status == 429:
print(f"NoSQL: 要求が多すぎるようだ")
time.sleep(self.sleep_time)
else:
break
except Exception as e:
print(f"NoSQL: セッションID取得中にエラーが発生しました: {e}")
break
return session_ids
def createtable( self ) :
self._create_table()
self._create_child_table()
self._create_table_index()
app.py
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 time
import pytz
import hashlib
import json
from chatdb import chatdb
# テーマの取得
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認証情報
config = oci.config.from_file("~/.oci/config", "DEFAULT")
COMPARTMENT_ID = "ocid1.compartment.oc1..aaaaaaaamgmw22hogecwqnunirb3urhoger4ihdgoilkdjkv2sabokaq5svc"
# Google認証
LOGINBTN = "Googleでログイン"
AUTHSECTION = "google"
# 認証ID識別子
def AUTHID(user) :
return user.get("sub")
# 許可確認する
def isContain(oid) :
return True
# チャットDB
db = chatdb(config,COMPARTMENT_ID)
# Generative AI クライアントの初期化
DEFAULT_MODEL = "cohere.command-latest"
client = GenerativeAiInferenceClient(config=config)
generative_ai_client = GenerativeAiClient(config)
# 日本タイムゾーン
jst_timezone = pytz.timezone('Asia/Tokyo')
#日時変換->JST
def parseDateTime( tm ) :
return datetime.datetime.fromisoformat(tm.replace('Z', '+00:00')).astimezone(jst_timezone)
# セッションID生成
def generate_unique_session_id() -> str:
return hashlib.md5(str(uuid.uuid4()).encode('utf-8')).hexdigest()[:12]
# エキスポート関数群
def export_as_text(chat_history, session_id, title):
"""テキスト形式でエクスポート"""
content = f"チャット履歴: {title}\nセッションID: {session_id}\n"
content += f"エクスポート日時: {datetime.datetime.now(jst_timezone).strftime('%Y-%m-%d %H:%M:%S')}\n"
content += "=" * 50 + "\n\n"
for msg in chat_history:
timestamp = parseDateTime(msg.get('timestamp', datetime.datetime.now().isoformat())) if 'timestamp' in msg else ""
role_display = "ユーザー" if msg['role'].upper() == "USER" else "アシスタント"
content += f"[{timestamp.strftime('%Y-%m-%d %H:%M:%S') if timestamp else 'N/A'}] {role_display}:\n"
content += f"{msg['message']}\n\n"
return content
def export_as_json(chat_history, session_id, title):
"""JSON形式でエクスポート"""
export_data = {
"session_id": session_id,
"title": title,
"export_datetime": datetime.datetime.now(jst_timezone).isoformat(),
"messages": []
}
for msg in chat_history:
timestamp = parseDateTime(msg.get('timestamp', datetime.datetime.now().isoformat())) if 'timestamp' in msg else None
export_data["messages"].append({
"role": msg['role'],
"message": msg['message'],
"timestamp": timestamp.isoformat() if timestamp else None
})
return json.dumps(export_data, ensure_ascii=False, indent=2)
def export_as_markdown(chat_history, session_id, title):
"""Markdown形式でエクスポート"""
content = f"# {title}\n\n"
content += f"**セッションID:** {session_id} \n"
content += f"**エクスポート日時:** {datetime.datetime.now(jst_timezone).strftime('%Y-%m-%d %H:%M:%S')} \n\n"
content += "---\n\n"
for i, msg in enumerate(chat_history, 1):
timestamp = parseDateTime(msg.get('timestamp', datetime.datetime.now().isoformat())) if 'timestamp' in msg else ""
role_display = "🧑 ユーザー" if msg['role'].upper() == "USER" else "🤖 アシスタント"
content += f"## {i}. {role_display}\n\n"
if timestamp:
content += f"*{timestamp.strftime('%Y-%m-%d %H:%M:%S')}*\n\n"
content += f"{msg['message']}\n\n"
content += "---\n\n"
return content
def import_from_json(json_content, oid):
"""JSON形式からチャット履歴をインポート"""
try:
import_data = json.loads(json_content)
# 必須フィールドの検証
if not all(key in import_data for key in ["session_id", "title", "messages"]):
return False, "不正なJSONフォーマットです。必須フィールドが不足しています。"
# 新しいセッションIDを生成（重複を避けるため）
new_session_id = generate_unique_session_id()
title = import_data.get("title", "インポートされたチャット")
# メッセージの検証とインポート
imported_count = 0
for msg in import_data["messages"]:
if "role" in msg and "message" in msg:
role = msg["role"].upper()
if role in ["USER", "CHATBOT", "ASSISTANT"]:
# ASSISTANTをCHATBOTに変換
if role == "ASSISTANT":
role = "CHATBOT"
# メッセージの保存
db.save_chat_message(oid, new_session_id, role, msg["message"], title)
imported_count += 1
print(f"インポート: セッションID={new_session_id}, ロール={role}, カウント={imported_count}")
if imported_count > 0:
return True, f"正常にインポートされました。{imported_count}件のメッセージをインポートしました。"
else:
return False, "有効なメッセージが見つかりませんでした。"
except json.JSONDecodeError:
return False, "JSON形式が不正です。"
except Exception as e:
return False, f"インポート中にエラーが発生しました: {str(e)}"
def validate_json_format(json_content):
"""JSONファイルの形式を事前検証"""
try:
data = json.loads(json_content)
# 基本構造の確認
required_fields = ["session_id", "title", "messages"]
missing_fields = [field for field in required_fields if field not in data]
if missing_fields:
return False, f"必須フィールドが不足しています: {', '.join(missing_fields)}"
# メッセージの構造確認
if not isinstance(data["messages"], list):
return False, "messages は配列である必要があります。"
valid_messages = 0
for i, msg in enumerate(data["messages"]):
if isinstance(msg, dict) and "role" in msg and "message" in msg:
if msg["role"].upper() in ["USER", "CHATBOT", "ASSISTANT"]:
valid_messages += 1
if valid_messages == 0:
return False, "有効なメッセージが見つかりません。"
return True, f"有効なJSONファイルです。{valid_messages}件のメッセージが含まれています。"
except json.JSONDecodeError:
return False, "JSON形式が不正です。"
except Exception as e:
return False, f"検証エラー: {str(e)}"
#モデル一覧
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(LOGINBTN):
st.login(AUTHSECTION)
st.stop()
else:
oid = AUTHID(st.user)
if 'nosql_table_checked' not in st.session_state:
db.createtable()
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を取得
all_session_ids: list = db.get_user_session_ids(oid)
# 新しいセッションを開始するためのオプションを追加
NEWCHAT = "新しいチャットを開始"
# 全セッション追加
options = []
options.append( ["-1", NEWCHAT, NEWCHAT] )
for item in all_session_ids :
session_id = item[0]
jst_timestamp = parseDateTime(item[1])
title = item[2]
options.append([session_id,jst_timestamp,title])
# サイドバーでセッションを選択
selected_session_option = st.sidebar.selectbox(
"過去チャットを選択",
options,
index=0,
format_func = lambda item: f"{item[2]}",
key="session_select_box"
)
session_id = selected_session_option[0]
message_timestamp = selected_session_option[1]
title = selected_session_option[2]
print(f"{selected_model},[{session_id}:{message_timestamp}:{title}],{st.session_state.current_chat_session_id}")
# 新しいセッションIDが既存のものと異なる場合のみリセット
if session_id == "-1":
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 = db.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"履歴ロード {session_id}")
# 選択された既存のセッションIDをロード
if st.session_state.current_chat_session_id != session_id:
st.session_state.current_chat_session_id = session_id
st.session_state.messages = db.load_chat_history_for_session(oid, st.session_state.current_chat_session_id)
st.session_state.messages_loaded_for_session = session_id
# セッションのリセットボタン
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 session_id != "-1":
if st.sidebar.button("削除"):
db.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()
# 拡張エキスポート機能
st.sidebar.subheader("エクスポート")
# エクスポート形式選択
export_format = st.sidebar.selectbox(
"形式選択",
["JSON", "Markdown", "テキスト"],
key="export_format"
)
# 単一セッションのエクスポート
if st.sidebar.button("エクスポート準備"):
chat_history = db.load_chat_history_for_session(oid, st.session_state.current_chat_session_id)
if export_format == "テキスト":
content = export_as_text(chat_history, st.session_state.current_chat_session_id, title)
filename = f"chat_{st.session_state.current_chat_session_id}.txt"
mime_type = "text/plain"
elif export_format == "JSON":
content = export_as_json(chat_history, st.session_state.current_chat_session_id, title)
filename = f"chat_{st.session_state.current_chat_session_id}.json"
mime_type = "application/json"
elif export_format == "Markdown":
content = export_as_markdown(chat_history, st.session_state.current_chat_session_id, title)
filename = f"chat_{st.session_state.current_chat_session_id}.md"
mime_type = "text/markdown"
st.sidebar.download_button(
label=f"{export_format}形式でダウンロード",
data=content,
file_name=filename,
mime=mime_type
)
else :
st.sidebar.subheader("インポート")
# ファイルアップローダー
uploaded_file = st.sidebar.file_uploader(
"JSONファイルを選択",
type=['json'],
help="エクスポートしたJSONファイルをアップロードしてください"
)
if uploaded_file is not None:
# ファイル内容を読み込み
json_content = uploaded_file.read().decode('utf-8')
# プレビュー表示
with st.sidebar.expander("ファイル内容プレビュー"):
try:
preview_data = json.loads(json_content)
st.write(f"**タイトル:** {preview_data.get('title', 'N/A')}")
st.write(f"**セッションID:** {preview_data.get('session_id', 'N/A')}")
st.write(f"**メッセージ数:** {len(preview_data.get('messages', []))}")
# 形式検証
is_valid, validation_message = validate_json_format(json_content)
if is_valid:
st.success(validation_message)
else:
st.error(validation_message)
except Exception as e:
st.error(f"プレビューエラー: {str(e)}")
# インポートボタン
if st.sidebar.button("インポート実行", type="primary"):
with st.spinner("インポート中..."):
success, message = import_from_json(json_content, oid)
if success:
st.sidebar.success(message)
# セッション状態をリセットして新しいチャット一覧を表示
time.sleep(1) # ユーザーが成功メッセージを見る時間を与える
st.rerun()
else:
st.sidebar.error(message)
# チャット履歴表示
for message in st.session_state.messages:
role = "assistant" if message["role"] == "assistant" else "user"
with st.chat_message(role):
if role == "assistant" :
st.markdown(message["message"])
else:
st.text(message["message"])
# チャット 入力待ち
if prompt := st.chat_input("ここにメッセージを入力してください..."):
# セッション チャット履歴追加
st.session_state.messages.append({"role": "user", "message": prompt})
# DB チャット履歴追加
jstnow = datetime.datetime.now(jst_timezone).strftime('%Y-%m-%d %H')
title = f"{jstnow} {prompt[:20]}"
db.save_chat_message(oid, st.session_state.current_chat_session_id, "USER", prompt, title)
with st.chat_message("user"):
st.text(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 チャット履歴追加
db.save_chat_message(oid, st.session_state.current_chat_session_id, "CHATBOT", bot_reply, title)
# 出力
st.markdown(bot_reply)
認証SSOは、GCP で、OAuth 2.0 クライアント ID を取得して 良い感じの 承認済みのリダイレクト URI を指定すればOkです。
./streamlit/secrets.toml
後は、ChatGPT とか、Gemeni に聞けばわかります。
こんな感じ Darkモードですね
