promptflow.contracts.flow 模块#

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

基类：object

此类别表示节点的激活条件。

参数:

condition (InputAssignment) – 激活条件。
condition_value (Any) – 条件的值。

condition: InputAssignment#

condition_value: Any#

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

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

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的激活条件。
返回类型:: 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

此类别表示弹性流。

参数:

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#

从字典中反序列化流。

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的流。
返回类型:: 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[Node], tools: List[Tool], node_variants: Optional[Dict[str, NodeVariants]] = None, program_language: str = 'python', environment_variables: Optional[Dict[str, object]] = None, message_format: str = 'basic')#

基类：FlowBase

此类别表示一个流。

参数:

id (str) – 流的 ID。
name (str) – 流的名称。
nodes (List[Node]) – 流的节点。
inputs (Dict[str, FlowInputDefinition]) – 流的输入。
outputs (Dict[str, FlowOutputDefinition]) – 流的输出。
tools (List[Tool]) – 流的工具。
node_variants (Dict[str, NodeVariants]) – 流的节点变体。
program_language (str) – 流的编程语言。
environment_variables (Dict[str, object]) – 流的默认环境变量。
message_format (str) – 流的消息格式类型，用于表示不同的多媒体契约。

static deserialize(data: dict) → Flow#

从字典中反序列化流。

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的流。
返回类型:: 流

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)#

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

参数:: 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: Node, other_node: Node)#: 给定两个节点，返回第一个节点是否被第二个节点引用。

is_normal_node(node_name)#: 返回节点是否为普通节点。

is_reduce_node(node_name)#: 返回节点是否为归约节点。

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, NodeVariants] = None#

nodes: List[Node]#

program_language: str = 'python'#

serialize()#

将流序列化为字典。

返回:: 流的字典。
返回类型:: dict

tools: List[Tool]#

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

基类：object

这是流的基类。

参数:

id (str) – 流的 ID。
name (str) – 流的名称。
inputs (Dict[str, FlowInputDefinition]) – 流的输入。
outputs (Dict[str, FlowOutputDefinition]) – 流的输出。

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

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

参数:: environment_variables_overrides – 用于覆盖流的环境变量。
返回:: 此流中使用的连接名称。

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#

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

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的流输入定义。
返回类型:: FlowInitDefinition

type: ValueType#

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

基类：InputAssignment

此类表示流输入值的赋值。

参数:: prefix (str) – 流输入的（前缀）。

static deserialize(value: str) → FlowInputAssignment#

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

参数:: value (str) – 要反序列化的字符串。
返回:: 从字符串构建的流输入赋值。
返回类型:: FlowInputAssignment

static is_flow_input(input_value: str) → bool#

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

参数:: input_value (str) – 要检查的输入值。
返回:: 输入值是否为流输入。
返回类型:: bool

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

此类表示流输入的定义。

参数:

type (ValueType) – 流输入的类型。
default (str) – 流输入的默认值。
description (str) – 流输入的描述。
enum (List[str]) – 流输入的枚举。
is_chat_input (bool) – 流输入是否为聊天输入。
is_chat_history (bool) – 流输入是否为聊天历史记录。

static deserialize(data: dict) → FlowInputDefinition#

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

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的流输入定义。
返回类型:: FlowInputDefinition

enum: List[str] = None#

is_chat_history: bool = None#

is_chat_input: bool = False#

serialize()#

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

返回:: 流输入定义的字典。
返回类型:: dict

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

基类：object

此类表示流输出的定义。

参数:

type (ValueType) – 流输出的类型。
reference (InputAssignment) – 流输出的引用。
description (str) – 流输出的描述。
evaluation_only (bool) – 流输出是否仅用于评估。
is_chat_output (bool) – 流输出是否为聊天输出。

description: str = ''#

static deserialize(data: dict)#

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

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的流输出定义。
返回类型:: FlowOutputDefinition

evaluation_only: bool = False#

is_chat_output: bool = False#

reference: InputAssignment#

serialize()#

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

返回:: 流输出定义的字典。
返回类型:: dict

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()#

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

返回:: 流参数定义的字典。
返回类型:: dict

type: ValueType#

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

基类：object

此类表示输入值的赋值。

参数:

value (Any) – 输入赋值的值。
value_type (InputValueType) – 输入赋值的类型。
section (str) – 输入赋值的部分，通常是输出。
property (str) – 部分中存在的输入赋值的属性。

static deserialize(value: str) → InputAssignment#

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

参数:: value (str) – 要反序列化的字符串。
返回:: 从字符串构建的输入赋值。
返回类型:: InputAssignment

static deserialize_node_reference(data: str) → InputAssignment#

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

参数:: data (str) – 要反序列化的字符串。
返回:: 节点引用类型的输入赋值。
返回类型:: InputAssignment

static deserialize_reference(value: str) → InputAssignment#

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

参数:: value (str) – 要反序列化的字符串。
返回:: 引用类型的输入赋值。
返回类型:: 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

此类表示流中的节点。

参数:

name (str) – 节点的名称。
tool (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) → Node#

从字典中反序列化节点。

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的节点。
返回类型:: 节点

enable_cache: bool = False#

inputs: Dict[str, InputAssignment]#

module: str = None#

name: str#

provider: str = None#

serialize()#

将节点序列化为字典。

返回:: 节点的字典。
返回类型:: dict

source: Optional[ToolSource] = None#

tool: str#

type: Optional[ToolType] = None#

use_variants: bool = False#

class promptflow.contracts.flow.NodeVariant(node: Node, description: str = '')#

基类：object

此类表示节点变体。

参数:

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

description: str = ''#

static deserialize(data: dict) → NodeVariant#

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

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的节点变体。
返回类型:: 节点变体

node: Node#

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

基类：object

此类表示节点的变体。

参数:

default_variant_id (str) – 节点的默认变体 ID。
variants (Dict[str, NodeVariant]) – 节点的变体。

default_variant_id: str#

static deserialize(data: dict) → NodeVariants#

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

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的节点变体。
返回类型:: 节点变体

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

此类表示一个 Prompty 流。

参数:

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 流。

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的流。
返回类型:: 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

此类表示工具的来源。

参数:

type (ToolSourceType) – 工具来源的类型。
tool (str) – 工具来源的工具。
path (str) – 工具来源的路径。

static deserialize(data: dict) → ToolSource#

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

参数:: data (dict) – 要反序列化的字典。
返回:: 从字典构建的工具源。
返回类型:: 工具源

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'#