promptflow.contracts.flow module#

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

Bases: object

This class represents the activate condition of a node.

Parameters:

condition (InputAssignment) – The condition of the activate condition.
condition_value (Any) – The value of the condition.

condition: InputAssignment#

condition_value: Any#

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

Deserialize the activate condition from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The activate condition constructed from the dict.
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)#

Bases: FlowBase

This class represents a flex flow.

Parameters:

id (str) – The id of the flow.
name (str) – The name of the flow.
inputs (Dict[str, FlowInputDefinition]) – The inputs of the flow.
outputs (Dict[str, FlowOutputDefinition]) – The outputs of the flow.
program_language (str) – The program language of the flow.
environment_variables (Dict[str, object]) – The default environment variables of the flow.
message_format (str) – The message format type of the flow to represent different multimedia contracts.
sample (Dict[str, object]) – Sample data for the flow. Will become default inputs & init kwargs if not provided.

static deserialize(data: dict) → FlexFlow#

Deserialize the flow from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The flow constructed from the dict.
Return type:: EagerFlow

environment_variables: Dict[str, object] = None#

get_connection_names(environment_variables_overrides: Optional[Dict[str, str]] = None)#: Return connection names.

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

Bases: FlowBase

This class represents a flow.

Parameters:

id (str) – The id of the flow.
name (str) – The name of the flow.
nodes (List[Node]) – The nodes of the flow.
inputs (Dict[str, FlowInputDefinition]) – The inputs of the flow.
outputs (Dict[str, FlowOutputDefinition]) – The outputs of the flow.
tools (List[Tool]) – The tools of the flow.
node_variants (Dict[str, NodeVariants]) – The node variants of the flow.
program_language (str) – The program language of the flow.
environment_variables (Dict[str, object]) – The default environment variables of the flow.
message_format (str) – The message format type of the flow to represent different multimedia contracts.

static deserialize(data: dict) → Flow#

Deserialize the flow from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The flow constructed from the dict.
Return type:: Flow

environment_variables: Dict[str, object] = None#

classmethod from_yaml(flow_file: Path, working_dir=None, name=None) → Flow#: Load flow from yaml file.

get_chat_input_name()#: Return the name of the chat input.

get_chat_output_name()#: Return the name of the chat output.

get_connection_input_names_for_node(node_name)#

Return connection input names for a node, will also return node connection inputs without assignment.

Parameters:: node_name – node name

get_connection_names(environment_variables_overrides: Optional[Dict[str, str]] = None)#: Return connection names.

get_node(node_name)#: Return the node with the given name.

get_tool(tool_name)#: Return the tool with the given name.

has_aggregation_node()#: Return whether the flow has aggregation node.

is_chat_flow()#: Return whether the flow is a chat flow.

is_llm_node(node)#: Given a node, return whether it uses LLM tool.

is_node_referenced_by(node: Node, other_node: Node)#: Given two nodes, return whether the first node is referenced by the second node.

is_normal_node(node_name)#: Return whether the node is a normal node.

is_reduce_node(node_name)#: Return whether the node is a reduce node.

is_referenced_by_flow_output(node)#: Given a node, return whether it is referenced by output.

is_referenced_by_other_node(node)#: Given a node, return whether it is referenced by other node.

classmethod load_env_variables(flow_file: Path, working_dir=None, environment_variables_overrides: Optional[Dict[str, str]] = None) → Dict[str, str]#: Read flow_environment_variables from flow yaml. If environment_variables_overrides exists, override yaml level configuration. Returns the merged environment variables dict.

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

Serialize the flow to a dict.

Returns:: The dict of the flow.
Return type:: dict

tools: List[Tool]#

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

Bases: object

This is base class of flow.

Parameters:

id (str) – The id of the flow.
name (str) – The name of the flow.
inputs (Dict[str, FlowInputDefinition]) – The inputs of the flow.
outputs (Dict[str, FlowOutputDefinition]) – The outputs of the flow.

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

Return connection names with environment variables overrides. Note: only environment variables exist in flow.environment_variables will be considered.

Parameters:: environment_variables_overrides – used to override flow’s environment variables.
Returns:: connection names used in this flow.

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

Bases: FlowParamDefinitionBase

This class represents the definition of a callable class flow’s init kwargs.

static deserialize(data: dict) → FlowInitDefinition#

Deserialize the flow init definition from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The flow input definition constructed from the dict.
Return type:: FlowInitDefinition

type: ValueType#

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

Bases: InputAssignment

This class represents the assignment of a flow input value.

Parameters:: prefix (str) – The prefix of the flow input.

static deserialize(value: str) → FlowInputAssignment#

Deserialize the flow input assignment from a string.

Parameters:: value (str) – The string to be deserialized.
Returns:: The flow input assignment constructed from the string.
Return type:: FlowInputAssignment

static is_flow_input(input_value: str) → bool#

Check whether the input value is a flow input.

Parameters:: input_value (str) – The input value to be checked.
Returns:: Whether the input value is a flow input.
Return type:: 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)#

Bases: FlowParamDefinitionBase

This class represents the definition of a flow input.

Parameters:

type (ValueType) – The type of the flow input.
default (str) – The default value of the flow input.
description (str) – The description of the flow input.
enum (List[str]) – The enum of the flow input.
is_chat_input (bool) – Whether the flow input is a chat input.
is_chat_history (bool) – Whether the flow input is a chat history.

static deserialize(data: dict) → FlowInputDefinition#

Deserialize the flow input definition from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The flow input definition constructed from the dict.
Return type:: FlowInputDefinition

enum: List[str] = None#

is_chat_history: bool = None#

is_chat_input: bool = False#

serialize()#

Serialize the flow input definition to a dict.

Returns:: The dict of the flow input definition.
Return type:: dict

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

Bases: object

This class represents the definition of a flow output.

Parameters:

type (ValueType) – The type of the flow output.
reference (InputAssignment) – The reference of the flow output.
description (str) – The description of the flow output.
evaluation_only (bool) – Whether the flow output is for evaluation only.
is_chat_output (bool) – Whether the flow output is a chat output.

description: str = ''#

static deserialize(data: dict)#

Deserialize the flow output definition from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The flow output definition constructed from the dict.
Return type:: FlowOutputDefinition

evaluation_only: bool = False#

is_chat_output: bool = False#

reference: InputAssignment#

serialize()#

Serialize the flow output definition to a dict.

Returns:: The dict of the flow output definition.
Return type:: dict

type: ValueType#

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

Bases: object

Base class for the definition of a flow param (input & init kwargs).

default: str = None#

description: str = None#

serialize()#

Serialize the flow param definition to a dict.

Returns:: The dict of the flow param definition.
Return type:: dict

type: ValueType#

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

Bases: object

This class represents the assignment of an input value.

Parameters:

value (Any) – The value of the input assignment.
value_type (InputValueType) – The type of the input assignment.
section (str) – The section of the input assignment, usually the output.
property (str) – The property of the input assignment that exists in the section.

static deserialize(value: str) → InputAssignment#

Deserialize the input assignment from a string.

Parameters:: value (str) – The string to be deserialized.
Returns:: The input assignment constructed from the string.
Return type:: InputAssignment

static deserialize_node_reference(data: str) → InputAssignment#

Deserialize the node reference part of an input assignment.

Parameters:: data (str) – The string to be deserialized.
Returns:: Input assignment of node reference type.
Return type:: InputAssignment

static deserialize_reference(value: str) → InputAssignment#

Deserialize the reference(including node/flow reference) part of an input assignment.

Parameters:: value (str) – The string to be deserialized.
Returns:: The input assignment of reference types.
Return type:: InputAssignment

property: str = ''#

section: str = ''#

serialize()#: Serialize the input assignment to a string.

value: Any#

value_type: InputValueType = 'Literal'#

class promptflow.contracts.flow.InputValueType(value)#

Bases: Enum

The enum of input value type.

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

Bases: object

This class represents a node in a flow.

Parameters:

name (str) – The name of the node.
tool (str) – The tool of the node.
inputs (Dict[str, InputAssignment]) – The inputs of the node.
comment (str) – The comment of the node.
api (str) – The api of the node.
provider (str) – The provider of the node.
module (str) – The module of the node.
connection (str) – The connection of the node.
aggregation (bool) – Whether the node is an aggregation node.
enable_cache (bool) – Whether the node enable cache.
use_variants (bool) – Whether the node use variants.
source (ToolSource) – The source of the node.
type (ToolType) – The tool type of the node.
activate (ActivateCondition) – The activate condition of the node.

activate: Optional[ActivateCondition] = None#

aggregation: bool = False#

api: str = None#

comment: str = ''#

connection: str = None#

static deserialize(data: dict) → Node#

Deserialize the node from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The node constructed from the dict.
Return type:: Node

enable_cache: bool = False#

inputs: Dict[str, InputAssignment]#

module: str = None#

name: str#

provider: str = None#

serialize()#

Serialize the node to a dict.

Returns:: The dict of the node.
Return type:: dict

source: Optional[ToolSource] = None#

tool: str#

type: Optional[ToolType] = None#

use_variants: bool = False#

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

Bases: object

This class represents a node variant.

Parameters:

node (Node) – The node of the node variant.
description (str) – The description of the node variant.

description: str = ''#

static deserialize(data: dict) → NodeVariant#

Deserialize the node variant from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The node variant constructed from the dict.
Return type:: NodeVariant

node: Node#

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

Bases: object

This class represents the variants of a node.

Parameters:

default_variant_id (str) – The default variant id of the node.
variants (Dict[str, NodeVariant]) – The variants of the node.

default_variant_id: str#

static deserialize(data: dict) → NodeVariants#

Deserialize the node variants from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The node variants constructed from the dict.
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')#

Bases: FlowBase

This class represents a prompty flow.

Parameters:

id (str) – The id of the flow.
name (str) – The name of the flow.
inputs (Dict[str, FlowInputDefinition]) – The inputs of the flow.
outputs (Dict[str, FlowOutputDefinition]) – The outputs of the flow.
program_language (str) – The program language of the flow.
environment_variables (Dict[str, object]) – The default environment variables of the flow.
message_format (str) – The message format type of the flow to represent different multimedia contracts.

classmethod deserialize(data: dict) → PromptyFlow#

Deserialize the prompty flow from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The flow constructed from the dict.
Return type:: PromptyFlow

environment_variables: Dict[str, object] = None#

get_connection_names(environment_variables_overrides: Optional[Dict[str, str]] = None)#: Return connection names.

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

Bases: object

This class represents the source of a tool.

Parameters:

type (ToolSourceType) – The type of the tool source.
tool (str) – The tool of the tool source.
path (str) – The path of the tool source.

static deserialize(data: dict) → ToolSource#

Deserialize the tool source from a dict.

Parameters:: data (dict) – The dict to be deserialized.
Returns:: The tool source constructed from the dict.
Return type:: ToolSource

path: Optional[str] = None#

tool: Optional[str] = None#

type: ToolSourceType = 'code'#

class promptflow.contracts.flow.ToolSourceType(value)#

Bases: str, Enum

The enum of tool source type.

Code = 'code'#

Package = 'package'#

PackageWithPrompt = 'package_with_prompt'#