如何评估应用程序的中间步骤

在许多场景中，评估任务的最终输出可能已足够，但在某些情况下，您可能希望评估流水线的中间步骤。

例如，对于检索增强生成（RAG），您可能希望：

1. 评估检索步骤，以确保根据输入查询检索到正确的文档。
2. 评估生成步骤，以确保根据检索到的文档生成正确的答案。

在本指南中，我们将使用一个简单的、完全自定义的评估器来评估标准1，并使用一个基于LLM的评估器来评估标准2，以突出这两种场景。

为了评估流水线的中间步骤，您的评估器函数应遍历并处理 run/rootRun 参数，这是一个包含流水线中间步骤的 Run 对象。

1. 定义您的LLM流水线

下面的RAG流水线包括：1）根据输入问题生成维基百科查询，2）从维基百科检索相关文档，以及3）根据检索到的文档生成答案。

Python

pip install -U langsmith langchain[openai] wikipedia

yarn add langsmith langchain @langchain/openai wikipedia

需要 langsmith>=0.3.13

Python

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:
    """生成用于在维基百科中搜索的查询。"""
    instructions = (
        "生成一个搜索查询以传入维基百科来回答用户的问题。"
        "仅返回搜索查询，不要返回其他内容。"
        "这将直接传递给维基百科搜索引擎。"
    )
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": question}
    ]
    result = oai_client.chat.completions.create(
        messages=messages,
        model="gpt-4.1-mini",
        temperature=0,
    )
    return result.choices[0].message.content

@traceable(run_type="retriever")
def retrieve(query: str) -> list:
    """获取最多两个维基百科搜索结果。"""
    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:
    """根据检索到的信息回答问题。"""
    instructions = f"仅根据以下内容回答用户的问题：\n\n{context}"
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": question}
    ]
    result = oai_client.chat.completions.create(
        messages=messages,
        model="gpt-4.1-mini",
        temperature=0
    )
    return result.choices[0].message.content

@traceable
def qa_pipeline(question: str) -> str:
    """完整的流水线。"""
    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:
          "生成一个搜索查询以传入维基百科来回答用户的问题。仅返回搜索查询，不要返回其他内容。这将直接传递给维基百科搜索引擎。",
      },
      { role: "user" as const, content: input.question },
    ];
    const chatCompletion = await openai.chat.completions.create({
      model: "gpt-4.1-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) {
        // 目前只返回前2个页面
        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: `仅根据以下内容回答用户的问题：\n\n${input.context}`,
      },
      { role: "user" as const, content: input.question },
    ];
    const chatCompletion = await openai.chat.completions.create({
      model: "gpt-4.1-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

Python

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. 定义您的自定义评估器

如上所述，我们将定义两个评估器：一个评估检索到的文档相对于输入查询的相关性，另一个评估生成的答案相对于检索到的文档的幻觉程度。我们将使用LangChain LLM包装器以及 with_structured_output 来定义幻觉评估器。

这里的关键在于，评估器函数应遍历 run / rootRun 参数以访问流水线的中间步骤。然后，评估器可以处理中间步骤的输入和输出，以根据所需标准进行评估。

示例使用 langchain 是为了方便，这不是必需的。

Python

from langchain.chat_models import init_chat_model
from langsmith.schemas import Run
from pydantic import BaseModel, Field

def document_relevance(run: Run) -> bool:
    """检查检索器输入是否存在于检索到的文档中。"""
    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

# 数据模型
class GradeHallucinations(BaseModel):
    """生成答案中是否存在幻觉的二元分数。"""
    is_grounded: bool = Field(..., description="如果答案基于事实，则为True，否则为False。")

# 用于分级幻觉的具有结构化输出的LLM
# 更多信息请参见：https://docs.langchain.com/oss/python/langchain/structured-output
grader_llm= init_chat_model("gpt-4.1-mini", temperature=0).with_structured_output(
    GradeHallucinations,
    method="json_schema",
    strict=True,
)

def no_hallucination(run: Run) -> bool:
    """检查答案是否基于文档。
    如果没有幻觉，则返回True，否则返回False。
    """
    # 获取文档和答案
    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"]
    )

    # 构建提示
    instructions = (
        "您是一个评估者，负责评估LLM生成是否基于/支持一组检索到的事实。"
        "给出一个二元分数1或0，其中1表示答案基于/支持这组事实。"
    )
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": f"事实集：\n{retrieved_content}\n\nLLM生成：{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;
}

// 一个非常简单的评估器，检查检索步骤的输入是否存在于检索到的文档中。
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",
      [
        `您是一个评估者，负责评估LLM生成是否基于/支持一组检索到的事实。\n`,
        `给出一个二元分数1或0，其中1表示答案基于/支持这组事实。`,
      ].join("\n"),
    ],
    [
      "human",
      "事实集： \n\n {documents} \n\n LLM生成： {generation}",
    ],
  ]);

  const llm = new ChatOpenAI({
    model: "gpt-4.1-mini",
    temperature: 0,
  }).withStructuredOutput(
    z
      .object({
        binary_score: z
          .number()
          .describe("答案基于事实，1或0"),
      })
      .describe("生成答案中是否存在幻觉的二元分数。")
  );

  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。

Python

def qa_wrapper(inputs: dict) -> dict:
  """包装qa_pipeline，使其可以接受Example.inputs字典作为输入。"""
  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",
});

实验将包含评估结果，包括评估器的分数和评论：

相关

• 评估 langgraph 图

---

Edit this page on GitHub or file an issue.

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.