为基于 LLM 的应用程序集成提示流与 DevOps

若要从工作目录中触发运行，请运行以下代码：

pf flow test --flow <directory-name>

from promptflow import PFClient

pf_client = PFClient()

flow_path = "<directory-name>"

# Test flow
flow_inputs = {"<input-type>": "<input-value>", "<input-type>": "<input-value>"}
flow_result = pf_client.test(flow=flow_path, inputs=inputs)
print(f"Flow outputs: {flow_result}")

# Test node in the flow
node_name = "<node-name>"  # The node name in the flow.
node_inputs = {"<node-input-type>": "<node-input-value>"}
node_result = pf_client.test(flow=flow_path, inputs=node_inputs, node=node_name)
print(f"Node outputs: {node_result}")

az login

导入所需的库和包、配置凭据并获取工作区的句柄。

import json

# Import required libraries
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential

# Import azure promptflow apis
from promptflow.azure import PFClient

# Configure credential
try:
    credential = DefaultAzureCredential()
    # Check if given credential can get token successfully.
    credential.get_token("https://management.chinacloudapi.cn/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential if DefaultAzureCredential doesn't work
    credential = InteractiveBrowserCredential()

# Get a handle to the workspace from the current credential or config.json in the parent directory
pf = PFClient.from_config(
    credential=credential,
)

准备 run.yml 文件，以在云中定义此流运行的配置。

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Run.schema.json
flow: <path-to-flow>
data: <path-to-flow>/<data-file>.jsonl

column_mapping:
  url: ${data.url}

# Define cloud compute resource

resources:
  instance_type: <compute-type>

# If using compute instance compute type, also specify instance name
#  compute: <compute-instance-name> 

# Specify connections

  <node-name>:
    connection: <connection-name>
    deployment_name: <deployment-name>

可为流中需要连接的每个工具指定连接和部署名称。 如果未指定连接和部署名称，该工具使用 flow.dag.yaml 文件中的连接和部署。 使用以下代码设置连接格式：

...
connections:
  <node-name>:
    connection: <connection-name>
      deployment_name: <deployment-name>
...

创建运行。

pfazure run create --file run.yml

加载流、定义资源和连接并创建运行。

flow = "<path-to-flow>"
data = "<path-to-flow>/<data-file>.jsonl"

# Define compute resource instance type when using serverless compute

# resources = {"instance_type": "serverless"}

# Also specify compute instance name when using a compute instance

# resources={
#     "instance_type": "<compute-instance-type>",
#     "compute": "<compute-instance-name>"
# }

# Specify the connection and deployment name for each tool in the flow that requires a connection

connections = {"<node-name>":
                  {"connection": <connection-name>,
                  "deployment_name": <deployment-name>},
               "<node-name>":
                  {"connection": <connection-name>,
                  "deployment_name": <deployment-name>}
                }

# Create the run

run = Run(
    flow=flow,
    data=data,
    column_mapping={
        "url": "${data.url}"
    }, 

    connections=connections,
    # To customize identity, you can provide it in identity

    # identity={
    #     "type": "managed",
    # }
)

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

准备 run_evaluation.yml 文件，以在云中定义此评估流运行的配置。

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Run.schema.json
flow: <path-to-flow>
data: <path-to-flow>/<data-file>.jsonl
run: <id-of-base-flow-run>
column_mapping:
  <input-name>: ${data.<column-from-test-dataset>}
  <input-name>: ${run.outputs.<column-from-run-output>}

resources:
  instance_type: <compute-type>
  compute: <compute_instance_name> 

connections:
  <node-name>:
    connection: <connection-name>
    deployment_name: <deployment-name>
  <node-name>:
    connection: <connection-name>
    deployment_name: <deployment-name>

创建评估运行。

pfazure run create --file run_evaluation.yml

加载评估流并创建运行。

flow = "<path-to-flow>"
data = "<path-to-flow>/<data-file>.jsonl"

resources={
    "instance_type": "<compute-instance-type>",
    "compute": "<compute-instance-name>"
    }

connections = {"<node-name>":
                  {"connection": <connection-name>,
                  "deployment_name": <deployment-name>},
               "<node-name>":
                  {"connection": <connection-name>,
                  "deployment_name": <deployment-name>}
                }
eval_run = Run(
    flow=flow,
    data=data,
    run=<base-run-id>,
    column_mapping={
        "<input-name>": "${data.<column-from-test-dataset>}",
        "<input-name>": "${run.outputs.<column-from-run-output>}",
    },
    connections=connections,
    identity={
        "type": "managed",
    }
)

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

pfazure run stream --name <run-name>

pf.stream("<run-name>")

pfazure run show-details --name <run-name>

details = pf.get_details("<run-name>")
details.head(10)

pfazure run show-metrics --name <evaluation-run-name>

pf.get_metrics("<evaluation-run-name>")