前回の記事で、Langfuseを使ってローカルでのPrompts （プロンプトの管理に関する機能）とTracing （トレースの収集に関する機能）を試してみました。
今回はEvaluation （評価に関する機能）です。書籍の7章のメインテーマである「RAGの評価（Evaluation）」に該当します。書籍ではLangSmithとRAG評価フレームワーク「Ragas」を組み合わせています。前回に引き続き、LangSmithの代わりにLangfuseを使うことでローカル＆無料で試してみます。
全体の流れ
RAGを使った処理の評価には、オフライン評価とオンライン評価がありますが、オフライン評価をやる。
全体の流れとしては以下の通り。
- Ragasを使ってデータセットを作成しLangfuseに登録する
- 評価対象の処理を実行する
- このとき、上記のデータセットを利用する
- 処理に加えて評価が行われ、評価結果がLangfuseに登録される
- Langfuseの画面から評価結果を確認する
1. Ragasを使ってデータセットを作成しLangfuseに登録する
RAGを評価するためには「問題」と「模範解答」のデータセット（Dataset）が必要。手で作るのは大変なので、Ragasの合成データ生成機能を利用する。LLMなどを使って生成したデータを合成データと呼ぶみたい。
LangChainリポジトリのドキュメントを元にRagasを使って合成データを生成し、Langfuseの Datasets に保存する。
では実装へ。
前々回、前回の流れの続きで、追加で $ uv add ragas==0.1.14 nest-asyncio==1.6.0 する。
ソースコードは以下の通り。以下の2点が書籍と異なる。
- modelがLM StudioでLoadしているものになっていること
- データセットの登録先がLangfuseになっていること
from langchain_community.document_loaders import GitLoader
def file_filter(file_path: str) -> bool:
return file_path.endswith(".md")
loader = GitLoader(
clone_url="https://github.com/langchain-ai/langchain",
repo_path="./langchain",
branch="master",
file_filter=file_filter,
)
documents = loader.load()
print(f"len(documents) = {len(documents)}")
import os
import nest_asyncio
from ragas.run_config import RunConfig
from ragas.testset.generator import TestsetGenerator
from ragas.testset.evolutions import simple, reasoning, multi_context
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
for document in documents:
document.metadata["filename"] = document.metadata["source"]
nest_asyncio.apply()
api_key = "dummy"
model = ChatOpenAI(
base_url = "http://localhost:1234/v1",
api_key = api_key,
temperature = 0,
)
embeddings = OpenAIEmbeddings(
base_url="http://localhost:1234/v1",
api_key = api_key,
model="text-embedding-nomic-embed-text-v1.5",
check_embedding_ctx_length=False,
)
generator = TestsetGenerator.from_langchain(
generator_llm = model,
critic_llm = model,
embeddings = embeddings,
)
testset = generator.generate_with_langchain_docs(
documents,
test_size = 4,
distributions = {
simple: 0.5,
reasoning: 0.25,
multi_context: 0.25,
},
)
print(testset.to_pandas())
from langfuse import Langfuse
os.environ["LANGFUSE_SECRET_KEY"] = "sk-..."
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-..."
os.environ["LANGFUSE_BASE_URL"] = "http://localhost:3000"
langfuse = Langfuse()
dataset_name = "langchain docs"
langfuse.create_dataset(
name=dataset_name,
description="My first dataset",
metadata={
"author": "kariyayo",
"date": "2026-07-01",
"type": "benchmark"
}
)
for testset_record in testset.test_data:
langfuse.create_dataset_item(
dataset_name = dataset_name,
input = {"question": testset_record.question},
expected_output = {"contexts": testset_record.contexts, "ground_truth": testset_record.ground_truth},
metadata = {"source": testset_record.metadata[0]["source"], "evolution_type": testset_record.evolution_type},
)
いざ実行。前回、前々回の続きなので、LM StudioのサーバーとLangfuseは起動している状態。
すると以下のメッセージが。そして最終的には例外を投げてエラー終了。
LangChainDeprecationWarning: As of langchain-core 0.3.0, LangChain uses pydantic v2 internally. The langchain_core.pydantic_v1 module was a compatibility shim for pydantic v1, and should no longer be used. Please update the code to import from Pydantic directly.
For example, replace imports like: `from langchain_core.pydantic_v1 import BaseModel`
with: `from pydantic import BaseModel`
or the v1 compatibility namespace if you are working in a code base that has not been fully upgraded to pydantic 2 yet. from pydantic.v1 import BaseModel
from ragas.metrics._answer_correctness import AnswerCorrectness, answer_correctness
ここまではLangChain v0.3系で進めてきた。書籍にも注釈として書いてあるのだけど、当時Ragasのバージョンは0.1系でLangChain v0.3に対応していなかったらしく、書籍ではRagasを使うところはLangChain v0.2系で進めている。
今ではRagas v0.2以降がリリースされていて、LangChain v0.3系に対応している。しかし、Ragas v0.2で破壊的変更があったようで、書籍の内容を読み進めることが難しくなる。なので、書籍の通りLangChainのバージョンを0.2系に下げて進めることにする。
$ uv add langchain-core==0.2.30 langchain-openai==0.1.21 langchain-community==0.2.12 "langchain-text-splitters<0.3.0" langchain-chroma==0.1.2 chromadb==0.5.3 ragas==0.1.14 nest-asyncio==1.6.0
これで実行できた！けど、すごく時間が掛かる。。PCのスペック、キツい。。
これが、、
% uv run regas_langfuse.py
len(documents) = 29
embedding nodes: 85%|█████████████████████████████████████████████████████████████████████████████████████████████████▌ | 73/86 [15:19<05:18, 24.47s/it]
こうなって、、
% uv run regas_langfuse.py
len(documents) = 29
Generating: 0%| | 0/4 [00:00<?, ?it/s]
こう！
% uv run regas_langfuse.py
len(documents) = 29
Generating: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 4/4 [19:25<00:00, 291.43s/it]
question contexts ... metadata episode_done
0 What are legacy chains in the LangChain ecosys... [# 🦜️🔗 LangChain Classic\n\n[![PyPI - Version]... ... [{'source': 'libs/langchain/README.md', 'file_... True
1 What is the purpose of the LangChain integrati... [# langchain-openrouter\n\n[
登録されている行をクリックするとこんな感じの画面。
データセットとは？
Langfuseのドキュメントにデータ構造の図があります。
Experiments Data Model - Langfuse
- 評価のための処理を実行すると、DatasetRunができる
- DatasetRunはn個のDatasetRunItemを持つ
- DatasetRunItemは、1つのDataSetItemと関連し、1つのTraceと関連する
この関係が重要。このページを読まずになんとなく評価実行のソースコードを書いても上手くいかず。。では、次は評価実行のソースコードへ。
2. 評価対象の処理を実行する
データセットの登録ができたので、評価のステップに進む。
まずは評価対象のコードを用意。
from langchain_community.document_loaders import GitLoader
def file_filter(file_path: str) -> bool:
return file_path.endswith(".md")
loader = GitLoader(
clone_url = "https://github.com/langchain-ai/langchain",
repo_path = "./langchain",
branch = "master",
file_filter = file_filter,
)
documents = loader.load()
print(len(documents))
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
api_key = "dummy"
model = ChatOpenAI(
base_url = "http://localhost:1234/v1",
api_key = api_key,
temperature = 0.0,
timeout = 600,
max_retries = 1,
)
embeddings = OpenAIEmbeddings(
base_url="http://localhost:1234/v1",
api_key = api_key,
model="text-embedding-nomic-embed-text-v1.5",
check_embedding_ctx_length=False,
)
db = Chroma.from_documents(documents, embeddings)
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
prompt = ChatPromptTemplate.from_template('''\
以下の文脈だけを踏まえて質問に回答してください。
文脈："""
{context}
"""
質問：{question}
''')
retriever = db.as_retriever()
chain = RunnableParallel(
{
"question": RunnablePassthrough(),
"context": retriever,
}
).assign(answer=prompt | model | StrOutputParser())
ここまでは特に新しいことはなし。
続けて実行と評価の処理のソースコード。ここは先述したDataSetsのデータ構造を把握する前に書き始めたらなかなか上手くいかなかった部分。
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from ragas.metrics import answer_relevancy, context_precision
ragas_llm = LangchainLLMWrapper(langchain_llm=model)
ragas_embeddings = LangchainEmbeddingsWrapper(embeddings)
context_precision.llm = ragas_llm
context_precision.embeddings = ragas_embeddings
answer_relevancy.llm = ragas_llm
answer_relevancy.embeddings = ragas_embeddings
from langfuse import Langfuse
os.environ["LANGFUSE_SECRET_KEY"] = "sk-..."
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-..."
os.environ["LANGFUSE_BASE_URL"] = "http://localhost:3000"
langfuse = Langfuse()
DATASET_NAME = "langchain-docs"
langfuse_dataset = langfuse.get_dataset(DATASET_NAME)
from datasets import Dataset
from ragas import evaluate
from langfuse.langchain import CallbackHandler
evaluation_data = {
"question": [],
"contexts": [],
"answer": [],
"ground_truth": [],
}
traces_to_link = []
trace_ids = []
print("評価を開始します。RAGアプリケーションに問題を解かせています...")
RUN_NAME = "LM-Studio-RAG-Evaluation"
for item in langfuse_dataset.items:
question = item.input["question"]
ground_truth = item.expected_output.get("ground_truth", "")
langfuse_handler = CallbackHandler()
response = chain.invoke(
question,
config={"callbacks": [langfuse_handler]},
)
answer = response["answer"]
raw_context = response["context"]
contexts = []
for doc in raw_context:
if hasattr(doc, "page_content"):
contexts.append(doc.page_content)
elif isinstance(doc, dict):
contexts.append(doc.get("page_content", str(doc)))
else:
contexts.append(str(doc))
if "ground_truths" not in evaluation_data:
evaluation_data["ground_truths"] = []
evaluation_data["question"].append(str(question))
evaluation_data["contexts"].append(contexts)
evaluation_data["answer"].append(str(answer))
evaluation_data["ground_truth"].append(str(ground_truth))
evaluation_data["ground_truths"].append([str(ground_truth)])
trace_ids.append(langfuse_handler.last_trace_id)
traces_to_link.append(item)
ragas_dataset = Dataset.from_dict(evaluation_data)
print("スコア計算を開始します...")
from ragas.run_config import RunConfig
result = evaluate(
dataset=ragas_dataset,
metrics=[context_precision, answer_relevancy],
run_config=RunConfig(max_workers=1, max_retries=3)
)
df = result.to_pandas()
print("\n=== 採点結果のプレビュー ===")
print(df)
print("\n 結果をLangfuseの Dataset Run に記録しています...")
def safe_score(val):
import math
return 0.0 if (val is None or (isinstance(val, float) and math.isnan(val))) else float(val)
for i, item in enumerate(traces_to_link):
row = df.iloc[i]
actual_trace_id = trace_ids[i]
try:
langfuse.api.dataset_run_items.create(
run_name=RUN_NAME,
dataset_item_id=item.id,
trace_id=actual_trace_id
)
except Exception as e:
print(f"Dataset Run Itemへの紐付けに失敗しました (item_id: {item.id}): {e}")
if "answer_relevancy" in row:
langfuse.create_score(
name="answer_relevancy",
value=safe_score(row["answer_relevancy"]),
trace_id=actual_trace_id
)
if "context_precision" in row:
langfuse.create_score(
name="context_precision",
value=safe_score(row["context_precision"]),
trace_id=actual_trace_id
)
print("評価が完了しました！")
いざ、実行！
$ uv run evaluation_langfuse.py
29
Failed to send telemetry event ClientStartEvent: capture() takes 1 positional argument but 3 were given
Failed to send telemetry event ClientCreateCollectionEvent: capture() takes 1 positional argument but 3 were given
評価を開始します。RAGアプリケーションに問題を解かせています... Failed to send telemetry event CollectionQueryEvent: capture() takes 1 positional argument but 3 were given
スコア計算を開始します...
Evaluating: 100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 8/8 [11:32<00:00, 86.53s/it]
=== 採点結果のプレビュー ===
question contexts ... context_precision answer_relevancy
0 How can devs contribute to LangChain while cho... [# 🦜️🔗 LangChain Classic\n\n[![PyPI - Version]... ... 0.0 0.808152
1 How does LangChain‑OpenRouter let you tap mult... [# langchain-openrouter\n\n[
データセットを選択して「Experiments」タブを選択する。
この画面に表示される行が、データ構造上のDatasetRunに対応している。
DatasetRunをクリック。DatasetRunItemの一覧画面へ。
「Trace」列のアイコンをクリック。Traceが見れる。ベクトル検索の結果、4つのmarkdownファイルの内容が返ってきてる。
それで、以下のプロンプトをOpenAI APIに投げてることが分かる。
以下の文脈だけを踏まえて質問に回答してください。
文脈："""
... <- ここにベクトル検索結果の4つのmarkdownファイルの中身
"""
質問：What are legacy chains in the LangChain ecosystem, and why should most users prefer using the main langchain package instead?
OpenAI APIからのレスポンスも確認できる。レスポンスにはしっかりと説明が書かれている。
だけど、データセットの期待値としては、「回答できない」という内容である。したがって、
- answer_relevancy: 0.96
- context_precision: 0.00
- 「以下の文脈だけを踏まえて」とプロンプトに書いたのに、データセットのExpected outputのContextからは得られないような情報まで出力されている
ということかな。そして、レスポンスの内容がハルシネーションではないと人間が見て判断できたなら、データセットが間違っていると。
おしまい
『LangChainとLangGraphによるRAG・AIエージェント』と、ローカルで動かすための試行錯誤のおかげで、知らない世界へ一歩踏み込めた気がします！
