Subworkflow Deployment Node

Example Usage

1 from vellum import ChatMessage, WorkflowExpandMetaRequest
2 from vellum.workflows.nodes import SubworkflowDeploymentNode
3 
4 class MySubworkflowNode(SubworkflowDeploymentNode):
5     deployment = "customer_support_workflow"  # or UUID("...")
6     subworkflow_inputs = {
7         "user_query": "How do I reset my password?",
8         "chat_history": [
9             ChatMessage(role="USER", text="Hi there"),
10             ChatMessage(role="ASSISTANT", text="Hello! How can I help?")
11         ],
12         "user_metadata": {
13             "user_id": "123",
14             "account_type": "premium"
15         },
16         "priority_score": 0.8
17     }
18     release_tag = "production"
19     external_id = "support-ticket-456"
20     expand_meta = WorkflowExpandMetaRequest(
21         include_inputs=True,
22         include_outputs=True
23     )
24     metadata = {
25         "source": "mobile_app",
26         "region": "us-west"
27     }

1 # Outputs are streamed as they're generated by the workflow
2 for output in node.run():
3     if output.is_streaming:
4         print(f"Received partial output for {output.name}: {output.delta}")
5     elif output.is_fulfilled:
6         print(f"Received final output for {output.name}: {output.value}")

from vellum import ChatMessage, WorkflowExpandMetaRequest
from vellum.workflows.nodes import SubworkflowDeploymentNode
class MySubworkflowNode(SubworkflowDeploymentNode):
    deployment = "customer_support_workflow"  # or UUID("...")
    subworkflow_inputs = {
        "user_query": "How do I reset my password?",
        "chat_history": [
            ChatMessage(role="USER", text="Hi there"),
            ChatMessage(role="ASSISTANT", text="Hello! How can I help?")
        ],
        "user_metadata": {
            "user_id": "123",
            "account_type": "premium"
        },
        "priority_score": 0.8
    }
    release_tag = "production"
    external_id = "support-ticket-456"
    expand_meta = WorkflowExpandMetaRequest(
        include_inputs=True,
        include_outputs=True
    )
    metadata = {
        "source": "mobile_app",
        "region": "us-west"
    }

# Outputs are streamed as they're generated by the workflow
for output in node.run():
    if output.is_streaming:
        print(f"Received partial output for {output.name}: {output.delta}")
    elif output.is_fulfilled:
        print(f"Received final output for {output.name}: {output.value}")

vellum.workflows.nodes.SubworkflowDeploymentNode

Used to execute a deployed Workflow as a subworkflow within another workflow. This allows you to compose workflows from other deployed workflows.

Attributes

deployment

Union[UUID, str]Required

Either the Workflow Deployment’s UUID or its name

subworkflow_inputs

Dict[str, Any]Required

The inputs for the Subworkflow. Supports:

Strings
Numbers (float)
Chat History (List[ChatMessage])
JSON objects (Dict[str, Any])

release_tag

strDefaults to LATEST

The release tag to use for the Workflow Execution

external_id

Optional[str]

The external ID to use for the Workflow Execution

expand_meta

Optional[WorkflowExpandMetaRequest]

An optionally specified configuration used to opt in to including additional metadata about this workflow execution in the API response. Corresponding values will be returned under the execution_meta key within NODE events in the response stream. See Execute Workflow: Expand Meta.

metadata

Optional[Dict[str, Optional[Any]]]

The metadata to use for the Workflow Execution

request_options

Optional[RequestOptions]

The request options to use for the Workflow Execution

Outputs

The outputs of this node are determined by the outputs defined in the deployed workflow. Each output from the workflow will be streamed through this node.

1	from vellum import ChatMessage, WorkflowExpandMetaRequest
2	from vellum.workflows.nodes import SubworkflowDeploymentNode
3
4	class MySubworkflowNode(SubworkflowDeploymentNode):
5	deployment = "customer_support_workflow" # or UUID("...")
6	subworkflow_inputs = {
7	"user_query": "How do I reset my password?",
8	"chat_history": [
9	ChatMessage(role="USER", text="Hi there"),
10	ChatMessage(role="ASSISTANT", text="Hello! How can I help?")
11	],
12	"user_metadata": {
13	"user_id": "123",
14	"account_type": "premium"
15	},
16	"priority_score": 0.8
17	}
18	release_tag = "production"
19	external_id = "support-ticket-456"
20	expand_meta = WorkflowExpandMetaRequest(
21	include_inputs=True,
22	include_outputs=True
23	)
24	metadata = {
25	"source": "mobile_app",
26	"region": "us-west"
27	}

1	# Outputs are streamed as they're generated by the workflow
2	for output in node.run():
3	if output.is_streaming:
4	print(f"Received partial output for {output.name}: {output.delta}")
5	elif output.is_fulfilled:
6	print(f"Received final output for {output.name}: {output.value}")