こんにちは。本記事では、Langfuse で取得したLLMアプリケーションのtrace情報を、Langfuse 外の評価システムで評価するパイプラインをご紹介します。
この記事で得られること
LLMアプリケーションのtrace情報と、別の評価システムによる評価結果を Langfuse に集約することで、
- 実行・評価結果の集約と可視化
- 評価の振り返り支援
- 判断材料の蓄積
ができます。
紹介するパイプラインの適用案
ユースケース1
- シナリオ: (1)CI(Continuous Integration)でpull requestをトリガーとしてLLMの動作系を動かし、(2)CI系と別のシステムでその結果を評価し、(3)再度CI系その評価結果を基にpull requestの承認可否を決める。
- 問題: 複数段階、複数システムで発生する関連情報が散在していると活用しづらい。
- 課題解決: (1)(2)(3)のtrace、出力をひとまとめにして活用できる。
ユースケース2
- シナリオ: LLM出力を伴うシステムの開発でプロンプトの表現調整をするとき、プロンプト修正->LLM生成を繰り返す。
- 問題: プロンプトとLLMの出力結果、その評価結果等の情報が都度発生する。この情報を振り返りやすい形で残すのは、開発者にとって必要だがきわめて面倒である。
- 課題解決: 各回の試行毎に関連する情報をひとまとめにしておけるため、振り返りがしやすくなる。
パイプライン
ユースケース1をイメージしています。
1. CI系で、LLMアプリケーションを実行する。
CI系で、以下の手順を実施します。
- LLMアプリケーションを動作させ実行結果が返ってくる。
- LLMアプリケーション実行時のtraceをLangfuseに登録する。
- Langfuseに問い合わせてtrace_idを取得する。
今回の紹介の中では、LLMアプリケーションはAzureOpenAIを利用します。
まず、準備としてlangfuseラップされたAzureOpenAIインスタンスを生成します。
import os
from dotenv import load_dotenv
_ = load_dotenv()
from langfuse.openai import AzureOpenAI
client = AzureOpenAI(
api_version=os.environ["AZURE_OPENAI_API_VERSION"],
azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_key=os.environ["AZURE_OPENAI_API_KEY"],
)
# 参考: AzureでないOpenAIクラスを使う場合
# from langfuse.openai import OpenAI
#
# client = OpenAI(
# base_url=os.environ["OPENAI_ENDPOINT"],
# api_key=os.environ["OPENAI_API_KEY"],
# )
先に挙げた手順を実行します。
import json
from langfuse import observe, get_client
@observe
def make_topic():
topic_suggestion = """
あなたは、ビジネス雑誌の編集者です。読者の注目をあびるトピックを5個生成してください。
出力は生成したトピックの配列形式にしてください。
"""
'''
1. LLMアプリケーションを動作
2. traceをLangfuseに登録 ... langfuseのOpenAIラッパーを使用しているので暗黙的に実行される。
'''
output = client.chat.completions.create( # LLM生成部のtraceを採るだけなら、@observeの関数内に入れる必要はない。
name="prep_topic",
model = "gpt-4o",
messages=[
{
"role": "user",
"content": topic_suggestion
}
],
response_format={ # 出力形式を定義
"type": "json_schema",
"json_schema": {
"name": "list_of_strings",
"schema": {
"type": "object",
"properties": {
"topics": {
"type": "array",
"items": {"type": "string"}
}
},
"required": ["topics"]
}
}
},
temperature=1
).choices[0].message.content
''' 3. Langfuseに問い合わせてtrace_idを取得 '''
langfuse = get_client()
trace_id = langfuse.get_current_trace_id() # langfuseでの観測中でないとcurrentのtrace_idが取得できないので、OpenAIインスタンスの外側に@observeの関数を設定している。
return trace_id, output
trace_id, output = make_topic()
for ind, topic in enumerate(json.loads(output)["topics"]):
print(f"{ind}. {topic}")
# 出力例:
# 0. AI技術の未来と現在—ビジネスへの影響
# 1. 働き方改革とリモートワークの潮流
# 2. グローバル市場におけるサステイナビリティの重要性
# 3. 金融テクノロジーの進化—デジタルトランスフォーメーション
# 4. スタートアップ成功事例—革新と挑戦の物語
langfuse = get_client()
trace_url = langfuse.get_trace_url(trace_id=trace_id)
print(trace_id) # 出力例: ddb25f1aefdd06a796bc58cdbd3d0c0c
print(trace_url) # 出力例: http://[host]/project/[project_id]/traces/ddb25f1aefdd06a796bc58cdbd3d0c0c ... 末尾はtrace_id
Langfuseのtrace管理画面に、trace情報が追加されます。（下図の赤枠）
また、上記スクリプト末尾のtrace_urlにアクセスすると、Langfuse に登録されたtraceを確認できます。
2. 評価系で評価を行い、結果をLangfuseに登録する。
CI系で出力した、trace_idとoutputを評価系の入力として、
評価系で、以下の手順を実行します。
- 評価を行う。
- 評価結果スコアをLangfuseのtrace_idが紐づくtraceにセットする。
今回の紹介の中では、LLMアプリケーションはAzureOpenAIを利用します。
まず、準備としてAzureOpenAIインスタンスを生成します。今回、評価の状況はLangfuseで管理しないので、langfuse でラップされていないクラスを使います。
注：langfuseのラップ有/無のAzureOpenAIインスタンスが名前空間で競合して、意図しないtrace記録が発生するので、CI系と評価系は別スクリプトにするのが望ましいです。
import os
from dotenv import load_dotenv
_ = load_dotenv()
from openai import AzureOpenAI
client = AzureOpenAI(
api_version=os.environ["AZURE_OPENAI_API_VERSION"],
azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_key=os.environ["AZURE_OPENAI_API_KEY"],
)
# 参考
# AzureでないOpenAIクラスを使う場合
#
# from openai import OpenAI
#
# client = OpenAI(
# base_url=os.environ["OPENAI_ENDPOINT"],
# api_key=os.environ["OPENAI_API_KEY"],
# )
trace_idとoutputをセットします。
''' 入力データをセット '''
trace_id = 'ddb25f1aefdd06a796bc58cdbd3d0c0c'
output = '{"topics":["AI技術の未来と現在—ビジネスへの影響","働き方改革とリモートワークの潮流","グローバル市場におけるサステイナビリティの重要性","金融テクノロジーの進化—デジタルトランスフォーメーション","スタートアップ成功事例—革新と挑戦の物語"]}'
topics = json.loads(output)["topics"]
評価を実施します。
from langfuse import get_client
def evaluate_topic(topics, trace_id):
prompt_template = """\
あなたはビジネス雑誌の長年の読者です。
以下のテキストに興味を持てるか判定してスコアを算出してください。
出力仕様は、0.0（最低、興味なし）- 1.0（最高、興味あり）、で、float値のみ出力してください。
テキスト: {text}
"""
''' 1. 評価を実行 '''
results = []
for i, topic in enumerate(topics):
prompt = prompt_template.format(text=topic)
output = client.chat.completions.create(
model = "gpt-4o",
messages=[
{
"role": "user",
"content": prompt
}
],
temperature=1
).choices[0].message.content
results.append(output)
results = [float(result) for result in results]
score = sum(results) / len(results) # 評価結果を判断するためのscoreを算出する。
''' 2. 評価結果スコアをLangfuseのtrace_idが紐づくtraceにセット '''
langfuse = get_client()
langfuse.create_score( # 取得しておいたtrace_idが紐づいたtraceにscoreを登録する。
trace_id=trace_id,
name="score",
value=score,
)
evaluate_topic(topics, trace_id)
Langfuseの管理画面にscore列が追加され、trace_idに紐づいたtraceのscore値が入力されている。
先ほどのtrace_urlの遷移先にもscore値が表示されている。
3. CI系で評価系で登録したスコアを参照し結果の判定をする。
最初にCI系で取得したtrace_idを使って、
CI系で、以下の手順を実行します。
- Langfuseに登録された先のtrace_idが紐づくtraceにセットされたスコアを参照
- 結果を判定
from dotenv import load_dotenv
_ = load_dotenv()
from langfuse import get_client
THRESHOLD = 0.7
def judge_score(trace_id):
''' 1. Langfuse のスコアを参照 '''
langfuse = get_client()
score = [s.value for s in langfuse.api.trace.get(trace_id=trace_id).scores if s.name=="score"][0]
''' 2. 結果を判定 '''
if score >= THRESHOLD: # scoreがしきい値以上ならTrueと判断する。
judge = True
else:
judge = False
return score, judge
score, judge = judge_score(trace_id)
print(f"score: {score:.2f}, judge: {judge}")
# 出力例
# score: 0.90, judge: True
まとめ
LLMアプリケーションのtrace情報と、別の評価システムによる評価結果を Langfuse に集約することにより、
- 実行・評価結果の集約と可視化
- 評価の振り返り支援
- 判断材料の蓄積
が可能に。CI、繰り返しのLLM実行調整がしやすくなります。LLMアプリケーションを開発する際は、とりあえずLangfuseに情報を蓄積させることをおすすめします。
Discussion