评估 AI 应用程序

要评估生成式 AI 应用程序,请遵循以下步骤:

  1. 使用 TrulensSDK(支持使用 Python 构建的应用程序)构建应用程序并对其进行检测。

  2. 在 Snowflake 中注册应用程序。

  3. 通过指定输入数据集来创建运行。

  4. 执行运行以生成跟踪并计算评估指标。

  5. 在 Snowsight 中查看评估结果。

对应用程序进行检测

在 Python 中创建生成式 AI 应用程序后,导入 TruLens SDK 以对其进行检测。TruLens SDK 提供了一个 @instrument() 装饰器,用于检测应用程序中的函数,以生成跟踪并计算指标。

  • 要使用装饰器,请在 Python 应用程序中添加以下导入:

    from trulens.core.otel.instrument import instrument
    

您可以根据需求更改 @instrument() 装饰器的粒度。

场景 1:跟踪函数

您可以在需要跟踪的函数前添加 @instrument()。这将自动捕获函数的输入、输出(返回值)和执行延迟。例如,以下代码演示了如何跟踪 answer_query 函数,该函数可自动捕获输入查询和最终响应:

@instrument()
def answer_query(self, query: str) -> str:
    context_str = self.retrieve_context(query)
    return self.generate_completion(query, context_str)

场景 2:使用特定 span 类型跟踪函数

A span type specifies the nature of the function and improves the readability and understanding of the traces. For example, in a RAG application you can specify span type as RETRIEVAL for your search service (or retriever) and specify the span type as GENERATION for the LLM inference call. The following span types are supported:

  • RETRIEVAL: Span type for retrieval or search functions

  • GENERATION: Span type for model inference calls from an LLM

  • RECORD_ROOT: Span type for the main function in your application

If you don’t specify a span type with the @instrument(), an UNKNOWN span type is assigned by default. To use span attributes, add the following import to your Python application.

from trulens.otel.semconv.trace import SpanAttributes

以下代码片段演示了如何跟踪 RAG 应用程序。span 类型必须始终将 SpanAttributes.SpanType 作为前缀。

@instrument(span_type=SpanAttributes.SpanType.RETRIEVAL)
def retrieve_context(self, query: str) -> list:
    """
    Retrieve relevant text from vector store.
    """
    return self.retrieve(query)

@instrument(span_type=SpanAttributes.SpanType.GENERATION)
def generate_completion(self, query: str, context_str: list) -> str:
    """
    Generate answer from context by calling an LLM.
    """
    return response

@instrument(span_type=SpanAttributes.SpanType.RECORD_ROOT)
def answer_query(self, query: str) -> str:
    context_str = self.retrieve_context(query)
    return self.generate_completion(query, context_str)

场景 3:跟踪函数并计算求值

除了提供 span 类型外,您还必须将应用程序中的相关参数分配给 span 属性,以便计算指标。例如,要在 RAG 应用程序中计算上下文相关性,必须将相关查询和检索结果参数分别分配给相应的属性 RETRIEVAL.QUERY_TEXT 和``RETRIEVAL.RETRIEVED_CONTEXTS``。计算每个指标所需的属性可在“Metrics”页面找到。

每种 span 类型都支持以下 span 属性:

  • RECORD_ROOT: INPUT, OUTPUT, GROUND_TRUTH_OUTPUT

  • RETRIEVAL: QUERY_TEXT, RETRIEVED_CONTEXTS

  • GENERATION: None

要使用 span 属性,您需要在 Python 应用程序中添加以下导入。

from trulens.otel.semconv.trace import SpanAttributes

下面的代码片段提供了为检索服务计算上下文相关性的示例。属性必须始终遵循以下格式:SpanAttributes.<span type>.<attribute name>``(例如,``SpanAttributes.RETRIEVAL.QUERY_TEXT)。

@instrument(
    span_type=SpanAttributes.SpanType.RETRIEVAL,
    attributes={
        SpanAttributes.RETRIEVAL.QUERY_TEXT: "query",
        SpanAttributes.RETRIEVAL.RETRIEVED_CONTEXTS: "return",
    }
)
def retrieve_context(self, query: str) -> list:
    """
    Retrieve relevant text from vector store.
    """
    return self.retrieve(query)

在上述示例中,query 表示 retrieve_context() 的输入参数,而 return 表示返回的值。将其分配给属性 RETRIEVAL.QUERY_TEXTRETRIEVAL.RETRIEVED_CONTEXTS,以计算上下文相关性。

Auto-instrument framework applications

In addition to manual instrumentation using the @instrument() decorator, TruLens provides specialized wrappers that automatically instrument applications built with popular LLM frameworks. These wrappers provide integration and automatic tracing without requiring manual decoration of individual functions.

TruChain for LangChain

TruChain provides automatic instrumentation for applications built with LangChain (https://www.langchain.com/). It automatically captures the execution of key LangChain classes including chains, LLMs, prompts, and retrievers.

from trulens.apps.langchain import TruChain

# Wrap your LangChain application
tru_recorder = TruChain(
    rag_chain,
    app_name="my_langchain_app",
    app_version="v1.0"
)

# Use the recorder as a context manager
with tru_recorder as recording:
    response = rag_chain.invoke(input_query)

TruChain supports:

  • Automatic instrumentation of LangChain Expression Language (LCEL) chains

  • Async support through the ainvoke method

  • Built-in selectors (on_input, on_output, on_context) for RAG triad evaluation

TruGraph for LangGraph

TruGraph provides automatic instrumentation for applications built with LangGraph (https://langchain-ai.github.io/langgraph/). It automatically detects LangGraph applications and instruments both LangChain and LangGraph components.

from trulens.apps.langgraph import TruGraph

# Wrap your LangGraph application
tru_recorder = TruGraph(
    graph,
    app_name="my_langgraph_app",
    app_version="v1.0"
)

# Use the recorder as a context manager
with tru_recorder as recording:
    response = graph.invoke({"messages": [("user", input_query)]})

TruGraph supports:

  • Automatic @task instrumentation with intelligent attribute extraction

  • Multi-agent evaluation capabilities

  • Combined instrumentation of both LangChain and LangGraph components

TruLlama for LlamaIndex

TruLlama provides automatic instrumentation for applications built with LlamaIndex (https://www.llamaindex.ai/). It automatically captures the execution of key LlamaIndex classes including query engines, retrievers, and response synthesizers.

from trulens.apps.llamaindex import TruLlama

# Wrap your LlamaIndex query engine
tru_recorder = TruLlama(
    query_engine,
    app_name="my_llamaindex_app",
    app_version="v1.0"
)

# Use the recorder as a context manager
with tru_recorder as recording:
    response = query_engine.query(input_query)

TruLlama supports:

  • Automatic instrumentation of query engines, chat engines, and retrievers

  • Async support through aquery, achat, and astream_chat methods

  • Streaming support for LlamaIndex applications

  • Built-in selectors (on_input, on_output, on_context) for RAG triad evaluation

For more information about framework-specific instrumentation, see the TruLens documentation (https://www.trulens.org/component_guides/instrumentation/).

在 Snowflake 中注册应用程序

To register your generative AI application in Snowflake for capturing traces and conducting evaluations, you need to create a TruApp object using the TruLens SDK that records the invocation (execution) of the user's app and exports traces to Snowflake.

tru_app = TruApp(
    app: Any,
    app_name: str,
    app_version: str,
    connector: SnowflakeConnector,
    main_method: callable  # i.e. app.query
)

备注

If your application is built using LangChain, LangGraph, or LlamaIndex, you can use TruChain, TruGraph, or TruLlama respectively in place of TruApp. These framework-specific wrappers provide the same registration functionality while also enabling automatic instrumentation of your application. See Auto-instrument framework applications for more details.

参数:

  • app: Any: an instance of the user-defined application that will later be invoked during a run for evaluation. i.e. app = RAG()

  • app_name: str:是用户可指定的应用程序名称,并将保存在用户的 Snowflake 账户中。

  • app_version: str: is the version user can specify for the app to allow experiments tracking and comparison.

  • connector: SnowflakeConnector:包装程序类,用于管理 Snowpark 会话和 Snowflake DB 连接。

  • main_method: callable (Optional): is the entry point method for the user’s application, which tells the SDK how the app is expected to be called by users and where to start tracing the invocation of the user app (specified by app). For the example of RAG class, the main_method can be specified as app.answer_query, assuming the answer method is the entry point of the app. Alternatively, instrument the entry point method with span attribute RECORD_ROOT. In that case, this parameter is not required.

创建运行

要开始一项评估作业,您需要创建一个运行。创建运行需要指定运行配置。add_run() 函数使用运行配置来创建新的运行。

运行配置

通过 RunConfig 创建运行

run_config = RunConfig(
    run_name=run_name,
    description="desc",
    label="custom tag useful for grouping comparable runs",
    source_type="DATAFRAME",
    dataset_name="My test dataframe name",
    dataset_spec={
        "RETRIEVAL.QUERY_TEXT": "user_query_field",
        "RECORD_ROOT.INPUT": "user_query_field",
        "RECORD_ROOT.GROUND_TRUTH_OUTPUT": "golden_answer_field",
    },
    llm_judge_name: "mistral-large2"
)
  • run_name: str:运行的名称在同一 TruApp 下应该唯一

  • ``description: str``(可选):运行的字符串描述

  • ``label: str``(可选):用于将运行分组的标签

  • source_type: str:指定数据集的来源。可以是 ``DATAFRAME``(适用于 python 数据框),也可以是 ``TABLE``(适用于 Snowflake 账户中的用户表)。

  • dataset_name: str:如果 source_type 为 DATAFRAME,用户指定的任意名称。或者,当前上下文(数据库和架构)的用户账户下的有效 Snowflake 表名,或者“database.schema.table_name”形式的 Snowflake 完全限定名称。

  • dataset_spec: Dict[str, str]:字典,可将支持的 span 属性映射到数据框或表中的用户列名。允许的键是“Dataset”页面中指定的 span 属性,允许的值是用户指定的数据框或表中的列名。例如,上述运行配置示例中的“golden_answer_field”必须是有效的列名

  • llm_judge_name: str``(可选):在基于 LLM 的指标计算过程中用作 LLM 裁判的名称。有关支持的裁判,请参阅“Models”页面。如果未指定,则默认值为 ``llama3.1-70b

run = tru_app.add_run(run_config=run_config)

请求参数:

  • run_config: RunConfig:包含运行配置。

检索运行

检索相应运行。

run = tru_app.get_run(run_name=run_name)

请求参数:

  • run_name: str:运行名称

查看运行元数据

描述运行详细信息。

run.describe()

调用运行

您可以使用 run.start() 函数调用运行。它从运行配置中指定的数据集读取输入,调用每个输入的应用程序,生成跟踪,并引入信息以存储在 Snowflake 账户中。run.start() 是一个阻塞调用,直到对数据集中的所有输入调用应用程序,并且引入完成或超时为止。

run.start()  # if source_type is "TABLE"

run.start(input_df=user_input_df)  # if source_type is "DATAFRAME"

请求参数:

  • input_df: DataFrame``(可选):是 SDK 中的 Pandas 数据框。如果将运行配置中的 source_type 指定为 ``DATAFRAME,则该字段为必填字段。如果 source_type 为 TABLE,则该字段不是必填字段。

计算指标

您可以在调用应用程序并引入所有跟踪后,使用 run.compute_metrics() 启动指标计算。只要运行状态为 INVOCATION_IN_PROGRESS,就无法启动计算。一旦状态为 INVOCATION_COMPLETEDINVOCATION_PARTIALLY_COMPLETED,即可启动 run.compute_metrics()run.compute_metrics() 是一个异步非阻塞函数。您可以在同一运行中使用不同的指标集多次调用 compute_metrics,每次调用都会触发一个新的计算作业。请注意,一旦计算出指标,就无法在同一次运行中再次计算。

run.compute_metrics(metrics=[
    "coherence",
    "answer_relevance",
    "groundedness",
    "context_relevance",
    "correctness",
])

请求参数:

  • metrics: List[str]:“指标”中列出的指标的字符串名称列表。指标名称应使用蛇形命名法来指定。例如,Context Relevance 应指定为 context_relevance

检查运行状态

您可以在运行开始后检查其状态。状态列表位于“运行状态”部分。

run.get_status()

取消运行

您可以使用 run.cancel() 取消现有运行。此操作将阻止今后对运行的任何更新,包括运行状态和元数据字段。

run.cancel()

删除运行

您可以使用 run.delete() 删除现有运行。此操作会删除与运行相关的元数据,且无法访问评估结果。不过,运行过程中生成的跟踪和评估结果不会删除,仍会保存。请参阅“可观察性数据”部分,了解有关存储和删除评估与跟踪的更多信息。

run.delete()

应用程序的运行列表

您可以使用 list_runs() 函数查看与特定 TruApp 应用程序对象对应的所有可用运行列表。

tru_app.list_runs()

响应:

返回 tru_app 下创建的所有运行的列表。

查看评估和跟踪

要查看评估结果,请执行以下操作:

  1. 登录 Snowsight

  2. 在导航菜单中,选择 AI & ML » Evaluations

执行以下操作以查看应用程序运行的评估结果:

  • 要查看与特定应用程序对应的运行,请选择该应用程序。

  • 要查看某个运行的评估结果,请选择该运行。您可以查看汇总结果和每条记录对应的结果。

  • 要查看某条记录的跟踪,请选择该记录。您可以查看应用程序每个暂存区的详细跟踪、延迟、输入和输出、评估结果,以及 LLM 裁判对已生成的准确度分数所做的说明。

要比较使用相同数据集的运行,请选择多个运行,然后选择 Compare 来比较输出和评估分数。