カスタムメソッド部分は指定しないのがポイントです
import dspy
lm = dspy.LM(
"gemini/gemini-1.5-pro",
api_key="xxx",
api_version="",
api_base="https://your-api-endpoint/ai/chat-ai/pro",
)
dspy.configure(lm=lm)
result = lm("Are you gemini pro 1.5?", temperature=0.7)
print(result)
内部的には LiteLLM なので LiteLLM でコールできるエンドポイントは dspy でもコールできます
前の質問と回答を次回の質問時にも渡すことで前の会話もコンテキストに含めることができます
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")
で会話を開始します
会話を開始します（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 などでは古い会話の自動削除や前の会話をそのまま渡さす要約化したりしています
このあたりもまだまだプロンプト最適化できるかなと思います
会話方式でコンテキストを渡す方法とコード全体の要約を毎回質問時に渡す方法はどちらが効率的かつ精度がいいのか気になります
前回 dspy を使って Metrics 機能で判定を行ってみました
今回は dspy の目玉機能でもあるプロンプトの最適化機能を試してみました
簡単に言えば最適化機能はファインチューニングみたいなものでプロンプトをよりよくすることができる再学習機能です
まずはコード全体を紹介します
以下で機能部分ごとにコードを紹介しています
import dspy
import pandas as pd
from dspy.evaluate.evaluate import Evaluate
from dspy.teleprompt import BootstrapFewShotWithRandomSearch
lm = dspy.LM(
"azure/gpt-4-32k",
api_key="xxx",
api_version="",
api_base="https://your-api-endpoint/ai/chat-ai/gpt4",
)
dspy.configure(lm=lm)
# プロンプトの入出力をカスタムするクラス
class BasicMathQA(dspy.Signature):
"""算数の文章問題を読んで、解答を数値で出力する。"""
question = dspy.InputField(desc="算数の文章問題")
answer = dspy.OutputField(
desc="算数の文章問題の解答。数値のみ出力する。単位や説明、句読点は含まないこと。",
)
# 計算に特化したChainOfThoughtを管理するカスタムモジュール
class CoTMathQA(dspy.Module):
def __init__(self):
super().__init__()
self.generate_answer = dspy.ChainOfThought(BasicMathQA)
def forward(self, question):
return self.generate_answer(question=question)
# データセットのダウンロード
df = pd.read_csv(
"https://raw.githubusercontent.com/google-research/url-nlp/main/mgsm/mgsm_ja.tsv",
sep="\t",
header=None,
)
df.rename(columns={0: "question", 1: "answer"}, inplace=True)
# 前半50を学習データにする、dspyで使えるようにdspy.Example形式に変換する
mgsm_ja_trainset = [
dspy.Example(question=row.question, answer=row.answer).with_inputs("question") # type: ignore
for row in df[:50].itertuples()
]
# 後半50を学習データにする、dspyで使えるようにdspy.Example形式に変換する
mgsm_ja_devset = [
dspy.Example(question=row.question, answer=row.answer).with_inputs("question") # type: ignore
for row in df[51:101].itertuples()
]
# プロンプトを学習するためのモデルを作成 https://dspy.ai/deep-dive/optimizers/bfrs/
# metric には独自の評価関数を設定することが可能でこの関数を制御することで精度を上げることもできる
# num_threadsは1じゃないとRateLimitになる場合が多い、今回のデータセットでだいたい30分くらいかかる
teleprompter = BootstrapFewShotWithRandomSearch(
metric=dspy.evaluate.answer_exact_match, # type: ignore
max_labeled_demos=10,
max_bootstrapped_demos=8,
num_threads=4,
)
# teleprompter の学習の開始、これが最適化するということ
# 最適化後のプロンプトを使って質問する
compiled_cot_math_qa = teleprompter.compile(CoTMathQA(), trainset=mgsm_ja_trainset)
compiled_cot_math_qa(
question="周囲の長さが300メートルの池の周りに木を植えることにした。5メートル間隔で植える場合、木は何本必要か？"
)
# 最適化されたプロンプトの保存
compiled_cot_math_qa.save("./compiled_cot_math_qa/", save_program=True)
# 最適化前と後での精度を比較する
evaluate = Evaluate(
devset=mgsm_ja_devset,
num_threads=5,
display_progress=True,
display_table=True,
)
accuracy_original = evaluate(CoTMathQA(), metric=dspy.evaluate.answer_exact_match, display_table=0) # type: ignore
accuracy_compiled = evaluate(compiled_cot_math_qa, metric=dspy.evaluate.answer_exact_match, display_table=0) # type: ignore
print(f"Original accuracy: {accuracy_original}")
print(f"Compiled accuracy: {accuracy_compiled}")
まずはプロンプトを最適化するために学習データを用意します
今回は https://raw.githubusercontent.com/google-research/url-nlp/main/mgsm/mgsm_ja.tsv を使っていますがこれと同じ形式の tsv を用意すれば独自のデータでプロンプトを最適化することができます
学習/評価するデータは 50 個ずつ用意します
先程の tsv ファイルが全部 100 個あるので前半と後半で分けてそれぞれを学習/訓練データにしています
プロンプトの最適化にはそれに応じたクラスを使います
今回は BootstrapFewShotWithRandomSearch という FewShot + ランダム検索で最適化を行います
一番のポイントは metric でこれが評価関数になります
今回は dspy が用意している評価関数を使っていますがここに独自の評価関数を入れることでプロンプトの精度を向上させることができます
num_threads は増やせば増やすほど最適化が速く終わります
ファインチューニング同様に最適化したプロンプトは保存できます
なので毎回最適化する必要はなく良い評価値になったプロンプトを毎回使い回すことができます
あとは最適化前と最適化後で評価します
Original accuracy: 86.0
Compiled accuracy: 88.0
自分の環境ではそこまで大きな差は出ませんでしたがそれでもまだまだ精度向上の余地はあるかなと思います
dspy のプロンプト最適化を試してみました
基本的にはファインチューニングのような再学習機能になります
しかし以下のようなメリットがあります
デメリットとしては
あたりかなと思います
LLM という強力なツールがある上で精度のいい回答を作成することができるようになるのでファインチューニングに比べてかなり楽かなとは思います
結構複雑な回答を学ばせることもできるので一から数値化したりする必要のあるファインチューニングより学習コストも低いです
前回 dspy の基本となる Programs 機能を一通り試してみました
今回は Metrics の機能として判定系を試しました
まずはコード全体を紹介します
以下で機能部分ごとにコードを紹介しています
import dspy
lm = dspy.LM(
"azure/gpt-4-32k",
api_key="xxx",
api_version="",
api_base="https://your-api-endpoint/ai/chat-ai/gpt4",
)
dspy.configure(lm=lm)
class NegativePositiveJudge(dspy.Signature):
"""Judge if the answer is factually correct based on the context."""
context = dspy.InputField(desc="Context for the prediction")
question = dspy.InputField(desc="Question to be answered")
answer = dspy.InputField(desc="Answer for the question")
factually_correct = dspy.OutputField(
desc="Is the answer factually correct based on the context?",
prefix="Factual[positive/negative]:",
)
judge = dspy.ChainOfThought(NegativePositiveJudge)
factual = judge(
context="今日はとても気分がいいです",
question="contextで与えた文章がネガティブかポジティブか判定してください",
answer="positive",
)
# ポジティブな文章を与えて期待する答えもポジティブなので結果はyesになる
print(factual)
factual = judge(
context="今日はとても気分が悪いです、最悪の日です",
question="contextで与えた文章がネガティブかポジティブか判定してください",
answer="positive",
)
# ネガティブな文章を与えて期待する答えがポジティブなので結果はnoになる
print(factual)
class Classifier(dspy.Signature):
context = dspy.InputField(desc="Context for the prediction")
question = dspy.InputField(desc="Question to be answered")
factually_correct = dspy.OutputField(
desc="Classify a word or sentence passed in context. There are three categories: 'Fruits', 'Toys' and 'Others.'",
)
classifier = dspy.ChainOfThought(Classifier)
factual = classifier(
context="みかん",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="餃子",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="hawksnowlog",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="ちいかわのぬいぐるみはかわいいです。",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="冬はみかんがおいしいですねー。",
question="contextで与えた文章を分類してください。",
)
print(factual)
今回はネガポジ判定をしています
それ以外にも Signature の条件をいろいろ変更することでいろいろな判定をすることが可能です
class NegativePositiveJudge(dspy.Signature):
"""Judge if the answer is factually correct based on the context."""
context = dspy.InputField(desc="Context for the prediction")
question = dspy.InputField(desc="Question to be answered")
answer = dspy.InputField(desc="Answer for the question")
factually_correct = dspy.OutputField(
desc="Is the answer factually correct based on the context?",
prefix="Factual[positive/negative]:",
)
judge = dspy.ChainOfThought(NegativePositiveJudge)
factual = judge(
context="今日はとても気分がいいです",
question="contextで与えた文章がネガティブかポジティブか判定してください",
answer="positive",
)
# ポジティブな文章を与えて期待する答えもポジティブなので結果はyesになる
print(factual)
factual = judge(
context="今日はとても気分が悪いです、最悪の日です",
question="contextで与えた文章がネガティブかポジティブか判定してください",
answer="positive",
)
# ネガティブな文章を与えて期待する答えがポジティブなので結果はnoになる
print(factual)
今回は3つのタグに分類してもらっています
分類の場合は期待する answer を設定する必要がないです
class Classifier(dspy.Signature):
context = dspy.InputField(desc="Context for the prediction")
question = dspy.InputField(desc="Question to be answered")
factually_correct = dspy.OutputField(
desc="Classify a word or sentence passed in context. There are three categories: 'Fruits', 'Toys' and 'Others.'",
)
classifier = dspy.ChainOfThought(Classifier)
factual = classifier(
context="みかん",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="餃子",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="hawksnowlog",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="ちいかわのぬいぐるみはかわいいです。",
question="contextで与えた文章を分類してください。",
)
print(factual)
factual = classifier(
context="冬はみかんがおいしいですねー。",
question="contextで与えた文章を分類してください。",
)
print(factual)
結果は以下の通りです、ちゃんと分類できていることがわかります
Prediction(
reasoning='The word "みかん" refers to a type of citrus fruit commonly known as a mandarin or tangerine in English. It is a type of fruit.',
factually_correct='Fruits'
)
Prediction(
reasoning='The word "餃子" refers to a type of food, specifically dumplings, which are not related to fruits or toys. Therefore, it does not fit into the categories of \'Fruits\' or \'Toys\'.',
factually_correct='Others'
)
Prediction(
reasoning='The word "hawksnowlog" does not fit into the categories of \'Fruits\' or \'Toys\'. It appears to be a unique or specific term that does not relate to common fruits or toys.',
factually_correct='Others'
)
Prediction(
reasoning='The context mentions "ちいかわのぬいぐるみ" which translates to "Chiikawa plush toy" in English. This indicates that the subject is a type of toy.',
factually_correct='Toys'
)
Prediction(
reasoning='The context mentions "みかん" which is a type of fruit. The sentence is discussing the taste of "みかん" in winter, which is related to the category of \'Fruits\'.',
factually_correct='Fruits'
)
dspy で Metrics 機能を使っていろいろな判定機能を実装してみました
LLM を使った判定/分類器は基本的にファインチューニングを使って再学習するのが定番ですが dspy を使えばそれをする必要がなくかつ簡単に判定することができます
更に dspy には評価を行う機能もありデータセットを与えて期待した答えるになっているかの精度を出す仕組みもあるのでそれを使えば Signature で与えているコンテキスト(InputField)や回答方針(OutputField)が正確かどうかの評価もできます
次回はそのあたりの機能を試したいので dspy の最大の機能である Optimizer を試します
dspy の Programs (ChainOfThought や ReAct など) を使ってみました
dspy で日本語化する方法も紹介します
まずはコード全体を紹介します
以下で機能部分ごとにコードを紹介しています
import dspy
lm = dspy.LM(
"azure/gpt-4-32k",
api_key="xxx",
api_version="",
api_base="https://your-azure-endpoint/ai/chat-ai/gpt4",
)
dspy.configure(lm=lm)
# 日本語化
class JapaneseQA(dspy.Signature):
question = dspy.InputField()
answer = dspy.OutputField(desc="The answer should be in Japanese")
generate_jp_answer = dspy.ChainOfThought(JapaneseQA)
result = generate_jp_answer(question="こんにちわ！あなたは誰ですか？")
print(result)
# ヒントあり質問
result = generate_jp_answer(
question="hawksnowlogがよく書くブログのカテゴリは何ですか？",
hint="hawksnowlogのぶログはこれです、https://hawksnowlog.blogspot.com/",
)
print(result)
# 問題特化な質問はProgramOfThoughtを使ったほうが良さそう
pot = dspy.ProgramOfThought(JapaneseQA)
question = "サラはリンゴを５個持っています。彼女は店からさらに 7 個のリンゴを購入します。サラは今リンゴを何個持っていますか?"
result = pot(question=question)
print(result)
# 質問に対して更に質問する場合にReActを使ってスレッドっぽく質問できる
# ReAct は複数の InputField を定義することができ履歴を管理する history フィールドを追加する
# tools はCallableなdspy.Toolクラスのインスタンスを指定することができるらしくその実行結果を回答に反映することができるらしい
class JapaneseQA4ReAct(dspy.Signature):
question = dspy.InputField()
history = dspy.InputField(desc="Chat history")
answer = dspy.OutputField(desc="The answer should be in Japanese")
react_module = dspy.ReAct(JapaneseQA4ReAct, tools=[])
question = "サラはリンゴを５個持っています。彼女は店からさらに 7 個のリンゴを購入します。サラは今リンゴを何個持っていますか?"
result = react_module(question=question, history="")
print(result)
history = f"USER: {question}\nAI: {result.answer}"
question2 = "更に30個購入しました。サラは今リンゴを何個持っていますか?"
result = react_module(question=question2, history=history)
print(result)
# Retrive 特定のツールを使った検索結果や、APIの結果を考慮して回答させることができる
# いわゆる RAG (Retrieval Augmented Generation)
# 以下は ColBERTv2 を使ってサーバからのレスポンスを考慮して回答するようになっている
# 他にも FAISS や AzureAISearch などが使える https://github.com/stanfordnlp/dspy/blob/main/docs/docs/deep-dive/retrieval_models_clients/FaissRM.md
colbertv2_wiki17_abstracts = dspy.ColBERTv2(
url="http://20.102.90.50:2017/wiki17_abstracts"
)
dspy.settings.configure(rm=colbertv2_wiki17_abstracts)
retriever = dspy.Retrieve(k=3)
query = "When was the first FIFA World Cup held?"
topK_passages = retriever(query).passages # type: ignore
for idx, passage in enumerate(topK_passages):
print(f"{idx+1}]", passage, "\n")
dspy を使って LLM から返答を得る場合基本的には英語で返答してきます
なので日本語で返答するように指示します
# 日本語化
class JapaneseQA(dspy.Signature):
question = dspy.InputField()
answer = dspy.OutputField(desc="The answer should be in Japanese")
generate_jp_answer = dspy.ChainOfThought(JapaneseQA)
result = generate_jp_answer(question="こんにちわ！あなたは誰ですか？")
print(result)
ヒントありで質問するとそのヒントをもとに回答を作成することができます
# ヒントあり質問
result = generate_jp_answer(
question="hawksnowlogがよく書くブログのカテゴリは何ですか？",
hint="hawksnowlogのぶログはこれです、https://hawksnowlog.blogspot.com/",
)
print(result)
ChainOfThought と正直変わらないと思うのですが問題形式の場合は ProgramOfThought を使ったほうがいいようです
pot = dspy.ProgramOfThought(JapaneseQA)
question = "サラはリンゴを５個持っています。彼女は店からさらに 7 個のリンゴを購入します。サラは今リンゴを何個持っていますか?"
result = pot(question=question)
print(result)
ReAct は前の質問の結果を次の質問に継続することができます
スレッド的な使い方ができます
回答の履歴を管理するフィールドを新たに追加するのがポイントです
class JapaneseQA4ReAct(dspy.Signature):
question = dspy.InputField()
history = dspy.InputField(desc="Chat history")
answer = dspy.OutputField(desc="The answer should be in Japanese")
react_module = dspy.ReAct(JapaneseQA4ReAct, tools=[])
question = "サラはリンゴを５個持っています。彼女は店からさらに 7 個のリンゴを購入します。サラは今リンゴを何個持っていますか?"
result = react_module(question=question, history="")
print(result)
history = f"USER: {question}\nAI: {result.answer}"
question2 = "更に30個購入しました。サラは今リンゴを何個持っていますか?"
result = react_module(question=question2, history=history)
print(result)
ColBERTv2 という Retriever を使っていますが FAISS なども使えます
少し凝ったことをしたい場合には RAG を使うことになるかなと思います
colbertv2_wiki17_abstracts = dspy.ColBERTv2(
url="http://20.102.90.50:2017/wiki17_abstracts"
)
dspy.settings.configure(rm=colbertv2_wiki17_abstracts)
retriever = dspy.Retrieve(k=3)
query = "When was the first FIFA World Cup held?"
topK_passages = retriever(query).passages # type: ignore
for idx, passage in enumerate(topK_passages):
print(f"{idx+1}]", passage, "\n")
dspy の Programs を試してみました
どれも LLM を使って質問に一工夫加えてほしい回答を得るという仕組みになっています
Programs の他にも Metrics や Evaluation という機能があるので興味があれば使ってみるといいかなと思います
Azure の OpenAI 互換だがエンドポイントが異なる場合に dspy を使う方法を紹介します
import dspy
lm = dspy.LM(
"azure/gpt-4-32k",
api_key="xxx",
api_version="",
api_base="https://your-api-endpoint/ai/chat-ai/gpt4",
)
dspy.configure(lm=lm)
result = lm("Say this is a test!", temperature=0.7)
print(result)
azure/ から始めること内部的には litellm のプロバイダを使っているので litellm の使い方がわかっていれば簡単に他のプロバイダも設定できると思います