跳到内容

可观察性工具。

Phoenix (Arize)

1. 引言

为 RAG 流水线构建基线通常不难，但增强它以使其适用于生产环境并确保响应质量几乎总是很困难。当有大量可用选项时，为 RAG 选择正确的工具和参数本身可能具有挑战性。本教程分享了一个健壮的工作流程，用于在构建 RAG 时做出正确选择并确保其质量。

本文介绍如何结合使用开源库来评估、可视化和分析您的 RAG。我们将使用

  • Ragas 用于合成测试数据生成和评估
  • Arize AI 的 Phoenix 用于追踪、可视化和集群分析
  • LlamaIndex 用于构建 RAG 流水线

本文将使用 arXiv 上关于提示工程的论文数据来构建 RAG 流水线。

ℹ️ 此 Notebook 需要一个 OpenAI API 密钥。

2. 安装依赖并导入库

运行以下单元格安装 Git LFS，我们将用它来下载数据集。

!git lfs install

安装并导入 Python 依赖项。

!pip install "ragas<0.1.1" pypdf arize-phoenix "openinference-instrumentation-llama-index<1.0.0" "llama-index<0.10.0" pandas

import pandas as pd

# Display the complete contents of dataframe cells.
pd.set_option("display.max_colwidth", None)

3. 配置您的 OpenAI API 密钥

如果尚未将 OpenAI API 密钥设置为环境变量，请设置它。

import os
from getpass import getpass
import openai

if not (openai_api_key := os.getenv("OPENAI_API_KEY")):
    openai_api_key = getpass("🔑 Enter your OpenAI API key: ")
openai.api_key = openai_api_key
os.environ["OPENAI_API_KEY"] = openai_api_key

4. 生成您的合成测试数据集

用于评估的黄金测试数据集的整理可能是一个漫长、繁琐且昂贵的过程，尤其是在刚开始或数据源不断变化时，这很不切实际。这可以通过合成生成高质量的数据点来解决，然后由开发人员进行验证。这可以将整理测试数据的时间和精力减少 90%。

运行以下单元格，从 arXiv 下载提示工程论文的 PDF 数据集，并使用 LlamaIndex 读取这些文档。

!git clone https://huggingface.co/datasets/explodinggradients/prompt-engineering-papers

from llama_index import SimpleDirectoryReader

dir_path = "./prompt-engineering-papers"
reader = SimpleDirectoryReader(dir_path, num_files_limit=2)
documents = reader.load_data()

理想的测试数据集应包含高质量且多样化的数据点，这些数据点的分布应与生产环境中观察到的相似。Ragas 使用独特的基于进化的合成数据生成范式来生成最高质量的问题，同时确保生成问题的多样性。Ragas 默认在底层使用 OpenAI 模型，但您可以自由选择使用任何模型。让我们使用 Ragas 生成 100 个数据点。

from ragas.testset import TestsetGenerator
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

TEST_SIZE = 25

# generator with openai models
generator_llm = ChatOpenAI(model="gpt-4o-mini")
critic_llm = ChatOpenAI(model="gpt-4o")
embeddings = OpenAIEmbeddings()

generator = TestsetGenerator.from_langchain(generator_llm, critic_llm, embeddings)

# generate testset
testset = generator.generate_with_llamaindex_docs(documents, test_size=TEST_SIZE)
test_df = testset.to_pandas()
test_df.head()

您可以根据需要自由更改问题类型的分布。既然我们的测试数据集已经准备好了，接下来让我们使用 LlamaIndex 构建一个简单的 RAG 流水线。

5. 使用 LlamaIndex 构建您的 RAG 应用

LlamaIndex 是一个易于使用且灵活的构建 RAG 应用框架。为了简单起见，我们使用默认的 LLM (gpt-3.5-turbo) 和嵌入模型 (openai-ada-2)。

在后台启动 Phoenix 并检测您的 LlamaIndex 应用，以便您的 OpenInference span 和 trace 可以发送到 Phoenix 并由其收集。OpenInference 是一个基于 OpenTelemetry 构建的开放标准，用于捕获和存储 LLM 应用的执行情况。它被设计为一种遥测数据类别，用于理解 LLM 的执行以及周围的应用上下文，例如从向量存储中检索以及使用搜索引擎或 API 等外部工具。

import phoenix as px
from llama_index import set_global_handler

session = px.launch_app()
set_global_handler("arize_phoenix")

构建您的查询引擎。

from llama_index.core import VectorStoreIndex, ServiceContext
from llama_index.embeddings.openai import OpenAIEmbedding


def build_query_engine(documents):
    vector_index = VectorStoreIndex.from_documents(
        documents,
        service_context=ServiceContext.from_defaults(chunk_size=512),
        embed_model=OpenAIEmbedding(),
    )
    query_engine = vector_index.as_query_engine(similarity_top_k=2)
    return query_engine


query_engine = build_query_engine(documents)

如果您查看 Phoenix，您应该会看到语料库数据索引时产生的嵌入 span。将这些嵌入导出并保存到数据框中，以便稍后在 Notebook 中进行可视化。

from phoenix.trace.dsl import SpanQuery

client = px.Client()
corpus_df = px.Client().query_spans(
    SpanQuery().explode(
        "embedding.embeddings",
        text="embedding.text",
        vector="embedding.vector",
    )
)
corpus_df.head()

重新启动 Phoenix 以清除累积的 trace。

px.close_app()
session = px.launch_app()

6. 评估您的 LLM 应用

Ragas 提供了一个全面的指标列表，可用于对 RAG 流水线进行组件级和端到端评估。

要使用 Ragas，我们首先构建一个评估数据集，该数据集包含问题、生成的答案、检索到的上下文以及真实答案（给定问题的实际预期答案）。

from datasets import Dataset
from tqdm.auto import tqdm
import pandas as pd


def generate_response(query_engine, question):
    response = query_engine.query(question)
    return {
        "answer": response.response,
        "contexts": [c.node.get_content() for c in response.source_nodes],
    }


def generate_ragas_dataset(query_engine, test_df):
    test_questions = test_df["question"].values
    responses = [generate_response(query_engine, q) for q in tqdm(test_questions)]

    dataset_dict = {
        "question": test_questions,
        "answer": [response["answer"] for response in responses],
        "contexts": [response["contexts"] for response in responses],
        "ground_truth": test_df["ground_truth"].values.tolist(),
    }
    ds = Dataset.from_dict(dataset_dict)
    return ds


ragas_eval_dataset = generate_ragas_dataset(query_engine, test_df)
ragas_evals_df = pd.DataFrame(ragas_eval_dataset)
ragas_evals_df.head()

查看 Phoenix 以查看您的 LlamaIndex 应用 trace。

print(session.url)

LlamaIndex application traces inside of Phoenix

我们保存了几个数据框，一个包含稍后将可视化的嵌入数据，另一个包含我们计划使用 Ragas 评估的导出 trace 和 span。

# dataset containing embeddings for visualization
query_embeddings_df = px.Client().query_spans(
    SpanQuery().explode(
        "embedding.embeddings", text="embedding.text", vector="embedding.vector"
    )
)
query_embeddings_df.head()

from phoenix.session.evaluation import get_qa_with_reference

# dataset containing span data for evaluation with Ragas
spans_dataframe = get_qa_with_reference(client)
spans_dataframe.head()

Ragas 使用 LangChain 来评估您的 LLM 应用数据。让我们使用 OpenInference 检测 LangChain，以便在评估 LLM 应用时可以看到底层发生了什么。

from openinference.instrumentation.langchain import LangChainInstrumentor

LangChainInstrumentor().instrument()

评估您的 LLM trace 并以数据框格式查看评估分数。

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_correctness,
    context_recall,
    context_precision,
)

evaluation_result = evaluate(
    dataset=ragas_eval_dataset,
    metrics=[faithfulness, answer_correctness, context_recall, context_precision],
)
eval_scores_df = pd.DataFrame(evaluation_result.scores)

将您的评估提交到 Phoenix，以便它们在您的 span 上显示为注释。

from phoenix.trace import SpanEvaluations

# Assign span ids to your ragas evaluation scores (needed so Phoenix knows where to attach the spans).
eval_data_df = pd.DataFrame(evaluation_result.dataset)
assert eval_data_df.question.to_list() == list(
    reversed(spans_dataframe.input.to_list())  # The spans are in reverse order.
), "Phoenix spans are in an unexpected order. Re-start the notebook and try again."
eval_scores_df.index = pd.Index(
    list(reversed(spans_dataframe.index.to_list())), name=spans_dataframe.index.name
)

# Log the evaluations to Phoenix.
for eval_name in eval_scores_df.columns:
    evals_df = eval_scores_df[[eval_name]].rename(columns={eval_name: "score"})
    evals = SpanEvaluations(eval_name, evals_df)
    px.Client().log_evaluations(evals)

如果您查看 Phoenix，您会看到您的 Ragas 评估在您的应用 span 上显示为注释。

print(session.url)

ragas evaluations appear as annotations on your spans

7. 可视化和分析您的嵌入

嵌入对检索到的文档和用户查询的含义进行编码。它们不仅是 RAG 系统的重要组成部分，而且对于理解和调试 LLM 应用性能也非常有用。

Phoenix 获取您的 RAG 应用中的高维嵌入，降低其维度，并将其聚类到具有语义意义的数据组中。然后，您可以选择您选择的指标（例如，Ragas 计算的忠实度或答案正确性）来直观地检查您的应用性能并发现有问题的数据集群。这种方法的优点在于，它在细粒度但有意义的数据子集上提供了指标，帮助您分析数据集上的局部性能，而不仅仅是全局性能。这也有助于您直观地了解您的 LLM 应用在回答哪些类型的查询时遇到了困难。

我们将重新启动 Phoenix 作为嵌入可视化工具，以检查我们的应用在测试数据集上的性能。

query_embeddings_df = query_embeddings_df.iloc[::-1]
assert ragas_evals_df.question.tolist() == query_embeddings_df.text.tolist()
assert test_df.question.tolist() == ragas_evals_df.question.tolist()
query_df = pd.concat(
    [
        ragas_evals_df[["question", "answer", "ground_truth"]].reset_index(drop=True),
        query_embeddings_df[["vector"]].reset_index(drop=True),
        test_df[["evolution_type"]],
        eval_scores_df.reset_index(drop=True),
    ],
    axis=1,
)
query_df.head()

query_schema = px.Schema(
    prompt_column_names=px.EmbeddingColumnNames(
        raw_data_column_name="question", vector_column_name="vector"
    ),
    response_column_names="answer",
)
corpus_schema = px.Schema(
    prompt_column_names=px.EmbeddingColumnNames(
        raw_data_column_name="text", vector_column_name="vector"
    )
)
# relaunch phoenix with a primary and corpus dataset to view embeddings
px.close_app()
session = px.launch_app(
    primary=px.Dataset(query_df, query_schema, "query"),
    corpus=px.Dataset(corpus_df.reset_index(drop=True), corpus_schema, "corpus"),
)

启动 Phoenix 后，您可以按照以下步骤使用您选择的指标可视化您的数据

  • 选择 `vector` 嵌入，
  • 选择 `Color By > dimension`，然后选择您希望按特定字段着色的维度，例如，根据 Ragas 评估分数（如忠实度或答案正确性）进行着色，
  • 从 `metric` 下拉菜单中选择您选择的指标，以查看每个集群的聚合指标。

inspect clusters of embeddings, view aggregate metrics, and color your data by the metric of your choice

8. 回顾

恭喜！您使用 Ragas 和 Phoenix 构建并评估了 LlamaIndex 查询引擎。让我们回顾一下我们学到了什么

  • 使用 Ragas，您引导生成了一个测试数据集，并计算了诸如忠实度和答案正确性等指标来评估您的 LlamaIndex 查询引擎。
  • 使用 OpenInference，您检测了您的查询引擎，以便能够观察 LlamaIndex 和 Ragas 的内部工作情况。
  • 使用 Phoenix，您收集了您的 span 和 trace，导入了您的评估以便于检查，并可视化了您的嵌入查询和检索到的文档，以识别性能不佳的区域。

本 Notebook 仅是对 Ragas 和 Phoenix 功能的介绍。要了解更多信息，请参阅 Ragas 文档和 Phoenix 文档。

如果您喜欢本教程，请在 GitHub 上留下 ⭐

LangSmith

LangSmith 是一款高级工具，旨在增强利用大型语言模型 (LLM) 的应用的开发和部署。它提供了一个全面的框架，用于追踪、分析和优化 LLM 工作流程，使开发人员更容易管理其应用内的复杂交互。

本教程解释了如何使用 LangSmith 记录 Ragas 评估的 trace。由于 Ragas 是基于 LangChain 构建的，您只需设置 LangSmith，它将自动处理 trace 的记录。

1. 设置 LangSmith

要设置 LangSmith，请确保设置以下环境变量（更多详细信息请参阅LangSmith 文档）

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
export LANGCHAIN_API_KEY=<your-api-key>
export LANGCHAIN_PROJECT=<your-project>  # Defaults to "default" if not set

2. 获取数据集

创建评估数据集或评估实例时，请确保术语与 `SingleTurnSample` 或 `MultiTurnSample` 中使用的模式匹配。

from ragas import EvaluationDataset


dataset = [
    {
        "user_input": "Which CEO is widely recognized for democratizing AI education through platforms like Coursera?",
        "retrieved_contexts": [
            "Andrew Ng, CEO of Landing AI, is known for his pioneering work in deep learning and for democratizing AI education through Coursera."
        ],
        "response": "Andrew Ng is widely recognized for democratizing AI education through platforms like Coursera.",
        "reference": "Andrew Ng, CEO of Landing AI, is known for democratizing AI education through Coursera.",
    },
    {
        "user_input": "Who is Sam Altman?",
        "retrieved_contexts": [
            "Sam Altman, CEO of OpenAI, has advanced AI research and advocates for safe, beneficial AI technologies."
        ],
        "response": "Sam Altman is the CEO of OpenAI and advocates for safe, beneficial AI technologies.",
        "reference": "Sam Altman, CEO of OpenAI, has advanced AI research and advocates for safe AI.",
    },
    {
        "user_input": "Who is Demis Hassabis and how did he gain prominence?",
        "retrieved_contexts": [
            "Demis Hassabis, CEO of DeepMind, is known for developing systems like AlphaGo that master complex games."
        ],
        "response": "Demis Hassabis is the CEO of DeepMind, known for developing systems like AlphaGo.",
        "reference": "Demis Hassabis, CEO of DeepMind, is known for developing AlphaGo.",
    },
    {
        "user_input": "Who is the CEO of Google and Alphabet Inc., praised for leading innovation across Google's product ecosystem?",
        "retrieved_contexts": [
            "Sundar Pichai, CEO of Google and Alphabet Inc., leads innovation across Google's product ecosystem."
        ],
        "response": "Sundar Pichai is the CEO of Google and Alphabet Inc., praised for leading innovation across Google's product ecosystem.",
        "reference": "Sundar Pichai, CEO of Google and Alphabet Inc., leads innovation across Google's product ecosystem.",
    },
    {
        "user_input": "How did Arvind Krishna transform IBM?",
        "retrieved_contexts": [
            "Arvind Krishna, CEO of IBM, transformed the company by focusing on cloud computing and AI solutions."
        ],
        "response": "Arvind Krishna transformed IBM by focusing on cloud computing and AI solutions.",
        "reference": "Arvind Krishna, CEO of IBM, transformed the company through cloud computing and AI.",
    },
]

evaluation_dataset = EvaluationDataset.from_list(dataset)

3. 追踪 ragas 指标

在您的数据集上运行 Ragas 评估，trace 将在您的 LangSmith 控制面板中指定项目名称下或“default”下显示。

from ragas import evaluate
from ragas.llms import LangchainLLMWrapper
from langchain_openai import ChatOpenAI
from ragas.metrics import LLMContextRecall, Faithfulness, FactualCorrectness

llm = ChatOpenAI(model="gpt-4o-mini")
evaluator_llm = LangchainLLMWrapper(llm)

result = evaluate(
    dataset=evaluation_dataset,
    metrics=[LLMContextRecall(), Faithfulness(), FactualCorrectness()],
    llm=evaluator_llm,
)

result

输出

Evaluating:   0%|          | 0/15 [00:00<?, ?it/s]

{'context_recall': 1.0000, 'faithfulness': 0.9333, 'factual_correctness': 0.8520}

4. LangSmith 控制面板

jpeg