애플리케이션의 중간 단계 평가하는 방법
애플리케이션의 중간 단계 평가하는 방법
많은 시나리오에서 작업의 최종 출력만 평가해도 충분하지만, 경우에 따라 파이프라인의 중간 단계를 평가하고 싶을 수 있어요.
예를 들어, 검색 증강 생성(RAG)의 경우 다음과 같이 하고 싶을 수 있습니다:
- 검색 단계를 평가하여 입력 쿼리에 대해 올바른 문서가 검색되는지 확인.
- 생성 단계를 평가하여 검색된 문서에 대해 올바른 답변이 생성되는지 확인.
이 가이드에서는 두 시나리오를 강조하기 위해 기준 1을 평가하는 간단한 완전 커스텀 평가기와 기준 2를 평가하는 LLM 기반 평가기를 사용할게요.
파이프라인의 중간 단계를 평가하려면 평가기 함수가 파이프라인의 중간 단계를 포함하는 Run 객체인 run/rootRun 인자를 순회하며 처리해야 합니다.
출처: 문서
본문
1. LLM 파이프라인 정의하기
아래 RAG 파이프라인은 1) 입력 질문이 주어지면 Wikipedia 쿼리 생성, 2) Wikipedia에서 관련 문서 검색, 3) 검색된 문서를 바탕으로 답변 생성으로 구성됩니다.
pip install -U langsmith langchain[openai] wikipedia
yarn add langsmith langchain @langchain/openai wikipedia
langsmith>=0.3.13 필요
import wikipedia as wp
from openai import OpenAI
from langsmith import traceable, wrappers
oai_client = wrappers.wrap_openai(OpenAI())
@traceable
def generate_wiki_search(question: str) -> str:
"""Generate the query to search in wikipedia."""
instructions = (
"Generate a search query to pass into wikipedia to answer the user's question. "
"Return only the search query and nothing more. "
"This will passed in directly to the wikipedia search engine."
)
messages = [
{"role": "system", "content": instructions},
{"role": "user", "content": question}
]
result = oai_client.chat.completions.create(
messages=messages,
model="gpt-5.4-mini",
temperature=0,
)
return result.choices[0].message.content
@traceable(run_type="retriever")
def retrieve(query: str) -> list:
"""Get up to two search wikipedia results."""
results = []
for term in wp.search(query, results = 10):
try:
page = wp.page(term, auto_suggest=False)
results.append({
"page_content": page.summary,
"type": "Document",
"metadata": {"url": page.url}
})
except wp.DisambiguationError:
pass
if len(results) >= 2:
return results
@traceable
def generate_answer(question: str, context: str) -> str:
"""Answer the question based on the retrieved information."""
instructions = f"Answer the user's question based ONLY on the content below:\n\n{context}"
messages = [
{"role": "system", "content": instructions},
{"role": "user", "content": question}
]
result = oai_client.chat.completions.create(
messages=messages,
model="gpt-5.4-mini",
temperature=0
)
return result.choices[0].message.content
@traceable
def qa_pipeline(question: str) -> str:
"""The full pipeline."""
query = generate_wiki_search(question)
context = "\n\n".join([doc["page_content"] for doc in retrieve(query)])
return generate_answer(question, context)
import OpenAI from "openai";
import wiki from "wikipedia";
import { Client } from "langsmith";
import { traceable } from "langsmith/traceable";
import { wrapOpenAI } from "langsmith/wrappers";
const openai = wrapOpenAI(new OpenAI());
const generateWikiSearch = traceable(
async (input: { question: string }) => {
const messages = [
{
role: "system" as const,
content:
"Generate a search query to pass into Wikipedia to answer the user's question. Return only the search query and nothing more. This will be passed in directly to the Wikipedia search engine.",
},
{ role: "user" as const, content: input.question },
];
const chatCompletion = await openai.chat.completions.create({
model: "gpt-5.4-mini",
messages: messages,
temperature: 0,
});
return chatCompletion.choices[0].message.content ?? "";
},
{ name: "generateWikiSearch" }
);
const retrieve = traceable(
async (input: { query: string; numDocuments: number }) => {
const { results } = await wiki.search(input.query, { limit: 10 });
const finalResults: Array<{
page_content: string;
type: "Document";
metadata: { url: string };
}> = [];
for (const result of results) {
if (finalResults.length >= input.numDocuments) {
// Just return the top 2 pages for now
break;
}
const page = await wiki.page(result.title, { autoSuggest: false });
const summary = await page.summary();
finalResults.push({
page_content: summary.extract,
type: "Document",
metadata: { url: page.fullurl },
});
}
return finalResults;
},
{ name: "retrieve", run_type: "retriever" }
);
const generateAnswer = traceable(
async (input: { question: string; context: string }) => {
const messages = [
{
role: "system" as const,
content: `Answer the user's question based only on the content below:\n\n${input.context}`,
},
{ role: "user" as const, content: input.question },
];
const chatCompletion = await openai.chat.completions.create({
model: "gpt-5.4-mini",
messages: messages,
temperature: 0,
});
return chatCompletion.choices[0].message.content ?? "";
},
{ name: "generateAnswer" }
);
const ragPipeline = traceable(
async ({ question }: { question: string }, numDocuments: number = 2) => {
const query = await generateWikiSearch({ question });
const retrieverResults = await retrieve({ query, numDocuments });
const context = retrieverResults
.map((result) => result.page_content)
.join("\n\n");
const answer = await generateAnswer({ question, context });
return answer;
},
{ name: "ragPipeline" }
);
이 파이프라인은 다음과 같은 트레이스를 생성합니다:
2. 파이프라인을 평가할 데이터셋과 예시 생성하기
파이프라인을 평가하기 위해 몇 개의 예시가 있는 매우 간단한 데이터셋을 만들고 있어요.
langsmith>=0.3.13 필요
from langsmith import Client
ls_client = Client()
dataset_name = "Wikipedia RAG"
if not ls_client.has_dataset(dataset_name=dataset_name):
dataset = ls_client.create_dataset(dataset_name=dataset_name)
examples = [
{"inputs": {"question": "What is LangChain?"}},
{"inputs": {"question": "What is LangSmith?"}},
]
ls_client.create_examples(
dataset_id=dataset.id,
examples=examples,
)
import { Client } from "langsmith";
const client = new Client();
const examples = [
[
"What is LangChain?",
"LangChain is an open-source framework for building applications using large language models.",
],
[
"What is LangSmith?",
"LangSmith is an observability and evaluation tool for LLM products, built by LangChain Inc.",
],
];
const datasetName = "Wikipedia RAG";
const inputs = examples.map(([input, _]) => ({ input }));
const outputs = examples.map(([_, expected]) => ({ expected }));
const dataset = await client.createDataset(datasetName);
await client.createExamples({ datasetId: dataset.id, inputs, outputs });
3. 커스텀 평가기 정의하기
위에서 언급했듯이, 두 개의 평가기를 정의할 거예요. 하나는 검색된 문서가 입력 쿼리와 관련이 있는지 평가하고, 다른 하나는 검색된 문서에 대해 생성된 답변의 환각(hallucination) 여부를 평가합니다. 환각 평가기는 LangChain LLM 래퍼와 with_structured_output을 함께 사용해 정의할게요.
여기서 핵심은 평가기 함수가 run / rootRun 인자를 순회하여 파이프라인의 중간 단계에 접근해야 한다는 것입니다. 그러면 평가기는 중간 단계의 입력과 출력을 처리하여 원하는 기준에 따라 평가할 수 있습니다.
예시는 편의를 위해 langchain을 사용하지만 필수는 아닙니다.
from langchain.chat_models import init_chat_model
from langsmith.schemas import Run
from pydantic import BaseModel, Field
def document_relevance(run: Run) -> bool:
"""Checks if retriever input exists in the retrieved docs."""
qa_pipeline_run = next(
r for run in run.child_runs if r.name == "qa_pipeline"
)
retrieve_run = next(
r for run in qa_pipeline_run.child_runs if r.name == "retrieve"
)
page_contents = "\n\n".join(
doc["page_content"] for doc in retrieve_run.outputs["output"]
)
return retrieve_run.inputs["query"] in page_contents
# Data model
class GradeHallucinations(BaseModel):
"""Binary score for hallucination present in generation answer."""
is_grounded: bool = Field(..., description="True if the answer is grounded in the facts, False otherwise.")
# LLM with structured output for grading hallucinations
# For more see: https://docs.langchain.com/oss/python/langchain/structured-output
grader_llm= init_chat_model("gpt-5.4-mini", temperature=0).with_structured_output(
GradeHallucinations,
method="json_schema",
strict=True,
)
def no_hallucination(run: Run) -> bool:
"""Check if the answer is grounded in the documents.
Return True if there is no hallucination, False otherwise.
"""
# Get documents and answer
qa_pipeline_run = next(
r for r in run.child_runs if r.name == "qa_pipeline"
)
retrieve_run = next(
r for r in qa_pipeline_run.child_runs if r.name == "retrieve"
)
retrieved_content = "\n\n".join(
doc["page_content"] for doc in retrieve_run.outputs["output"]
)
# Construct prompt
instructions = (
"You are a grader assessing whether an LLM generation is grounded in / "
"supported by a set of retrieved facts. Give a binary score 1 or 0, "
"where 1 means that the answer is grounded in / supported by the set of facts."
)
messages = [
{"role": "system", "content": instructions},
{"role": "user", "content": f"Set of facts:\n{retrieved_content}\n\nLLM generation: {run.outputs['answer']}"},
]
grade = grader_llm.invoke(messages)
return grade.is_grounded
import { EvaluationResult } from "langsmith/evaluation";
import { Run, Example } from "langsmith/schemas";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";
function findNestedRun(run: Run, search: (run: Run) => boolean): Run | null {
const queue: Run[] = [run];
while (queue.length > 0) {
const currentRun = queue.shift()!;
if (search(currentRun)) return currentRun;
queue.push(...currentRun.child_runs);
}
return null;
}
// A very simple evaluator that checks to see if the input of the retrieval step exists
// in the retrieved docs.
function documentRelevance(rootRun: Run, example: Example): EvaluationResult {
const retrieveRun = findNestedRun(rootRun, (run) => run.name === "retrieve");
const docs: Array<{ page_content: string }> | undefined =
retrieveRun.outputs?.outputs;
const pageContents = docs?.map((doc) => doc.page_content).join("\n\n");
const score = pageContents.includes(retrieveRun.inputs?.query);
return { key: "simple_document_relevance", score };
}
async function hallucination(
rootRun: Run,
example: Example
): Promise<EvaluationResult> {
const rag = findNestedRun(rootRun, (run) => run.name === "ragPipeline");
const retrieve = findNestedRun(rootRun, (run) => run.name === "retrieve");
const docs: Array<{ page_content: string }> | undefined =
retrieve.outputs?.outputs;
const documents = docs?.map((doc) => doc.page_content).join("\n\n");
const prompt = ChatPromptTemplate.fromMessages<{
documents: string;
generation: string;
}>([
[
"system",
[
`You are a grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts. \n`,
`Give a binary score 1 or 0, where 1 means that the answer is grounded in / supported by the set of facts.`,
].join("\n"),
],
[
"human",
"Set of facts: \n\n {documents} \n\n LLM generation: {generation}",
],
]);
const llm = new ChatOpenAI({
model: "gpt-5.4-mini",
temperature: 0,
}).withStructuredOutput(
z
.object({
binary_score: z
.number()
.describe("Answer is grounded in the facts, 1 or 0"),
})
.describe("Binary score for hallucination present in generation answer.")
);
const grader = prompt.pipe(llm);
const score = await grader.invoke({
documents,
generation: rag.outputs?.outputs,
});
return { key: "answer_hallucination", score: score.binary_score };
}
4. 파이프라인 평가하기
마지막으로, 위에서 정의한 커스텀 평가기로 evaluate를 실행합니다.
def qa_wrapper(inputs: dict) -> dict:
"""Wrap the qa_pipeline so it can accept the Example.inputs dict as input."""
return {"answer": qa_pipeline(inputs["question"])}
experiment_results = ls_client.evaluate(
qa_wrapper,
data=dataset_name,
evaluators=[document_relevance, no_hallucination],
experiment_prefix="rag-wiki-oai"
)
import { evaluate } from "langsmith/evaluation";
await evaluate((inputs) => ragPipeline({ question: inputs.input }), {
data: datasetName,
evaluators: [hallucination, documentRelevance],
experimentPrefix: "rag-wiki-oai",
});
실험에는 평가 결과(평가기의 점수와 주석 포함)가 담깁니다.