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で動かせるようにすることが今回の目標です。

さて先の引用スクリプトのことについて質問しました。

GPT-4の教えは下記のとおり。

わかりやすい説明です。この部分を活用して MLX 用のterminalスクリプトを改造しました。

さらに、ファイルの読み出しもつけてみようと思います。

LLMではこの形式が一般的なのかを尋ねました。

どうやら、こうやら試行錯誤してできました。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()

覚え書き
　StableBeluga2のtemplateはこちら

chatmlのtemplateのtemplateはこっち。　動いたからヨシとしてます。

#AI #AIとやってみた #やってみた #ローカルLLM #大規模言語モデル #Python入門 #MacbookPro