Azure 中的流运行管理#

作者：

要求 - 为了从本教程中获益，您需要：

一个拥有活动订阅的 Azure 帐户 - 免费创建帐户
一个 Azure ML 工作区 - 配置工作区
一个 Python 环境
已安装 Prompt flow SDK

学习目标 - 完成本教程后，您应该能够：

使用远程数据创建运行
创建引用另一个运行输入内容的运行
通过 run.yaml 管理运行
创建具有连接覆盖的运行

动机 - 本指南将引导您了解云运行管理功能。

0. 安装依赖包#

%pip install -r ../../requirements.txt

1. 连接到 Azure 机器学习工作区#

工作区是 Azure 机器学习的顶级资源，提供了一个集中位置来处理您在使用 Azure 机器学习时创建的所有工件。本节我们将连接到将运行作业的工作区。

1.1 导入所需的库#

from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
from azure.ai.ml.entities import Data
from azure.core.exceptions import ResourceNotFoundError

from promptflow.azure import PFClient
from promptflow.entities import Run

1.2 配置凭据#

我们使用 DefaultAzureCredential 来获取对工作区的访问权限。DefaultAzureCredential 应该能够处理大多数 Azure SDK 身份验证场景。

如果它对您不起作用，请参考更多可用的凭据：配置凭据示例、azure-identity 参考文档。

try:
    credential = DefaultAzureCredential()
    # Check if given credential can get token successfully.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential in case DefaultAzureCredential not work
    credential = InteractiveBrowserCredential()

1.3 获取工作区的句柄#

我们使用配置文件连接到工作区。Azure ML 工作区应配置有计算群集。查看此笔记本以配置工作区

# Get a handle to workspace
pf = PFClient.from_config(credential=credential)

1.4 创建必要的连接#

连接有助于安全地存储和管理与 LLM 和其他外部工具（例如 Azure 内容安全）交互所需的密钥或其他敏感凭据。

在本笔记本中，我们将使用流 web-classification，它内部使用连接 open_ai_connection。如果我们以前没有添加过，我们需要设置此连接。

遵循此说明准备您的 Azure OpenAI 资源，如果您没有 api_key，请获取一个。

请前往工作区门户，点击 Prompt flow -> Connections -> Create，然后按照说明创建您自己的连接。了解更多关于连接的信息。

2. 使用远程数据创建运行#

在提交流时，除了依赖本地文件外，有时您可能希望重用工作区中已有的数据。以下代码单元展示了如何使用远程数据创建流运行。

2.1 创建或更新远程数据#

data_name, data_version = "flow_run_test_data", "1"

try:
    data = pf.ml_client.data.get(name=data_name, version=data_version)
except ResourceNotFoundError:
    data = Data(
        name=data_name,
        version=data_version,
        path=f"../../flows/standard/web-classification/data.jsonl",
        type="uri_file",
    )
    data = pf.ml_client.data.create_or_update(data)

2.2 准备远程数据 ID#

data_id = f"azureml:{data.name}:{data.version}"
print(data_id)

2.3 使用远程数据创建流运行#

您可以更改运行时的实例类型或空闲时间，或将其重置为干净状态。以下代码单元展示了如何执行此操作。

# create run
run = Run(
    # local flow file
    flow="../../flows/standard/web-classification",
    # remote data
    data=data_id,
    # to customize runtime instance type and compute instance, you can provide them in resources
    # resources={
    #     "instance_type": "STANDARD_DS11_V2",
    #     "compute": "my_compute_instance"
    # }
    # to customize identity, you can provide them in identity
    # identity={
    #     "type": "managed",
    # }
)

base_run = pf.runs.create_or_update(run=run)

2.4 流式传输流运行以确保其成功运行#

pf.runs.stream(base_run)

3 创建一个使用现有运行输入的流运行#

当运行带有现有运行的流时，您可以在列映射中引用其输入或输出。以下代码单元展示了如何在列映射中引用运行的输入。

run = Run(
    # local flow file
    flow="../../flows/standard/web-classification",
    # run name
    run=run,
    column_mapping={
        # reference another run's input data columns
        "url": "${run.inputs.url}",
        "answer": "${run.inputs.answer}",
        "evidence": "${run.inputs.evidence}",
    },
)

base_run = pf.runs.create_or_update(
    run=run,
)

pf.runs.stream(base_run)

4. 创建具有连接覆盖的流运行#

有时您希望在提交流时切换流中的连接或部署名称。连接覆盖提供了一种简单的方法来完成此操作，而无需更改原始 flow.dag.yaml。在以下代码单元中，我们将提交流 web-classification 并将其连接 open_ai_connection 覆盖为 azure_open_ai_connection。请确保连接 azure_open_ai_connection 存在于您的工作区中。

run = Run(
    # local flow file
    flow="../../flows/standard/web-classification",
    data="../../flows/standard/web-classification/data.jsonl",
    # override connection for node classify_with_llm & summarize_text_content
    connections={
        "classify_with_llm": {"connection": "azure_open_ai_connection"},
        "summarize_text_content": {"connection": "azure_open_ai_connection"},
    },
)

base_run = pf.runs.create_or_update(
    run=run,
)

pf.runs.stream(base_run)