Llama-cpp-python研究: StableBeluga2用 Terminalスクリプト
Llama-cpp-pythonのドキュメントを読んでもピンとこないので、githubにあるコードを見取り学習してます。
こちらは難しくてよくわからなかった。
わかりやすい感じだったのはこちら。
特に参考になったのは、こちら
とくに、これを参考 してストリーミングで出力させるあたりを活用させてもらいます。
def chat(
model_name: Annotated[str, typer.Option(help="Model name")] = "go-bruins-v2.Q5_K_M.gguf",
repo_id: Annotated[str, typer.Option(help="Name of the huggingface repo")] = "TheBloke/go-bruins-v2-GGUF",
temp: Annotated[float, typer.Option(help="The model temperature between 0-1. Larger values increase creativity but decrease factuality.")] = 0.2,
top_k: Annotated[int, typer.Option(help="Top k")] = 40,
top_p: Annotated[float, typer.Option(help="Randomly sample at each generation step from the top most likely tokens whose probabilities add up to top_p.")] = 0.95,
repeat_penalty: Annotated[float, typer.Option(help="Penalize the model for repetition. Higher values result in less repetition.")] = 1.1,
max_tokens: Annotated[int, typer.Option(help="The maximum number of tokens to generate.")] = 200,
):
llm = get_model(repo_id, model_name)
messages = [
{
"role": "system",
"content": "You are an assistant who gives answers to everything a user asks.",
},
]
while True:
prompt = input("User:>> ")
messages.append({"role": "user", "content": prompt})
response_stream = llm.create_chat_completion(messages, temperature=temp, max_tokens=max_tokens, top_k=top_k, top_p=top_p, repeat_penalty=repeat_penalty, stream=True)
responses = []
for r in response_stream:
if "content" not in r["choices"][0]["delta"]:
continue
token = r["choices"][0]["delta"]["content"].replace("<0x0A>", "\n")
token = token.replace("<br>", "\n")
print(token, end="", flush=True)
responses.append(token)
print("")
messages.append({"role": "assistant", "content": "".join(responses)})
さて、久しぶりに見た下記のボードです。
stabilityai/StableBeluga2 が相変わらず上位に位置しています。
text-generation-webuiでも触ってみて日本語がとてもいい感じだった記憶（←記事にしてました）があるので、このggufファイルを活用してterminalで動かせるようにすることが今回の目標です。
さて先の引用スクリプトのことについて質問しました。
以下について日本語で教えて。特にdeltaって何かな？
```
responses = []
for r in response_stream:
if "content" not in r["choices"][0]["delta"]:
continue
token = r["choices"][0]["delta"]["content"].replace("<0x0A>", "\n")
token = token.replace("<br>", "\n")
print(token, end="", flush=True)
responses.append(token)
GPT-4の教えは下記のとおり。
このコードは、`response_stream` と呼ばれるストリーム（連続的なデータの流れ）からレスポンス（応答）を取得し、処理するためのPythonコードの一部です。
ここでの `delta` は、ストリームからの各レスポンスに含まれるデータ構造の一部を指していると思われます。通常、`delta` は変更や差分を意味する用語ですが、このコンテキストでは、レスポンスの新しい部分、つまり以前に送信された内容以降に追加されたテキストを指している可能性があります。
コードの説明を日本語で行います：
- `responses = []`: `responses` という空のリストを作成しています。このリストは後ほど、処理済みのトークンを格納するために使用されます。
- `for r in response_stream:`: `response_stream` からレスポンス `r` を1つずつ取り出すためのループを開始します。
- `if "content" not in r["choices"][0]["delta"]:`: `r` というレスポンスの中にある `choices` というキーの最初の要素（`[0]`）の中の `delta` というキーに `content` が含まれていない場合、`continue` によってループの次の繰り返しに進みます。つまり、`content` が存在しない場合は何もしません。
- `token = r["choices"][0]["delta"]["content"].replace("<0x0A>", "\n")`: `content` に含まれる文字列を取得し、`<0x0A>` という文字列を改行文字 (`\n`) に置き換えています。
- `token = token.replace("<br>", "\n")`: 同じく、`<br>` というHTMLの改行タグも改行文字に置き換えています。
- `print(token, end="", flush=True)`: 変更後の `token`（トークン）を出力します。`end=""` は出力後の末尾に追加する文字を空にしており、`flush=True` は出力を即座にコンソールに反映させるための引数です。
- `responses.append(token)`: 処理した `token` を `responses` リストに追加しています。
このコードの目的は、ストリームから受け取ったデータのうち、新しい「コンテンツ」が含まれている部分のみを処理し、それを表示してリストに格納することです。特定の形式でエンコードされた改行を実際の改行文字に変換することも行っています。
わかりやすい説明です。この部分を活用して MLX 用のterminalスクリプトを改造しました。
さらに、ファイルの読み出しもつけてみようと思います。
ファイルから読み込んだアウトプットに含まれるバックスラッシュ `\` は、特殊文字が文字列リテラルの一部として正しく解釈されるためのエスケープ文字です。Pythonの文字列リテラルでは、特定の文字の前にバックスラッシュを置くと、その文字が特別な意味を持つことを示します。例えば、`\n` は改行を示し、`\'` はシングルクオートが文字列の一部であることを示します。
アウトプットに表示されている内容は、Pythonの辞書オブジェクトが文字列に変換されたものです。辞書のキーと値はシングルクオート `'` で囲まれており、辞書を文字列として表示するときには、これらのシングルクオートが文字列の一部としてエスケープされる必要があります。これは、文字列の外側がシングルクオートで囲まれているためです。Pythonは内部のシングルクオートをエスケープするために `\` を使用します。
たとえば、以下の辞書オブジェクトを文字列に変換したときに、シングルクオートがエスケープされます：
```python
{
'role': 'assistant',
'content': "I'm glad you enjoyed it! Is there anything else I can help you with?"
}
```
この辞書を文字列に変換すると、以下のようになります：
```python
"{'role': 'assistant', 'content': 'I\\'m glad you enjoyed it! Is there anything else I can help you with?'}"
```
`I'm` の中のシングルクオートの前にエスケープ文字 `\` が挿入されています。
また、改行文字 `\n` が文字列リテラルの中にあるときは、これもエスケープされて表示されます。これは、改行を表す特殊文字であり、文字列リテラルの中で実際の改行として解釈されます。ファイルからの読み込みにおいては、これらはリテラルの一部として保存されたものとして見られるため、このようなエスケープが発生します。
ファイルに保存する際には、文字列の表現をそのまま保存するか、もしくはJSON形式などで保存することで、読み込み時にエスケープされた文字が表示されるのを避けることができます。例えば、以下のように `json` モジュールを使用して保存・読み込みを行うことができます：
```python
import json
# 辞書をJSON文字列に変換して保存
with open('data.json', 'w') as f:
json.dump(my_dict, f)
# JSON文字列を辞書に変換して読み込み
with open('data.json', 'r') as f:
my_dict = json.load(f)
```
この方法ならば、読み込まれたデータはエスケープされずに、元の辞書オブジェクトとして扱うことができます。
LLMではこの形式が一般的なのかを尋ねました。
はい、対話履歴やその他の構造化データの保存・呼び出しにはJSON形式が一般的に使用されます。JSON（JavaScript Object Notation）は軽量なデータ交換フォーマットであり、その読みやすさとプログラミング言語間での広いサポートにより、多くのアプリケーションで標準的な選択肢となっています。
JSONを使用する利点は以下の通りです：
1. **可読性**: JSONはテキストベースであり、人が読むにも機械が解析するにも容易です。
2. **構造化**: JSONは階層的なデータ構造を持つことができ、複雑なデータも整理して保存することができます。
3. **互換性**: ほとんどのプログラミング言語にはJSONデータを解析・生成するためのライブラリがあります。
4. **軽量**: JSONはコンパクトな形式でデータを表現できるため、ネットワークを介したデータの送受信に適しています。
対話履歴を保存する場合、以下のようにJSON形式で保存することができます：
```python
import json
# 対話履歴をリストの辞書としてシミュレート
conversation_history = [
{'role': 'user', 'content': 'Hello, how are you?'},
{'role': 'assistant', 'content': 'I am fine, thank you! How can I help you?'}
# ... その他の対話エントリ ...
]
# 対話履歴をJSONファイルに保存
with open('conversation_history.json', 'w', encoding='utf-8') as f:
json.dump(conversation_history, f, ensure_ascii=False, indent=4)
# JSONファイルから対話履歴を読み込み
with open('conversation_history.json', 'r', encoding='utf-8') as f:
loaded_conversation_history = json.load(f)
print(loaded_conversation_history)
```
このコードでは、対話履歴が辞書のリストとして保存され、`json.dump()` 関数を使用してJSONファイルに書き込みます。`ensure_ascii=False` は非ASCII文字をエスケープしないようにし、`indent=4` は読みやすい形式での整形を意味します。その後、`json.load()` 関数を使用してファイルからデータを読み込み、元のリストの辞書構造として再構築します。
どうやら、こうやら試行錯誤してできました。Python入門者ですので、おかしなところがあるかもしれませんのでご注意ください。
import time
import datetime
import json
from llama_cpp import Llama
# llamaenv で実行
################ ここから必要に応じて設定 ######################
model_path="/Users/liberoscholar/ai/projects/text-generation-webui/models/stablebeluga2-70B.Q4_K_M.gguf"
model_name="stablebeluga2-70B.Q4_K_M.gguf"
# 参考にしたの https://github.com/krenzaslv/llm-terminal/blob/main/llmterminal/main.py
# LLMのパラメータの設定
ai_name = "StableBeluga2" # 好きな表示の名前を設定
chat_format = "chatml" # chatmlでうまくいったけど本当は違うみたい。モデルに合わせる必要あり
temp = 0.6
top_k = 0
top_p = 0.95
repeat_penalty = 1.1
max_tokens = 750
stop=["</s>"]
llm = Llama(model_path=model_path, chat_format=chat_format, n_ctx=0, n_gpu_layers= -1, verbose = False)
# 最初のシステムメッセージを設定
system_message = [
{
"role": "system","content": "You are Stable Beluga, an AI that follows instructions extremely well. Help as much as you can. Remember, be safe, and don't do anything illegal.",
},
]
# 対話の発言回数を設定 とりあえず10回(5回の対話やりとり)の発言のみを保持
number_memory = 10
#################### ここまでの上を設定 ################
# 最初のユーザー入力を保持するためのフラグと変数
is_first_input = True
first_user_input = ""
last_input = ""
# 会話の履歴を保持するリスト （ユーザーとアシスタントの会話）
conversation_history = []
conversation_all = []
# max_tokensを設定する関数
def set_max_tokens():
global max_tokens
while True:
try:
max_tokens_input = input("Enter max tokens (100-1000): ")
max_tokens = int(max_tokens_input)
if 100 <= max_tokens <= 1000:
print(f"Max tokens set to {max_tokens}.")
return
else:
print("Invalid input. Please enter a number between 100 and 1000.")
except ValueError:
print("Invalid input. Please enter a valid integer.") # ここでループの先頭に戻り、再入力を促す
# 対話記録のための関数
def record_conversation_history(user_prompt, assistant_response):
global conversation_history, conversation_all # 変更を加えるため宣言が必要
conversation_all.append({"role": "user", "content": user_prompt})
conversation_all.append({"role": "assistant", "content": assistant_response})
# conversation_historyが最新の10個の要素のみを保持するようにする
conversation_history = conversation_all[-number_memory:]
# 参考 切り詰めてメモリー節約方法は、messages.append({"role": "assistant", "content": "".join(responses)})
def save_conversation(conversation_all):
"""会話履歴をファイルに保存する関数"""
# 現在の日付と時刻を取得し、ファイル名にするための文字列を作成
current_time = datetime.datetime.now()
timestamp_str = current_time.strftime("%Y%m%d_%H%M")
filename = f"conversation_all_{timestamp_str}.txt"
save_content = system_message + conversation_all
if save_content[1] == save_content[2]:
del save_content[2] # 同一であれば、3番目の要素を削除 initial user contentの重複を避ける
try: # JSON形式でファイルにデータを書き込む
with open(filename, 'w', encoding='utf-8') as file:
json.dump(save_content, file, ensure_ascii=False, indent=4)
print(f"=== Conversation history saved as {filename}! ===")
except IOError as e:
print(f"Error while saving conversation history: {e}")
def load_conversation():
"""ファイルから会話履歴を読み込む関数"""
# ファイル名の入力をユーザーに求める
filename = input("Please enter the filename to load: ")
conversation_all = []
try:
with open(filename, 'r', encoding='utf-8') as file:
# ファイルの各行をリストの要素として読み込む
conversation_all = json.load(file)
print(f"=== Successfully loaded conversation history from {filename} ===")
except IOError as e:
print(f"Error while loading conversation history: {e}")
return conversation_all
# テキスト生成のための関数
def produce_text(the_prompt):
response_stream = llm.create_chat_completion(messages=the_prompt, temperature=temp, max_tokens=max_tokens, top_k=top_k, top_p=top_p, repeat_penalty=repeat_penalty, stop=stop, stream=True)
responses = []
for r in response_stream:
if "content" not in r["choices"][0]["delta"]:
continue
token = r["choices"][0]["delta"]["content"].replace("<0x0A>", "\n")
token = token.replace("<br>", "\n")
yield token
# ユーザーの入力と翻訳フラグを受け取る。 produce_textに渡すfull_promptを作成、生成したテキストを表示する
def show_chat(user_input):
global conversation_history, conversation_all, initial_prompt, is_first_input, first_user_input, last_input, system_message
# 上は、これによってグローバル変数として扱う
full_prompt = []
full_response = ""
# 最初のユーザー入力を確認し、保持するとともにsystem promptを含むinitial_promptを設定
if is_first_input:
if user_input in ('h', 'c', 'r', '/show', '/clear', '/history', '/save', '/reset', '/tokens'):
print('No initial prompt, yet.')
return
else:
first_user_input = user_input # showのため グローバル変数として保存
first_message = {"role": "user", "content": first_user_input}
system_message.append(first_message)
is_first_input = False # 最初の入力が処理されたのでフラグを更新
# プロントに用いる会話履歴をクリアーにするコマンドの実行部分
if user_input == "/clear":
conversation_history = []
print("===! Conversation history cleared! ===")
return
# ユーザーが会話履歴を表示するコマンドを入力した場合
if user_input == '/history':
print("\n===== Recent Conversation History =====\n")
for message in conversation_history:
print(f"{message['role']}: {message['content']}")
return # 会話履歴を表示し、次の入力を待つ
# initial promptからすべての会話履歴をクリアーにするコマンドの実行部分
if user_input == "/reset":
conversation_all = []
conversation_history = []
first_user_input =""
is_first_input = True
print("===! ALL Conversation history and Initial Prompt cleared! ===")
return
if user_input == "/tokens":
set_max_tokens()
return
# 会話履歴を保存する
if user_input == "/save":
# save関数を呼び出し
save_conversation(conversation_all)
return
if user_input == "/load":
# 関数を呼び出して、結果を確認します。
conversation_all = load_conversation()
print(conversation_all) # 読み込んだ内容を出力
system_message = [conversation_all[0], conversation_all[1]]
conversation_all = conversation_all[2:]
conversation_history = conversation_all[-number_memory:]
is_first_input = False
return
# システムプロンプトとイニシャルプロンプトを表示する実行
if user_input == "/show":
print("=== System Prompt and Initial Prompt ===")
print(system_message)
print("")
return
if user_input == "h":
print_commands()
return
# 続きを促すショートカットがあったら、続きを促す文章にする
if user_input == 'c':
user_input = 'Continue from your last line.'
# regemerateを表示する実行
if user_input == "r":
print("=== Regeneration ===")
user_input = last_input
conversation_history = conversation_history[:-2] # スライス操作でリストの最後の2要素(対話)を取り除く
conversation_all = conversation_all[:-2]
# 連結して full_promptを設定
new_user_message = {"role": "user", "content": user_input}
full_prompt = system_message + conversation_history + [new_user_message]
# full_prompt の2番目と3番目の要素が同一かチェック indexは0から
if full_prompt[1] == full_prompt[2]:
del full_prompt[2] # 同一であれば、3番目の要素を削除
#print(full_prompt) 確認用のprint文
print(f"\n{ai_name}: ", end="", flush=True)
for chunk in produce_text(full_prompt): #produce_text関数を呼び出す
full_response += chunk # 生成されたテキスト全文を収納して会話記録更新に使う
print(chunk, end="", flush=True) # chunkテキストを出力
print("\n")
# conversation_allとconversation_historyを更新する関数の実行
record_conversation_history(user_input, full_response)
last_input = user_input
# ユーザーからの入力を処理し、整形されたテキストを返す関数
def get_user_input():
user_input = ""
multi_line = False
while True:
line = input("User: ")
if line.startswith('"""'):
if multi_line:
multi_line = False
if line.endswith('"""'):
user_input += line[:-3] # 末尾の引用符を除去する
else:
user_input += line + "\n"
break
else:
multi_line = True
if line.endswith('"""') and len(line) > 3:
user_input += line[3:-3] + "\n" # 先頭と末尾の引用符を除去する
break
else:
user_input += line[3:] + "\n" # 先頭の引用符を除去する
else:
if multi_line:
if line.endswith('"""'):
user_input += line[:-3] # 末尾の引用符を除去する
break
else:
user_input += line + "\n"
else:
user_input += line
break
return user_input.strip()
def print_commands():
print("\n⭐️⭐️⭐️ Llama.cpp Language Model Interactive Terminal ⭐️⭐️⭐️\n")
print("Model: ", model_name)
print("-" * 70)
print("Available Commands:")
print(
" type `h`: Help for display available commands.\n"
" type `c`: Continue from the last response.\n"
" type `r`: Regenerate another response.\n"
" type `q`: Quit the application.\n"
" type `/history`: View recent conversation history.\n"
" type `/save`: Save all conversation history to a file.\n"
" type `/load`: Load the saved conversation history file.\n"
" type `/reset`: Restart with a new initial prompt.\n"
" type `/show`: Display system and initial prompt.\n"
" type `/tokens`: Set the output's max token limit.\n"
)
print("-" * 70)
print('For multi-line input, enclose your text with triple quotes (""") ')
print('at the beginning and the end.')
print("=" * 70 + "\n")
# メイン部分
def main():
print_commands()
while True:
formatted_input = get_user_input()
# ユーザーが終了コマンドを入力した場合
if formatted_input.lower() == 'q':
print("\n" + "=" * 70 + "\n")
print("Exiting the interactive terminal.")
print("\n")
break # スクリプトが終了する
# ユーザー入力が何もなかった場合に警告を出して次の入力を待つ
if not formatted_input:
print("Please enter some text to proceed.")
continue # Whileのループの頭に戻る
show_chat(formatted_input)
# このスクリプトを走らせるためのコマンド
if __name__ == "__main__":
main()Llama-cpp-pythonを使った、 StableBeluga2 用 Terminalができた〜 やっぱり日本語がちゃんとしてるよ♪ pic.twitter.com/7LZ9MoPSui
— Lucas (@LucasChatGPT) February 11, 2024
覚え書き
StableBeluga2のtemplateはこちら
### System:
This is a system prompt, please behave and help the user.
### User:
Your prompt here
### Assistant:
The output of Stable Beluga 2
chatmlのtemplateのtemplateはこっち。 動いたからヨシとしてます。
<|im_start|>system
{system_message}<|im_end|>
<|im_start|>user
{prompt}<|im_end|>
<|im_start|>assistant
#AI #AIとやってみた #やってみた #ローカルLLM #大規模言語モデル #Python入門 #MacbookPro
いいなと思ったら応援しよう！
