import json import os import uuid import threading import time from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo import gradio as gr from huggingface_hub import CommitScheduler from openai import OpenAI from transformers import AutoTokenizer openai_api_key = os.getenv("api_key") openai_api_base = os.getenv("api_url") hf_token = os.getenv("hf_token") model_name = "Aratako/calm3-22b-RP-v2" client = OpenAI( api_key=openai_api_key, base_url=openai_api_base, ) tokenizer = AutoTokenizer.from_pretrained("Aratako/calm3-22b-RP-v2", token=hf_token) # Define the file where to save the data. Use UUID to make sure not to overwrite existing data from a previous run. log_file = Path("logs/") / f"data_{uuid.uuid4()}.json" log_folder = log_file.parent # Schedule regular uploads. Remote repo and local folder are created if they don't already exist. scheduler = CommitScheduler( repo_id="Aratako/calm3-22b-RP-v2-logs", # Replace with your actual repo ID repo_type="dataset", folder_path=log_folder, path_in_repo="data", every=10, # Upload every 10 minutes token=hf_token, ) def save_chat_logs(messages, response): """ Save conversation data in a JSON Lines file. """ with scheduler.lock: entry = { "timestamp": datetime.now(ZoneInfo("Asia/Tokyo")).isoformat(), "messages": messages, "response": response, } with log_file.open("a") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") def count_tokens(messages): return sum(len(tokenizer.encode(msg["content"])) for msg in messages) def trim_messages(messages, max_tokens): # システムメッセージは常に保持 system_message = messages[0] trimmed_messages = [system_message] current_tokens = count_tokens([system_message]) # 最新のメッセージから逆順に追加していく for message in reversed(messages[1:]): message_tokens = count_tokens([message]) if current_tokens + message_tokens <= max_tokens: trimmed_messages.insert(1, message) # システムメッセージの後に挿入 current_tokens += message_tokens else: break # トークン制限を超えたら終了 return trimmed_messages def respond( message, history: list[tuple[str, str]], system_message, max_tokens, temperature, top_p, ): try: messages = [{"role": "system", "content": system_message}] for val in history: if val[0]: messages.append({"role": "user", "content": val[0]}) if val[1]: messages.append({"role": "assistant", "content": val[1]}) messages.append({"role": "user", "content": message}) # メッセージを調整して8192トークン以内に収める max_input_tokens = 8192 - max_tokens trimmed_messages = trim_messages(messages, max_input_tokens) response = "" for chunk in client.chat.completions.create( model=model_name, messages=trimmed_messages, max_tokens=max_tokens, stream=True, temperature=temperature, top_p=top_p, ): token = chunk.choices[0].delta.content if token is not None: response += token yield response # Save conversation after the full response is generated save_chat_logs(trimmed_messages, response) except Exception as e: yield f"エラーが発生しました: {str(e)}" """ For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface """ description = """ ### [Aratako/calm3-22b-RP-v2](https://huggingface.co/Aratako/calm3-22b-RP-v2)のデモです。 - 対話ログを収集します。ただしデータは非公開リポジトリに保存され、私（Aratako）以外の第三者に公開はしません。 - 収集したログやそれを元に合成したデータを今後開発するモデルの学習に利用する可能性があります。ただしその際も本デモが元となるデータは公開しません。 - 特定のシチュエーションや状況での対話を好む方は、積極的に使ってもらうと今後開発するモデルがその状況のRPが上手くなる可能性もあります。 - **上記の条件に同意する場合のみ**､以下のChatbotを利用してください｡ """ HEADER = description FOOTER = """### 注意 - コンテクスト長は8192までです。超えた場合、古い対話から順番に削除されます。""" example_system_message = """（ここにロールプレイの設定を書いてください。以下は一例です。） ---- 今からロールプレイを行いましょう。"桜"というキャラとしてロールプレイしてください。以下に示す設定に従い、キャラに成りきって返答してください。 ### 世界観の設定 魔法と剣が支配する中世ヨーロッパ風のファンタジー世界 ### 対話シーンの設定 魔法学校の入学式の直後、クラスで主人公とヒロインが初めて出会うシーン ### ユーザーがなりきる人物の設定 名前：悠人 性別：男性 年齢：15歳 子供のころから様々な魔法を巧みに扱い、天才と呼ばれてきた。ただここ数年は成長が停滞しており、新たな刺激を求め魔法学校に入学した。 ### あなたがなりきる人物の設定 名前：桜 性別：女性 年齢：15歳 とある大貴族の長女。両親からとても大事に育てられた箱入り娘で、やや世間知らずなところがある。先祖代々伝わる特殊な魔法を操る。 ### 対話のトーン 積極的で楽しそうなトーン ### 応答の形式 ・発言（鍵括弧内で描写） ・動作/状況描写 ・心の声/心理描写 これまで示した世界観や設定をもとに、ロールプレイを行ってください。ユーザー側のセリフやナレーションは書かないでください。""" def run(): chatbot = gr.Chatbot( elem_id="chatbot", scale=1, show_copy_button=True, height="70%", layout="panel", ) with gr.Blocks(fill_height=True, theme="soft") as demo: gr.Markdown(HEADER) chat_interface = gr.ChatInterface( fn=respond, stop_btn="Stop Generation", cache_examples=False, multimodal=False, chatbot=chatbot, additional_inputs_accordion=gr.Accordion( label="Parameters", open=True, render=False ), additional_inputs=[ gr.Textbox( value=example_system_message, label="システムメッセージ （ロールプレイの設定）", render=False, ), gr.Slider( minimum=1, maximum=4096, step=1, value=1024, label="Max tokens", visible=True, render=False, ), gr.Slider( minimum=0, maximum=1, step=0.1, value=0.7, label="Temperature", visible=True, render=False, ), gr.Slider( minimum=0, maximum=1, step=0.1, value=0.9, label="Top-p", visible=True, render=False, ), ], analytics_enabled=False, ) gr.Markdown(FOOTER) demo.queue(max_size=256, api_open=True) demo.launch(share=False, quiet=True) def keep_alive(): while True: try: # ダミーのメッセージを使用してエンドポイントにリクエストを送信 dummy_messages = [{"role": "user", "content": "keep-alive"}] client.chat.completions.create( model=model_name, messages=dummy_messages, max_tokens=1, ) print("Keep-alive request sent.") except Exception as e: print(f"Error in keep_alive: {e}") time.sleep(3600) if __name__ == "__main__": threading.Thread(target=keep_alive, daemon=True).start() run()