promptflow.contracts.run_info 模块#

class promptflow.contracts.run_info.FlowRunInfo(run_id: str, status: Status, error: object, inputs: object, output: object, metrics: Dict[str, Any], request: object, parent_run_id: str, root_run_id: str, source_run_id: str, flow_id: str, start_time: datetime, end_time: datetime, index: Optional[int] = None, api_calls: Optional[List[Dict[str, Any]]] = None, name: str = '', description: str = '', tags: Optional[Mapping[str, str]] = None, system_metrics: Optional[Dict[str, Any]] = None, result: Optional[object] = None, upload_metrics: bool = False, otel_trace_id: Optional[str] = '', message_format: str = 'basic')#

基类：object

表示运行信息的数据类。

参数:

  • run_id (str) – 运行 ID，等于 flow_run_id:child_flow_run_id

  • status (Status) – 流运行状态

  • error (Dict[str, Any]) – 流运行期间发生的错误

  • inputs (object) – 流运行的输入

  • output (object) – 流运行的输出

  • metrics (Dict[str, Any]) – 流运行的指标

  • request (object) – 流运行的请求

  • parent_run_id (str) – 流运行的父运行 ID

  • root_run_id (str) – 流运行的根运行 ID

  • source_run_id (str) – 触发流运行的运行 ID

  • flow_id (str) – 流运行的流 ID

  • start_time (datetime) – 流运行的开始时间

  • end_time (datetime) – 流运行的结束时间

  • index (Optional[int]) – 流运行的索引（用于批量测试模式）

  • api_calls (Optional[List[Dict[str, Any]]]) – 流运行期间的 API 调用

  • name (Optional[str]) – 流运行的名称

  • description (Optional[str]) – 流运行的描述

  • tags (Optional[Dict[str, str]]) – 流运行的标签

  • system_metrics (Optional[Dict[str, Any]]) – 流运行的系统指标

  • result (Optional[object]) – 流运行的结果

  • upload_metrics (Optional[bool]) – 指示是否上传流运行指标的标志

  • message_format (str) – 表示不同多媒体契约的消息格式类型。

api_calls: Optional[List[Dict[str, Any]]] = None#
static create_with_error(start_time, inputs, index, run_id, error)#
description: str = ''#
static deserialize(data: dict) → FlowRunInfo#

从字典中反序列化 FlowRunInfo。

end_time: datetime#
error: object#
flow_id: str#
index: Optional[int] = None#
inputs: object#
message_format: str = 'basic'#
metrics: Dict[str, Any]#
name: str = ''#
otel_trace_id: Optional[str] = ''#
output: object#
parent_run_id: str#
request: object#
result: object = None#
root_run_id: str#
run_id: str#
source_run_id: str#
start_time: datetime#
status: Status#
system_metrics: Dict[str, Any] = None#
tags: Optional[Mapping[str, str]] = None#
upload_metrics: bool = False#
class promptflow.contracts.run_info.RunInfo(node: str, flow_run_id: str, run_id: str, status: Status, inputs: Mapping[str, Any], output: object, metrics: Dict[str, Any], error: Dict[str, Any], parent_run_id: str, start_time: datetime, end_time: datetime, index: Optional[int] = None, api_calls: Optional[List[Dict[str, Any]]] = None, cached_run_id: Optional[str] = None, cached_flow_run_id: Optional[str] = None, logs: Optional[Dict[str, str]] = None, system_metrics: Optional[Dict[str, Any]] = None, result: Optional[object] = None, message_format: str = 'basic')#

基类：object

表示运行信息的数据类。

参数:

  • node (str) – 节点名称

  • flow_run_id (str) – 流运行的 ID

  • run_id (str) – 运行 ID，等于 flow_run_id:step_run_id

  • status (Status) – 运行状态

  • inputs (list) – 运行的输入列表

  • output (object) – 运行的输出

  • metrics (Dict[str, Any]) – 运行的指标

  • error (Dict[str, Any]) – 运行期间发生的错误

  • parent_run_id (str) – 父运行 ID

  • start_time (datetime) – 运行的开始时间

  • end_time (datetime) – 运行的结束时间

  • index (Optional[int]) – 运行的索引

  • api_calls (Optional[List[Dict[str, Any]]]) – 运行期间的 API 调用

  • cached_run_id (Optional[str]) – 缓存的运行 ID

  • cached_flow_run_id (Optional[str]) – 缓存的流运行 ID

  • logs (Optional[Dict[str, str]]) – 运行日志

  • system_metrics (Optional[Dict[str, Any]]) – 运行的系统指标

  • result (Optional[object]) – 运行结果

  • message_format (str) – 表示不同多媒体契约的消息格式类型。

api_calls: Optional[List[Dict[str, Any]]] = None#
cached_flow_run_id: str = None#
cached_run_id: str = None#
static deserialize(data: dict) → RunInfo#

从字典中反序列化 RunInfo。

end_time: datetime#
error: Dict[str, Any]#
flow_run_id: str#
index: Optional[int] = None#
inputs: Mapping[str, Any]#
logs: Optional[Dict[str, str]] = None#
message_format: str = 'basic'#
metrics: Dict[str, Any]#
node: str#
output: object#
parent_run_id: str#
result: object = None#
run_id: str#
start_time: datetime#
status: Status#
system_metrics: Dict[str, Any] = None#
class promptflow.contracts.run_info.Status(value)#

基类: Enum

不同运行状态的枚举类。

Bypassed = 'Bypassed'#
CancelRequested = 'CancelRequested'#
Canceled = 'Canceled'#
Completed = 'Completed'#
Failed = 'Failed'#
NotStarted = 'NotStarted'#
Preparing = 'Preparing'#
Running = 'Running'#
static is_terminated(status)#

检查给定状态是否已终止。

参数:

status (str or Status) – 要检查的状态

返回:

如果状态已终止，则为 True，否则为 False

返回类型:

bool