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'