[Dify]LangfuseでDifyの評価用コードを書いてみた

はじめに

基本ローカルDifyとローカルLangfuseを使った評価コードを書いています。
多少環境変数やURLには注意してください。

Langfuseへのデータセットの用意

基本は以下のCookbookに従って、データセットを作成しました。

まずは必要なライブラリをinstallしておきましょう

pip install langfuse requests

Langfuse周りの環境変数設定

import os
 
os.environ["LANGFUSE_SECRET_KEY"] = "LANGFUSE_SECRET_KEY"
os.environ["LANGFUSE_PUBLIC_KEY"] = "LANGFUSE_PUBLIC_KEY"
os.environ["LANGFUSE_HOST"] ="LANGFUSE_HOST"

そしてデータセットを作成しましょう。

from langfuse import Langfuse
 
# init
langfuse = Langfuse()
langfuse.create_dataset(name="capital_cities");

そしてデータを入れます。

local_items = [
    {"input": {"country": "Italy"}, "expected_output": "Rome"},
    {"input": {"country": "Spain"}, "expected_output": "Madrid"},
    {"input": {"country": "Brazil"}, "expected_output": "Brasília"},
    {"input": {"country": "Japan"}, "expected_output": "Tokyo"},
    {"input": {"country": "India"}, "expected_output": "New Delhi"},
    {"input": {"country": "Canada"}, "expected_output": "Ottawa"},
    {"input": {"country": "South Korea"}, "expected_output": "Seoul"},
    {"input": {"country": "Argentina"}, "expected_output": "Buenos Aires"},
    {"input": {"country": "South Africa"}, "expected_output": "Pretoria"},
    {"input": {"country": "Egypt"}, "expected_output": "Cairo"},
]

for item in local_items:
  langfuse.create_dataset_item(
      dataset_name="capital_cities",
      # any python object or value
      input=item["input"],
      # any python object or value, optional
      expected_output=item["expected_output"]
)

Difyのワークフロー

今回はできるだけ簡単にするために特にプロンプトもいじっていません。

DifyへQueryを投げる関数

APIKeyやURLに関しては各自設定してください。

import requests

def dify_chat(query):
    api_key = "app-Cq7nSjZCMWT3EX14g8GBHfLU"
    url = "http://localhost/v1/chat-messages"
    headers = {
        'Authorization': f"Bearer {api_key}",
        'Content-Type': 'application/json',
    }

    data = {
        "inputs": {},
        "query": query,
        "response_mode": "blocking",  # ストリーミングではなくブロッキングモードに変更
        "user": "abc-123"
    }

    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()  # エラーチェック

    response_data = response.json()
    return response_data.get('answer', '')  # 'answer'キーの値を返す

評価コード

実際に評価していきます。
simple_evaluationでは実際の評価方法を記載しますが、今回は単に文字列一致してたらOKというような形式にしています。

from datetime import datetime
def simple_evaluation(output, expected_output):
  return output == expected_output
  
def run_my_custom_llm_app(input, system_prompt):
  messages = [
      {"role":"system", "content": system_prompt},
      {"role":"user", "content": input["country"]}
  ]
 
  generationStartTime = datetime.now()
 
  dify_answer = dify_chat(input["country"])
 
  langfuse_generation = langfuse.generation(
    name="guess-countries",
    input=messages,
    output=dify_answer,
    model="gpt-3.5-turbo",
    start_time=generationStartTime,
    end_time=datetime.now()
  )
 
  return dify_answer, langfuse_generation
  
def run_experiment(experiment_name, system_prompt):
  dataset = langfuse.get_dataset("capital_cities")
 
  for item in dataset.items:
    completion, langfuse_generation = run_my_custom_llm_app(item.input, system_prompt)
 
    item.link(langfuse_generation, experiment_name) # pass the observation/generation object or the id
 
    langfuse_generation.score(
      name="exact_match",
      value=simple_evaluation(completion, item.expected_output)
    )

評価

では実際に評価して見ましょう。

run_experiment(
    "famous_city",
    "The user will input countries, respond with the most famous city in this country"
)

結果

んー、さすがにプロンプトいじらず、評価もただの文字列一致しているか、なので精度は出ていませんね。

最後に

Difyのワークフローなどの評価はこのようにして行うとLangfuse側で精度管理ができていいですね。
ただプロンプトの管理のバージョン管理がいまいちですね。
現状はメタデータにDifyのワークフローをexportしたDSLファイルを管理しているストレージのURLを貼るとかしておかないとその辺りの管理が大変そうですね。この辺どうにかならないですかね？

次はRagasでの評価にチャレンジして見ます。

Xやってるので気になる方はフォローお願いします。
https://x.com/hudebakonosoto