promptflow.contracts.flow 模块#

class promptflow.contracts.flow.ActivateCondition(condition: InputAssignment, condition_value: Any)#

基础类: object

这个类表示一个节点的激活条件。

Parameters:

  • condition (InputAssignment) – 激活条件的条件。

  • condition_value (Any) – 条件的值。

condition: InputAssignment#
condition_value: Any#
static deserialize(data: dict, node_name: Optional[str] = None) → ActivateCondition#

从字典中反序列化激活条件。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的激活条件。

Return type:

ActivateCondition

class promptflow.contracts.flow.FlexFlow(id: str, name: str, inputs: Dict[str, FlowInputDefinition], outputs: Dict[str, FlowOutputDefinition], init: Optional[Dict[str, FlowInputDefinition]] = None, program_language: str = 'python', environment_variables: Optional[Dict[str, object]] = None, message_format: str = 'basic', sample: Optional[Dict[str, dict]] = None)#

基础类：FlowBase

这个类表示一个弹性流。

Parameters:

  • id (str) – 流程的id。

  • name (str) – 流程的名称。

  • inputs (Dict[str, FlowInputDefinition]) – 流程的输入。

  • outputs (Dict[str, FlowOutputDefinition]) – 流程的输出。

  • program_language (str) – 流程的编程语言。

  • environment_variables (Dict[str, object]) – 流程的默认环境变量。

  • message_format (str) – 用于表示不同多媒体合同的流的消息格式类型。

  • sample (Dict[str, object]) – 流程的样本数据。如果未提供，将成为默认输入和初始化参数。

static deserialize(data: dict) → FlexFlow#

从字典中反序列化流程。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的流程。

Return type:

EagerFlow

environment_variables: Dict[str, object] = None#
get_connection_names(environment_variables_overrides: Optional[Dict[str, str]] = None)#

返回连接名称。

init: Dict[str, FlowInputDefinition] = None#
message_format: str = 'basic'#
program_language: str = 'python'#
sample: Dict[str, dict] = None#
class promptflow.contracts.flow.Flow(id: str, name: str, inputs: Dict[str, FlowInputDefinition], outputs: Dict[str, FlowOutputDefinition], nodes: List[节点], tools: List[工具], node_variants: Optional[Dict[str, 节点变体]] = None, program_language: str = 'python', environment_variables: Optional[Dict[str, object]] = None, message_format: str = 'basic')#

基础类：FlowBase

这个类代表一个流程。

Parameters:

  • id (str) – 流程的id。

  • name (str) – 流程的名称。

  • nodes (List[Node]) – 流程的节点。

  • inputs (Dict[str, FlowInputDefinition]) – 流程的输入。

  • outputs (Dict[str, FlowOutputDefinition]) – 流程的输出。

  • 工具 (列表[工具]) – 流程的工具。

  • node_variants (Dict[str, NodeVariants]) – 流程的节点变体。

  • program_language (str) – 流程的编程语言。

  • environment_variables (Dict[str, object]) – 流程的默认环境变量。

  • message_format (str) – 用于表示不同多媒体合同的流的消息格式类型。

static deserialize(data: dict) → Flow#

从字典中反序列化流程。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的流程。

Return type:

流程

environment_variables: Dict[str, object] = None#
classmethod from_yaml(flow_file: Path, working_dir=None, name=None) → Flow#

从yaml文件加载流程。

get_chat_input_name()#

返回聊天输入的名称。

get_chat_output_name()#

返回聊天输出的名称。

get_connection_input_names_for_node(node_name)#

返回节点的连接输入名称，还将返回未分配的节点连接输入。

Parameters:

node_name – 节点名称

get_connection_names(environment_variables_overrides: Optional[Dict[str, str]] = None)#

返回连接名称。

get_node(node_name)#

返回具有给定名称的节点。

get_tool(tool_name)#

返回具有给定名称的工具。

has_aggregation_node()#

返回流程是否具有聚合节点。

is_chat_flow()#

返回流程是否为聊天流程。

is_llm_node(node)#

给定一个节点，返回它是否使用了LLM工具。

is_node_referenced_by(node: 节点, other_node: 节点)#

给定两个节点，返回第一个节点是否被第二个节点引用。

is_normal_node(node_name)#

返回节点是否为普通节点。

is_reduce_node(node_name)#

返回节点是否为reduce节点。

is_referenced_by_flow_output(node)#

给定一个节点，返回它是否被输出引用。

is_referenced_by_other_node(node)#

给定一个节点，返回它是否被其他节点引用。

classmethod load_env_variables(flow_file: Path, working_dir=None, environment_variables_overrides: Optional[Dict[str, str]] = None) → Dict[str, str]#

从流程 yaml 中读取 flow_environment_variables。 如果存在 environment_variables_overrides，则覆盖 yaml 级别的配置。 返回合并后的环境变量字典。

static load_message_format_from_yaml(flow_file: Path, working_dir=None) → str#
message_format: str = 'basic'#
node_variants: Dict[str, 节点变体] = None#
nodes: List[节点]#
program_language: str = 'python'#
serialize()#

将流程序列化为字典。

Returns:

流程的字典。

Return type:

字典

tools: List[工具]#
class promptflow.contracts.flow.FlowBase(id: str, name: str, inputs: Dict[str, FlowInputDefinition], outputs: Dict[str, FlowOutputDefinition])#

基础类: object

这是流程的基类。

Parameters:
get_connection_names(environment_variables_overrides: Optional[Dict[str, str]] = None)#

返回带有环境变量覆盖的连接名称。 注意：只有在flow.environment_variables中存在的环境变量才会被考虑。

Parameters:

environment_variables_overrides – 用于覆盖流程的环境变量。

Returns:

此流程中使用的连接名称。

get_environment_variables_with_overrides(environment_variables_overrides: Optional[Dict[str, str]] = None) → Dict[str, str]#
id: str#
inputs: Dict[str, FlowInputDefinition]#
name: str#
outputs: Dict[str, FlowOutputDefinition]#
class promptflow.contracts.flow.FlowInitDefinition(type: ValueType, default: Optional[str] = None, description: Optional[str] = None)#

基础：FlowParamDefinitionBase

这个类表示可调用类流程的初始化关键字的定义。

static deserialize(data: dict) → FlowInitDefinition#

从字典中反序列化流程初始化定义。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的流程输入定义。

Return type:

FlowInitDefinition

type: ValueType#
class promptflow.contracts.flow.FlowInputAssignment(value: Any, value_type: InputValueType = InputValueType.LITERAL, section: str = '', property: str = '', prefix: str = 'flow.')#

基础：InputAssignment

此类表示流程输入值的分配。

Parameters:

prefix (str) – 流程输入的前缀。

static deserialize(value: str) → FlowInputAssignment#

将流程输入分配从字符串反序列化。

Parameters:

value (str) – 要反序列化的字符串。

Returns:

从字符串构建的流程输入分配。

Return type:

FlowInputAssignment

static is_flow_input(input_value: str) → bool#

检查输入值是否为流程输入。

Parameters:

input_value (str) – 要检查的输入值。

Returns:

输入值是否为流程输入。

Return type:

布尔

prefix: str = 'flow.'#
class promptflow.contracts.flow.FlowInputDefinition(type: ValueType, default: Optional[str] = None, description: Optional[str] = None, enum: Optional[List[str]] = None, is_chat_input: bool = False, is_chat_history: Optional[bool] = None)#

基础：FlowParamDefinitionBase

此类表示流程输入的定义。

Parameters:

  • type (ValueType) – 流程输入的类型。

  • default (str) – 流程输入的默认值。

  • description (str) – 流程输入的描述。

  • enum (List[str]) – 流程输入的枚举。

  • is_chat_input (bool) – 流程输入是否为聊天输入。

  • is_chat_history (bool) – 流程输入是否为聊天历史记录。

static deserialize(data: dict) → FlowInputDefinition#

从字典中反序列化流程输入定义。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的流程输入定义。

Return type:

FlowInputDefinition

enum: List[str] = None#
is_chat_history: bool = None#
is_chat_input: bool = False#
serialize()#

将流程输入定义序列化为字典。

Returns:

流程输入定义的字典。

Return type:

字典

class promptflow.contracts.flow.FlowOutputDefinition(type: ValueType, reference: InputAssignment, description: str = '', evaluation_only: bool = False, is_chat_output: bool = False)#

基础类: object

此类表示流程输出的定义。

Parameters:

  • type (ValueType) – 流程输出的类型。

  • reference (InputAssignment) – 流程输出的引用。

  • description (str) – 流程输出的描述。

  • evaluation_only (bool) – 流程输出是否仅用于评估。

  • is_chat_output (bool) – 流程输出是否为聊天输出。

description: str = ''#
static deserialize(data: dict)#

从字典中反序列化流程输出定义。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的流程输出定义。

Return type:

FlowOutputDefinition

evaluation_only: bool = False#
is_chat_output: bool = False#
reference: InputAssignment#
serialize()#

将流程输出定义序列化为字典。

Returns:

流程输出定义的字典。

Return type:

字典

type: ValueType#
class promptflow.contracts.flow.FlowParamDefinitionBase(type: ValueType, default: Optional[str] = None, description: Optional[str] = None)#

基础类: object

用于定义流参数（输入和初始化关键字参数）的基类。

default: str = None#
description: str = None#
serialize()#

将流程参数定义序列化为字典。

Returns:

流程参数定义的字典。

Return type:

字典

type: ValueType#
class promptflow.contracts.flow.InputAssignment(value: Any, value_type: InputValueType = InputValueType.LITERAL, section: str = '', property: str = '')#

基础类: object

这个类表示输入值的分配。

Parameters:

  • value (Any) – 输入分配的值。

  • value_type (InputValueType) – 输入分配的类型。

  • section (str) – 输入分配的部分，通常是输出。

  • property (str) – 输入分配中存在于该部分的属性。

static deserialize(value: str) → InputAssignment#

将输入的赋值从字符串反序列化。

Parameters:

value (str) – 要反序列化的字符串。

Returns:

从字符串构建的输入赋值。

Return type:

输入分配

static deserialize_node_reference(data: str) → InputAssignment#

反序列化输入分配的节点引用部分。

Parameters:

data (str) – 要反序列化的字符串。

Returns:

输入节点引用类型的赋值。

Return type:

InputAssignment

static deserialize_reference(value: str) → InputAssignment#

反序列化输入分配中的引用部分（包括节点/流程引用）。

Parameters:

value (str) – 要反序列化的字符串。

Returns:

引用类型的输入赋值。

Return type:

InputAssignment

property: str = ''#
section: str = ''#
serialize()#

将输入的任务序列化为字符串。

value: Any#
value_type: InputValueType = 'Literal'#
class promptflow.contracts.flow.InputValueType(value)#

基础：Enum

输入值类型的枚举。

FLOW_INPUT = 'FlowInput'#
LITERAL = 'Literal'#
NODE_REFERENCE = 'NodeReference'#
class promptflow.contracts.flow.Node(name: str, tool: str, inputs: Dict[str, InputAssignment], comment: str = '', api: Optional[str] = None, provider: Optional[str] = None, module: Optional[str] = None, connection: Optional[str] = None, aggregation: bool = False, enable_cache: bool = False, use_variants: bool = False, source: Optional[ToolSource] = None, type: Optional[ToolType] = None, activate: Optional[ActivateCondition] = None)#

基础类: object

这个类表示流程中的一个节点。

Parameters:

  • name (str) – 节点的名称。

  • 工具 (str) – 节点的工具。

  • inputs (Dict[str, InputAssignment]) – 节点的输入。

  • comment (str) – 节点的注释。

  • api (str) – 节点的api。

  • provider (str) – 节点的提供者。

  • module (str) – 节点的模块。

  • connection (str) – 节点的连接。

  • aggregation (bool) – 节点是否为聚合节点。

  • enable_cache (bool) – 节点是否启用缓存。

  • use_variants (bool) – 节点是否使用变体。

  • source (ToolSource) – 节点的来源。

  • type (ToolType) – 节点的工具类型。

  • activate (ActivateCondition) – 节点的激活条件。

activate: Optional[ActivateCondition] = None#
aggregation: bool = False#
api: str = None#
comment: str = ''#
connection: str = None#
static deserialize(data: dict) → 节点#

从字典中反序列化节点。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构造的节点。

Return type:

节点

enable_cache: bool = False#
inputs: Dict[str, InputAssignment]#
module: str = None#
name: str#
provider: str = None#
serialize()#

将节点序列化为字典。

Returns:

节点的字典。

Return type:

字典

source: Optional[ToolSource] = None#
tool: str#
type: Optional[ToolType] = None#
use_variants: bool = False#
class promptflow.contracts.flow.NodeVariant(node: 节点, description: str = '')#

基础类: object

这个类表示一个节点变体。

Parameters:

  • node (Node) – 节点变体的节点。

  • description (str) – 节点变体的描述。

description: str = ''#
static deserialize(data: dict) → NodeVariant#

从字典中反序列化节点变体。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构造的节点变体。

Return type:

NodeVariant

node: 节点#
class promptflow.contracts.flow.NodeVariants(default_variant_id: str, variants: Dict[str, NodeVariant])#

基础类: object

这个类表示一个节点的变体。

Parameters:

  • default_variant_id (str) – 节点的默认变体ID。

  • variants (Dict[str, NodeVariant]) – 节点的变体。

default_variant_id: str#
static deserialize(data: dict) → 节点变体#

从字典中反序列化节点变体。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的节点变体。

Return type:

NodeVariants

variants: Dict[str, NodeVariant]#
class promptflow.contracts.flow.PromptyFlow(id: str, name: str, inputs: Dict[str, FlowInputDefinition], outputs: Dict[str, FlowOutputDefinition], program_language: str = 'python', environment_variables: Optional[Dict[str, object]] = None, message_format: str = 'basic')#

基础类：FlowBase

这个类代表一个提示流。

Parameters:

  • id (str) – 流程的id。

  • name (str) – 流程的名称。

  • inputs (Dict[str, FlowInputDefinition]) – 流程的输入。

  • outputs (Dict[str, FlowOutputDefinition]) – 流程的输出。

  • program_language (str) – 流程的编程语言。

  • environment_variables (Dict[str, object]) – 流程的默认环境变量。

  • message_format (str) – 用于表示不同多媒体合同的流的消息格式类型。

classmethod deserialize(data: dict) → PromptyFlow#

从字典中反序列化prompty流。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的流程。

Return type:

PromptyFlow

environment_variables: Dict[str, object] = None#
get_connection_names(environment_variables_overrides: Optional[Dict[str, str]] = None)#

返回连接名称。

message_format: str = 'basic'#
program_language: str = 'python'#
class promptflow.contracts.flow.ToolSource(type: ToolSourceType = ToolSourceType.Code, tool: Optional[str] = None, path: Optional[str] = None)#

基础类: object

这个类表示工具的来源。

Parameters:

  • type (ToolSourceType) – 工具源的类型。

  • tool (str) – 工具源的工具。

  • path (str) – 工具源的路径。

static deserialize(data: dict) → ToolSource#

从字典中反序列化工具源。

Parameters:

data (dict) – 要反序列化的字典。

Returns:

从字典构建的工具源。

Return type:

ToolSource

path: Optional[str] = None#
tool: Optional[str] = None#
type: ToolSourceType = 'code'#
class promptflow.contracts.flow.ToolSourceType(value)#

基础：str, Enum

工具源类型的枚举。

Code = 'code'#
Package = 'package'#
PackageWithPrompt = 'package_with_prompt'#