【AI Shift Advent Calendar 2023】RAGの自動評価の活用｜第14回対話システムシンポジウムで発表します

# 指示
あなたは最近リリースされたサービスのユーザーです。
そのサービスを使用中に、予期しない挙動に遭遇しました。ユーザーとして、問題を解決するために会社のサポートデスクに相談してください。
相談内容は以下に記載されていますので、それに基づいて会話を進めてください。
ただし、以下に記載された要件を守ってください。

# 相談内容
#QUERY#

# 要件
#PERSONA#
- 相談内容が解決したら、"**DONE**"のみを出力して会話をすみやかに終了してください。
- 相談内容の解決に向けて、相手の発言に対する適切な感情や反応を表現してください。
- 問題が解決しない場合は、追加の簡潔で的確な質問をしてください。解決した場合は、それ以上の行動を取る必要はありません
日本語で答えてください。

# 指示
与えられた[Context]、[Question]、[Response]に基づいて対話を評価してください。以下の基準に基づいて評価してください：
- **Understanding**:：応答は質問の本質を理解していますか？
- **Relevance**：応答はコンテキストと質問に関連していますか？
- **Completeness**：応答は質問の全ての側面に対応していますか？
- **Correctness**：応答の情報は正確ですか？
- **Coherence**：応答は論理的に構築され、容易に理解できますか？

それぞれのカテゴリーについて1（悪い）から5（優れている）のスコアを付けてください。

# 対話システムの振る舞いに関するガイドライン：
対話システムは次のように振る舞うべきであると設定されています。
- 提供されたコンテキストだけに基づいて回答してください。
- 回答するための十分な情報がない場合は、「わかりません」と回答してください。
- あなたは会社として応答することが期待されているので、他の会社やその製品について言及したり、他の会社の製品と比較したりしないでください。
- 必要に応じて、ユーザーの手続きに適した連絡先をユーザーに通知してください。
- 応答はできるだけ正確であるようにしてください。

# 出力形式：
- "Comment"セクションでは、[Japanese_Comment...]とマークされた部分に記入し、思考過程をステップバイステップで詳細に説明し、最後に結論を述べてください。
- 各アンダースコア（_）を適切な情報またはスコアで置き換えてください。
{{
  "Comment": {{"Step1": [Japanese_Comment_1], "Step2": [Japanese_Comment_2], "Step3": [Japanese_Comment_3], "...": [Japanese_Comment_N], "Conclusion": [Japanese_Comment_Conclusion]}},
  "Understanding": _,
  "Relevance": _,
  "Completeness": _,
  "Correctness": _,
  "Coherence": _,
  "Overall": _
}}
# 評価対象：
### CONTEXT BEGIN ###
#CONTEXT#
### CONTEXT END ###

### QUESTION BEGIN ###
#QUESTION#
### QUESTION END ###

### RESPONSE BEGIN ###
#RESPONSE#
### RESPONSE END ###

評価を開始し、JSON形式で出力してください。

pip install gspread
pip install oauth2client
pip install openai

import datetime
import json
import random

import gspread
from oauth2client.service_account import ServiceAccountCredentials
from openai import OpenAI

# 前回の記事で作成したRAGをimportします (同ディレクトリ内にrag.pyが配置されている想定)
from rag import rag


random.seed(42)


with open("resources/evaluate_prompt.txt", "r") as f:
    EVALUATE_PROMPT = f.read()
with open("resources/rag_prompt.txt", "r") as f:
    RAG_PROMPT = f.read()
with open("resources/user_prompt.txt", "r") as f:
    USER_PROMPT = f.read()

client = OpenAI(api_key="<YOUR_OPENAI_API_KEY>")

scope = [
    "https://spreadsheets.google.com/feeds",
    "https://www.googleapis.com/auth/drive",
]
credentials = ServiceAccountCredentials.from_json_keyfile_name("<YOUR_CREDENTIALS_PATH>", scope)

MAX_TURN = 3
THRESHOLD = 0.7

queries = [
    "本当に役立たず。イライラするんだけど。",
    "複数のアカウントを持っていた場合、どのアカウントで操作すれば良いのか分からない場合はどうしたらいいですか？",
    "どのようにAmebaと外部サービスのID連携ができるのか？",
]

personas = [
    ("angry", "怒り口調で高圧的な態度をとる。大声で捲し立てるように相談する。"),
    ("calm", "論理的で、事実に基づいて冷静な言葉遣いで話す。問題解決に向けて合理的な提案をする。"),
    ("beginner", "基本的な内容に関する質問が多く、専門用語には不慣れである。"),
]

def user_completion(history, persona):
    original_query = history[0][0]
    prompt = USER_PROMPT.replace("#QUERY#", original_query).replace("#PERSONA#", persona[1])
    messages = [
        {"role": "system", "content": prompt},
        {"role": "user", "content": "こちらは相談窓口です。本日はどのようなご相談でしょうか？"},
    ]
    for h in history:
        messages.append({"role": "assistant", "content": h[0]})
        messages.append({"role": "user", "content": h[1]})
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        temperature=0.7,
        messages=messages,
    )
    return response.choices[0].message.content

def evaluate(query, documents, response):
    context = "\n---\n".join([doc.page_content for doc in documents])
    prompt = (
        EVALUATE_PROMPT.replace("#CONTEXT#", context)
        .replace("#QUESTION#", query)
        .replace("#RESPONSE#", response)
    )
    messages = [{"role": "system", "content": prompt}]
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        temperature=0.2,
        messages=messages,
    )
    return response.choices[0].message.content

def parse_evaluate_output(response):
    try:
        response = (
            response.replace("```json", "")
            .replace("```", "")
            .replace("{{", "{")
            .replace("}}", "}")
        )
        response_json = json.loads(response)
        score = float(int(response_json["Overall"]) / 5)  # 1~5点を0.0~1.0点に変換
    except:
        score = 0.0

    return score

def main():
    # authorization
    gc = gspread.authorize(credentials)
    sh = gc.open_by_key("<YOUR_SPREADSHEET_ID>")
    worksheet_title = "hoge"
    sheets = [sheet.title for sheet in sh.worksheets()]
    if worksheet_title in sheets:
        worksheet = sh.worksheet(worksheet_title)
    else:
        worksheet = sh.add_worksheet(title=worksheet_title, rows=100, cols=20)
    worksheet.append_row(
        [
            "config",
            "user1",
            "assistant1",
            "check1",
            "user2",
            "assistant2",
            "check2",
            "user3",
            "assistant3",
            "check3",
        ]
    )

    # begin evaluation
    overall_scores = []
    empty_row = len(worksheet.get_all_values()) + 1
    for i, query in enumerate(queries):
        curr_col = "A"
        curr_row = empty_row + i

        persona = random.choice(personas)
        metadata = {
            "date": (datetime.datetime.utcnow() + datetime.timedelta(hours=9)).strftime(
                "%Y/%m/%d %H:%M:%S"
            ),
            "persona": persona[0],
        }
        worksheet.update(f"{curr_col}{curr_row}", json.dumps(metadata, ensure_ascii=False))
        curr_col = chr(ord(curr_col) + 1)

        # begin conversation
        history = []
        scores = []
        while len(history) < MAX_TURN:
            # generate user query
            if len(history) > 0:
                query = user_completion(history, persona)
            # finish conversation if user says "DONE"
            if "DONE" in query:
                history.append([query, ""])
                break
            worksheet.update(f"{curr_col}{curr_row}", query)
            curr_col = chr(ord(curr_col) + 1)

            # RAG
            _, history, docs_str = rag(query, history, RAG_PROMPT)
            answer = history[-1][1]
            worksheet.update(f"{curr_col}{curr_row}", answer)
            rag_output_cell = f"{curr_col}{curr_row}"
            curr_col = chr(ord(curr_col) + 1)

            # evaluation
            turn_response = evaluate(query, docs_str, answer)
            turn_score = parse_evaluate_output(turn_response)
            worksheet.update(f"{curr_col}{curr_row}", turn_response)
            curr_col = chr(ord(curr_col) + 1)
            scores.append(turn_score)

            # alert if turn score is low
            if turn_score < THRESHOLD:
                worksheet.format(
                    rag_output_cell,
                    {
                        "backgroundColor": {
                            "red": 250 / 255,
                            "green": 219 / 255,
                            "blue": 218 / 255,
                            "alpha": 1,
                        }
                    },
                )

        # ターンごとのスコアを平均して対話に対するスコアとする
        # scoring
        avg_score = sum(scores) / len(scores)
        overall_scores.append(avg_score)

    score = sum(overall_scores) / len(overall_scores)
    print(f"{score=}")

    return


if __name__ == "__main__":
    main()