为流部署启用跟踪并收集反馈

在生产环境中部署生成 AI 应用程序后，可能需要增强理解和优化性能。 每个请求的跟踪数据、聚合指标和用户反馈发挥着关键作用。

本文介绍如何在流部署的推理时间启用跟踪、收集聚合指标和收集用户反馈。

先决条件

• 一份 Azure 订阅。 如果没有 Azure 订阅，可以创建一个免费帐户。
• 如果没有项目， 请创建基于中心的项目。

• Azure CLI 和安装到 Azure CLI 的 Azure 机器学习扩展。
• 一个 Azure AI Foundry 项目。 如果还没有项目，可以 创建一个项目。
• 应用程序信息 (Application Insights) 资源。 如果还没有 Application Insights 资源，可以 创建一个。
• Azure 基于角色的访问控制用于授予对 Azure 机器学习中的操作的访问权限。 若要执行本文中的步骤，必须具有所选资源组的“所有者”或“参与者”权限。 有关详细信息，请参阅 Azure AI Foundry 门户中基于角色的访问控制。

部署实时推理工作流

在对您的流（无论是弹性流还是 DAG 流）进行适当测试后，您可以将该流部署到生产环境中。 在本文中，我们使用 部署流进行实时推理 作为示例。 对于 flex 流，需要准备 flow.flex.yaml 文件而不是 flow.dag.yaml。

还可以 部署到其他平台，例如 Docker 容器和 Kubernetes 群集。

使用最新的提示流基础映像来部署流，以支持跟踪和反馈收集 API。

为部署启用跟踪并收集系统指标

如果使用 Azure AI Foundry 门户进行部署，请在部署向导中选择 “部署>Application Insights 诊断>高级设置 ”。 这样，跟踪数据和系统指标将被收集到一个与 Application Insights 连接的项目。

如果使用 SDK 或 CLI，请在部署 .yaml 文件中添加 app_insights_enabled: true 该属性，该文件收集数据到链接到 Application Insights 的项目。

app_insights_enabled: true

还可以通过部署 .yaml 文件中的环境变量 APPLICATIONINSIGHTS_CONNECTION_STRING 指定其他应用程序见解。 可以在 Azure 门户中的 “概述 ”页上找到 Application Insights 的连接字符串。

environment_variables:
  APPLICATIONINSIGHTS_CONNECTION_STRING: <connection_string>

在 Application Insights 中查看跟踪数据

跟踪记录应用程序在执行期间的特定事件或状态。 它可以包括有关函数调用、变量值和系统事件的数据。 跟踪有助于将应用程序的组件分解为离散输入和输出。 此过程对于调试和理解应用程序至关重要。 若要了解有关跟踪的详细信息，请参阅 此网站。 跟踪数据遵循 OpenTelemetry 规范。

您可以在您指定的应用程序洞察中查看详细跟踪。 以下屏幕截图显示了包含多个节点的已部署流的事件示例。 选择 Application Insights>调查>事务搜索，然后选择每个节点以查看其详细跟踪。

“依赖项”类型事件记录来自部署的调用。 事件的名称是流文件夹的名称。 若要了解详细信息，请参阅 Application Insights 中的事务搜索和诊断。

在 Application Insights 中查看系统指标

可以在 Azure 门户中的工作区概述页上找到工作区默认 Application Insights 指标。

1. 打开 Application Insights，然后在左窗格中选择 “使用情况”和“估计成本 ”。 选择“自定义指标（预览）”>“具有维度”，并保存更改。
2. 选择左窗格中的“ 指标 ”选项卡。 从 指标命名空间中，选择 promptflow 标准指标。 可以使用不同的聚合方法浏览 指标 下拉列表中的指标。

收集反馈并发送到 Application Insights

提示流服务提供了一个新的 /feedback API 来帮助客户收集反馈。 反馈有效负载可以是任何 JSON 格式数据。 提示流服务可帮助客户将反馈数据保存到跟踪范围。 数据将保存到客户配置的跟踪导出程序目标。 提示流服务还支持 OpenTelemetry 标准跟踪上下文传播。 它遵循请求头中设置的跟踪上下文，并使用该上下文作为请求父范围上下文。 可以使用分布式跟踪功能将反馈跟踪与其聊天请求跟踪相关联。

以下示例代码演示如何为部署到启用跟踪的托管终结点的流进行评分，并将反馈发送到评分请求的对应跟踪范围。 流具有输入 question 和 chat_history。 输出为 answer。 对终结点进行评分后，将收集反馈，并在部署流时按指定发送到 Application Insights。

import urllib.request
import json
import os
import ssl
from opentelemetry import trace, context
from opentelemetry.baggage.propagation import W3CBaggagePropagator
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.sdk.trace import TracerProvider

# Initialize your tracer.
tracer = trace.get_tracer("my.genai.tracer")
trace.set_tracer_provider(TracerProvider())

# Request data goes here.
# The example below assumes JSON formatting, which might be updated
# depending on the format your endpoint expects.
data = {
    "question": "hello",
    "chat_history": []
}

body = str.encode(json.dumps(data))

url = 'https://basic-chat-endpoint-0506.eastus.inference.ml.azure.com/score'
feedback_url = 'https://basic-chat-endpoint-0506.eastus.inference.ml.azure.com/feedback'
# Replace this with the primary/secondary key, AMLToken, or Microsoft Entra ID token for the endpoint.
api_key = ''
if not api_key:
    raise Exception("A key should be provided to invoke the endpoint")

# The azureml-model-deployment header will force the request to go to a specific deployment.
# Remove this header to have the request observe the endpoint traffic rules.
headers = {'Content-Type':'application/json', 'Authorization':('Bearer '+ api_key), 'azureml-model-deployment': 'basic-chat-deployment' }

try:
    with tracer.start_as_current_span('genai-request') as span:

        ctx = context.get_current()
        TraceContextTextMapPropagator().inject(headers, ctx)
        print(headers)
        print(ctx)
        req = urllib.request.Request(url, body, headers)
        response = urllib.request.urlopen(req)

        result = response.read()
        print(result)

        # Now you can process the answer and collect feedback.
        feedback = "thumbdown"  # Example feedback (modify as needed).

        # Make another request to save the feedback.
        feedback_body = str.encode(json.dumps(feedback))
        feedback_req = urllib.request.Request(feedback_url, feedback_body, headers)
        urllib.request.urlopen(feedback_req)


except urllib.error.HTTPError as error:
    print("The request failed with status code: " + str(error.code))

    # Print the headers - they include the requert ID and the timestamp, which are useful for debugging the failure.
    print(error.info())
    print(error.read().decode("utf8", 'ignore'))

可以在 Application Insights 中查看请求的跟踪以及反馈。

高级用法：将跟踪导出到自定义的 OpenTelemetry 收集器服务

在某些情况下，可能需要将跟踪数据导出到已部署的 OpenTelemetry 收集器服务。 若要启用此服务，请设置 OTEL_EXPORTER_OTLP_ENDPOINT。 当你想要自定义自己的范围处理逻辑和自己的跟踪持久目标时，请使用此导出程序。

相关内容

• 使用提示流 SDK 开始构建聊天应用
• 使用 Visual Studio Code 中的项目