为流部署启用跟踪并收集反馈（预览版）

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

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

先决条件

• Azure CLI 和安装到 Azure CLI 的 Azure 机器学习扩展。 有关详细信息，请参阅安装、设置和使用 CLI (v2)。
• Azure 机器学习工作区。 如果没有，请按照快速入门：创建工作区资源一文中的步骤创建一个。
• Application Insights。 通常，机器学习工作区具有默认链接的 Application Insights。 若要使用新资源，可以创建 Application Insights 资源。
• 对托管联机终结点有基本的了解。 托管联机终结点以可缩放的完全托管方式与 Azure 中功能强大的 CPU 和 GPU 计算机配合使用，使你无需设置和管理底层部署基础结构的开销。 有关托管联机终结点的详细信息，请参阅用于实现实时推理的联机终结点和部署。
• Azure 基于角色的访问控制 (Azure RBAC) 用于授予对 Azure 机器学习中的操作的访问权限。 要执行本文中的步骤，必须为用户帐户分配 Azure 机器学习工作区的所有者或参与者角色，或者分配一个允许“Microsoft.MachineLearningServices/workspaces/onlineEndpoints/”的自定义角色。 如果使用工作室创建/管理联机终结点/部署，则需要资源组所有者提供的另一个权限“Microsoft.Resources/deployments/write”。 有关详细信息，请参阅管理对 Azure 机器学习工作区的访问。

部署实时推理流

正确测试流（无论是 Flex 流还是 DAG 流）后，可以在生产环境中部署该流。 在本文中，我们以将流部署到 Azure 机器学习托管联机终结点为例。 对于 flex 流，需要准备 flow.flex.yaml 文件而不是 flow.dag.yaml。

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

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

如果使用工作室 UI 进行部署，则可以在“高级设置”->“部署向导中的部署步骤”中启用”Application Insights 诊断”，从而将跟踪数据和系统指标收集到工作区链接的 Application Insights。

如果使用 SDK 或 CLI，则可以在部署 yaml 文件中添加一个 app_insights_enabled: true 属性，用于将数据收集到工作区链接的 Application Insights。 还可以通过部署 yaml 文件中的环境变量 APPLICATIONINSIGHTS_CONNECTION_STRING 指定其他 Application Insights，如下所示。 可以在 Azure 门户的“概述”页面中找到 Application Insights 的连接字符串。

# below is the property in deployment yaml
# app_insights_enabled: true

# you can also use the environment variable
environment_variables:
  APPLICATIONINSIGHTS_CONNECTION_STRING: <connection_string>

在 Application Insights 中查看跟踪数据

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

可以在指定的 Application Insights 中查看详细跟踪。 以下屏幕截图显示了包含多个节点的已部署流的事件示例。 在“Application Insights”->“调查”->“事务搜索”中，可以选择每个节点以查看其详细跟踪。

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

在 Application Insights 中查看系统指标

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

打开 Application Insights，然后从左侧导航中选择“使用情况和预估成本”。 选择“自定义指标(预览版)”，然后选择“使用维度”并保存更改。

在左侧导航中选择“指标”选项卡。 从“指标命名空间”中选择“promptflow 标准指标”，然后可以使用不同的聚合方法从“指标”下拉列表中浏览指标。

收集反馈并发送到 Application Insights

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

以下示例代码演示如何对已部署的托管终结点启用跟踪的流进行评分，并将反馈发送到评分请求的同一跟踪范围。 该流具有输入 question 和 chat_hisotry，以及输出 answer。 为终结点评分后，我们将收集反馈并将其发送到部署流时指定的 Application Insights。 你需要根据自己的用例填写 api_key 值或修改代码。

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 may be updated
# depending on the format your endpoint expects.
# More information can be found here:
# https://docs.microsoft.com/azure/machine-learning/how-to-deploy-advanced-entry-script
data = {
    "question": "hello",
    "chat_history": []
}

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

url = 'https://basic-chat-endpoint.eastus.inference.ml.azure.com/score'
feedback_url = 'https://basic-chat-endpoint.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 收集器服务

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

后续步骤

• 排查托管联机终结点错误。
• 将流部署到其他平台（例如 Docker）。