2025年4月17日木曜日

dspy.LM で会話形式で質問する方法

dspy.LM で会話形式で質問する方法

概要

前の質問と回答を次回の質問時にも渡すことで前の会話もコンテキストに含めることができます

環境

  • Ubuntu 24.04
  • Python 3.12.9
  • dspy 2.6.17

サンプルコード

  • vim app.py
import json
import os
import sys

import dspy

# モデル設定
lm = dspy.LM(
    "azure/gpt-4-32k",
    api_key="xxx",
    api_version="",
    api_base="https://your-llm-endpont/chat-ai/gpt4",
)
dspy.configure(lm=lm)

# チャット履歴ファイルパス
HISTORY_FILE = "chat_history.json"

class ProjectChat(dspy.Module):
    def __init__(self):
        super().__init__()
        self.chat_history = self.load_history()

    def load_history(self):
        if os.path.exists(HISTORY_FILE):
            with open(HISTORY_FILE, "r", encoding="utf-8") as f:
                return json.load(f)
        # 初期 system メッセージ（なければ context なしでもOK）
        return [{"role": "system", "content": "あなたは優秀なソフトウェアエンジニアとして、コードに関する質問に日本語で答えてください。"}]

    def save_history(self):
        with open(HISTORY_FILE, "w", encoding="utf-8") as f:
            json.dump(self.chat_history, f, ensure_ascii=False, indent=2)

    def ask(self, user_question):
        self.chat_history.append({"role": "user", "content": user_question})
        response = dspy.settings.lm(messages=self.chat_history)
        # LLM からのレスポンスが配列なので文字列に変換して保存する
        self.chat_history.append({"role": "assistant", "content": ",".join(response)})
        self.save_history()
        return response


if __name__ == "__main__":
    chat = ProjectChat()

    print("会話を開始します（Ctrl+Dで入力完了 / Ctrl+C or 'exit' のみの入力で終了）\n")

    while True:
        print("質問をどうぞ（複数行ペースト対応）：")
        user_input = ""
        try:
            user_input = sys.stdin.read()
        except KeyboardInterrupt:
            print("終了します")
            break
        if user_input.strip().lower() in {"exit", "quit"}:
            print("終了します")
            break
        print("\n入力が終了しました")
        if not user_input.strip():
            print("質問が入力されていません。再度入力してください。")
            continue

        res = chat.ask(user_input.strip())
        print(f"\n回答: {res}\n")

動作確認

  • pipenv run python app.py

で会話を開始します

会話を開始します（Ctrl+Dで入力完了 / Ctrl+C or 'exit' のみの入力で終了）

質問をどうぞ（複数行ペースト対応）：
以下のコードについて説明してください
import time


class Message:
    def say(self, **kwargs):
        print("Start say")
        print(kwargs)
        time.sleep(10)
        raise ValueError()
        # print("End say")

入力が終了しました
[{'role': 'system', 'content': 'あなたは優秀なソフトウェアエンジニアとして、コードに関する質問に日本語で答えてください。'}, {'role': 'user', 'content': '以下のコードについて説明してください\nimport time\n\n\nclass Message:\n    def say(self, **kwargs):\n        print("Start say")\n        print(kwargs)\n        time.sleep(10)\n        raise ValueError()\n        # print("End say")'}]

回答: ['このコードは、Pythonのクラスと例外処理に関する基本的な例です。以下にコードの各部分について説明します。\n\n1. **`import time`**:\n   - Pythonの標準ライブラリである`time`モジュールをインポートしています。このモジュールは、時間に関連する機能を提供します。\n\n2. **`class Message:`**:\n   - `Message`という名前のクラスを定義しています。クラスはオブジェクト指向プログラミングの基本単位で、データとメソッドを持つことができます。\n\n3. **`def say(self, **kwargs):`**:\n   - `say`という名前のメソッドを定義しています。このメソッドは、キーワード引数を受け取ることができます。`**kwargs`は、任意の数のキーワード引数を辞書として受け取ることを意味します。\n\n4. **`print("Start say")`**:\n   - メソッドの実行が開始されたことを示すメッセージをコンソールに出力します。\n\n5. **`print(kwargs)`**:\n   - 受け取ったキーワード引数をコンソールに出力します。`kwargs`は辞書として渡されるので、その内容が表示されます。\n\n6. **`time.sleep(10)`**:\n   - プログラムの実行を10秒間停止します。これは、`time`モジュールの`sleep`関数を使用して行われます。\n\n7. **`raise ValueError()`**:\n   - `ValueError`という例外を発生させます。これは、プログラムの実行を中断し、エラーメッセージを表示するために使用されます。\n\n8. **`# print("End say")`**:\n   - この行はコメントアウトされています。もしコメントアウトが解除されていた場合、メソッドの終了を示すメッセージをコンソールに出力することになります。しかし、現在のコードではこの行は実行されません。\n\nこのコードを実行すると、`say`メソッドが呼び出された際に、開始メッセージとキーワード引数が表示され、10秒間の待機の後に`ValueError`が発生します。例外が発生するため、コメントアウトされた行は実行されません。']

一度プロセスを終了して再度実行します
その場合でも前の会話をちゃんと LLM 側が把握していることが確認できます

質問をどうぞ（複数行ペースト対応）：
time.sleep(10)をランダムな時間sleepするように修正してください

入力が終了しました
[{'role': 'system', 'content': 'あなたは優秀なソフトウェアエンジニアとして、コードに関する質問に日本語で答えてください。'}, {'role': 'user', 'content': '以下のコードについて説明してください\nimport time\n\n\nclass Message:\n    def say(self, **kwargs):\n        print("Start say")\n        print(kwargs)\n        time.sleep(10)\n        raise ValueError()\n        # print("End say")'}, {'role': 'assistant', 'content': 'このコードは、Pythonのクラスと例外処理に関する基本的な例です。以下にコードの各部分について説明します。\n\n1. **`import time`**:\n   - Pythonの標準ライブラリである`time`モジュールをインポートしています。このモジュールは、時間に関連する機能を提供します。\n\n2. **`class Message:`**:\n   - `Message`という名前のクラスを定義しています。クラスはオブジェクト指向プログラミングの基本単位で、データとメソッドを持つことができます。\n\n3. **`def say(self, **kwargs):`**:\n   - `say`という名前のメソッドを定義しています。このメソッドは、キーワード引数を受け取ることができます。`**kwargs`は、任意の数のキーワード引数を辞書として受け取ることを意味します。\n\n4. **`print("Start say")`**:\n   - メソッドの実行が開始されたことを示すメッセージをコンソールに出力します。\n\n5. **`print(kwargs)`**:\n   - 受け取ったキーワード引数をコンソールに出力します。`kwargs`は辞書として渡されるので、その内容が表示されます。\n\n6. **`time.sleep(10)`**:\n   - プログラムの実行を10秒間停止します。これは、`time`モジュールの`sleep`関数を使用して行われます。\n\n7. **`raise ValueError()`**:\n   - `ValueError`という例外を発生させます。これは、プログラムの実行を中断し、エラーメッセージを表示するために使用されます。\n\n8. **`# print("End say")`**:\n   - この行はコメントアウトされています。もしコメントアウトが解除されていた場合、メソッドの終了を示すメッセージをコンソールに出力することになります。しかし、現在のコードではこの行は実行されません。\n\nこのコードを実行すると、`say`メソッドが呼び出された際に、開始メッセージとキーワード引数が表示され、10秒間の待機の後に`ValueError`が発生します。例外が発生するため、コメントアウトされた行は実行されません。'}, {'role': 'user', 'content': 'time.sleep(10)をランダムな時間sleepするように修正してください'}]

回答: ['`time.sleep(10)`をランダムな時間だけスリープするように修正するには、Pythonの`random`モジュールを使用します。このモジュールを使うことで、指定した範囲内でランダムな数値を生成することができます。以下に修正したコードを示します。\n\n python\nimport time\nimport random\n\nclass Message:\n    def say(self, **kwargs):\n        print("Start say")\n        print(kwargs)\n        \n        # 1秒から10秒の間でランダムな時間だけスリープする\n        sleep_time = random.uniform(1, 10)\n        print(f"Sleeping for {sleep_time:.2f} seconds")\n        time.sleep(sleep_time)\n        \n        raise ValueError()\n        # print("End say")\n \n\n### 修正点の説明\n\n1. **`import random`**:\n   - `random`モジュールをインポートしました。このモジュールは、ランダムな数値を生成するための機能を提供します。\n\n2. **`sleep_time = random.uniform(1, 10)`**:\n   - `random.uniform(a, b)`関数を使用して、1秒から10秒の間でランダムな浮動小数点数を生成しています。この数値がスリープする時間になります。\n\n3. **`print(f"Sleeping for {sleep_time:.2f} seconds")`**:\n   - スリープする時間をコン
ソールに出力しています。`:.2f`は、浮動小数点数を小数点以下2桁まで表示するためのフォーマット指定子です。\n\nこの修正により、`say`メソッドは1秒から10秒の間でランダムな時間だけスリープするようになります。']

質問をどうぞ（複数行ペースト対応）：

トラブルシューティング

litellm.exceptions.APIError: litellm.APIError: AzureException APIError - Error code: 500 - {'fault': {'faultstring': 'Execution of ServiceCallout SC-Count-Token failed. Reason: ResponseCode 500 is treated as error', 'detail': {'errorcode': 'steps.servicecallout.ExecutionFailed'}}}

LLM によっては質問時にフォーマットが決まっていることがあるようです
自分の場合は履歴を渡すときに content が配列だと上記エラーに遭遇しました
なので会話の履歴を保存する際に配列ではなく文字列で保存するようにしています

最後に

この方法で一番気をつけなければいけないことは会話が長くなることでトークン制限引っかかることです
トークン制限に引っかかった場合には会話の履歴ファイルを直接編集し前の会話を削除するなりしてください
もちろん Module 内で会話のファイルが一定サイズより大きくなったら削除するなどしても OK です

この方法は chatgpt の WebUI などでも行っています
ただ完全に同じではなく chatgpt WebUI などでは古い会話の自動削除や前の会話をそのまま渡さす要約化したりしています
このあたりもまだまだプロンプト最適化できるかなと思います

会話方式でコンテキストを渡す方法とコード全体の要約を毎回質問時に渡す方法はどちらが効率的かつ精度がいいのか気になります

時刻:
ラベル: , ,

0 件のコメント:

コメントを投稿

登録: コメントの投稿 (Atom)