Search Node

vellum.workflows.nodes.SearchNode

Used to perform a hybrid search against a Document Index in Vellum.

Attributes

document_index

Union[UUID, str]Required

Either the UUID or name of the Vellum Document Index that you’d like to search against

query

strRequired

The query to search for

options

SearchRequestOptionsRequest

Runtime configuration for the search

chunk_separator

str

The separator to use when joining the text of each search result

Outputs

text

str

The concatenated text output from the search

results

List[SearchResult]

The raw results from the search

Example Usage

1 from vellum import (
2     SearchRequestOptionsRequest,
3     SearchWeightsRequest,
4     SearchResultMergingRequest,
5     SearchFiltersRequest,
6 )
7 from vellum.workflows.nodes.displayable import SearchNode
8 
9 class MySearchNode(SearchNode):
10     query = "Hello, world"
11     document_index = "my_document_index"
12     options = SearchRequestOptionsRequest(
13         limit=8,
14         weights=SearchWeightsRequest(
15             semantic_similarity=0.8,
16             keywords=0.2,
17         ),
18         result_merging=SearchResultMergingRequest(
19             enabled=True
20         ),
21         filters=SearchFiltersRequest(
22             external_ids=None,
23             metadata=None,
24         ),
25     )
26     chunk_separator = "\n\n#####\n\n"

Example Outputs

1 from vellum import SearchResult, SearchResultDocument
2 
3 MySearchNode.Outputs(
4     text="Goodbye, world",
5     results=[
6         SearchResult(
7             text="Goodbye, world",
8             score=0.5,
9             keywords=[...],
10             document=SearchResultDocument(
11                 id="<id>",
12                 label="My Document",
13                 external_id="<external-id>",
14                 metadata={},
15             ),
16             meta=None,
17         ),
18         ...
19     ],
20 )

Inline Prompt Node

vellum.workflows.nodes.InlinePromptNode

Used to execute a prompt directly within a workflow, without requiring a prompt deployment.

Attributes

prompt_inputs

EntityInputsInterfaceRequired

The inputs for the Prompt

blocks

List[PromptBlock]Required

The blocks that make up the Prompt

ml_model

strRequired

The model to use for execution (e.g., “gpt-4o”, “claude-3.5-sonnet”)

functions

Optional[List[FunctionDefinition]]

The functions to include in the prompt

parameters

Optional[PromptParameters]

Model parameters for execution. Defaults to:

temperature: 0.0
max_tokens: 1000
top_p: 1.0
top_k: 0
frequency_penalty: 0.0
presence_penalty: 0.0
logitBias:
customParameters:

expand_meta

Optional[PromptDeploymentExpandMetaRequest]

Expandable execution fields to include in the response. See more here.

request_options

RequestOptions

Additional options for request-specific configuration when calling APIs via the SDK. This is used primarily as an optional final parameter for service functions.

timeout_in_seconds: The number of seconds to await an API call before timing out
max_retries: The max number of retries to attempt if the API call fails
additional_headers: A dictionary containing additional parameters to spread into the request’s header dict
additional_query_parameters: A dictionary containing additional parameters to spread into the request’s query parameters dict
additional_body_parameters: A dictionary containing additional parameters to spread into the request’s body parameters dict

Outputs

text

str

The generated text output from the prompt execution

results

List[PromptOutput]

The array of results from the prompt execution. PromptOutput is a union of the following types:

StringVellumValue
JsonVellumValue
ErrorVellumValue
FunctionCallVellumValue

Example Usage

1 from vellum.workflows.nodes import InlinePromptNode
2 from vellum import (
3     ChatMessagePromptBlock,
4     JinjaPromptBlock,
5 )
6 
7 class MyInlinePromptNode(InlinePromptNode):
8     ml_model = "gpt-4o"
9     blocks = [
10         ChatMessagePromptBlock(
11             chat_role="SYSTEM",
12             blocks=[
13                 JinjaPromptBlock(
14                     template="""\
15 Answer the user's questions based on the context provided below, if you don't know the answer say "Sorry I don't know"
16 
17 **Context**
18 ``
19 {{ context }}
20 ``
21 
22 Limit your answer to 250 words and provide a citation at the end of your answer\
23 """,
24                 ),
25             ],
26         ),
27         ChatMessagePromptBlock(
28             chat_role="USER",
29             blocks=[JinjaPromptBlock(template="{{ question }}")],
30         ),
31     ]
32     prompt_inputs = {"question": "What is the best AI development platform in the world?", "context": "..."}

Inline Subworkflow Node

vellum.workflows.nodes.InlineSubworkflowNode

Used to execute a Subworkflow defined inline within your workflow. This allows you to modularize and reuse workflow logic.

Attributes

subworkflow

Type[BaseWorkflow[WorkflowInputsType, InnerStateType]]Required

The Subworkflow class to execute

subworkflow_inputs

Dict[str, Any]Required

The inputs to pass to the subworkflow

Outputs

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

1 from vellum.workflows.nodes import InlineSubworkflowNode
2 from vellum.workflows.inputs.base import BaseInputs
3 from vellum.workflows.state import BaseState
4 from vellum.workflows import BaseWorkflow
5 
6 class SubworkflowInputs(BaseInputs):
7     query: str
8     max_results: int
9 
10 class SubworkflowState(BaseState):
11     pass
12 
13 class MySubworkflow(BaseWorkflow[SubworkflowInputs, SubworkflowState]):
14     # Define your subworkflow logic here
15     pass
16 
17 class MyInlineSubworkflowNode(InlineSubworkflowNode):
18     subworkflow = MySubworkflow
19     subworkflow_inputs = {
20         "query": "search term",
21         "max_results": 10
22     }

1 # Outputs depend on the subworkflow's defined outputs
2 MyInlineSubworkflowNode.Outputs(
3     search_results=["Result 1", "Result 2", "Result 3"],
4     metadata={
5         "total_found": 100,
6         "processing_time": 0.5
7     }
8 )

Prompt Deployment Node

vellum.workflows.nodes.PromptDeploymentNode

Used to execute a Prompt Deployment and surface a string output for convenience.

Attributes

deployment

Union[UUID, str]Required

Either the Prompt Deployment’s UUID or its name

prompt_inputs

EntityInputsInterfaceRequired

The inputs for the Prompt

release_tag

strDefaults to LATEST

The release tag to use for the Prompt Execution

external_id

Optional[str]

Optionally include a unique identifier for tracking purposes. Must be unique within a given Prompt Deployment.

expand_meta

Optional[PromptDeploymentExpandMetaRequest]

Expandable execution fields to include in the response. See more here.

raw_overrides

Optional[RawPromptExecutionOverridesRequest]

The raw overrides to use for the Prompt Execution

expand_raw

Optional[Sequence[str]]

Expandable raw fields to include in the response

metadata

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

The metadata to use for the Prompt Execution

request_options

Optional[RequestOptions]

The request options to use for the Prompt Execution

Outputs

text

str

The generated text output from the prompt execution

results

List[PromptOutput]

The array of results from the prompt execution. PromptOutput is a union of the following types:

StringVellumValue
FunctionCallVellumValue

Example Usage

1 from vellum.workflows.nodes import PromptDeploymentNode
2 from vellum.client import PromptDeploymentExpandMetaRequest, RawPromptExecutionOverridesRequest
3 
4 class MyPromptDeploymentNode(PromptDeploymentNode):
5     deployment = "my_prompt_deployment"
6     prompt_inputs = {
7         "question": "What is the meaning of life?",
8         "chat_history": [
9             ChatMessage(role="USER", text="Hello!"),
10             ChatMessage(role="ASSISTANT", text="Hi there!"),
11         ],
12         "context": {
13             "source": "philosophy_book",
14             "chapter": 42
15         }
16     }
17     release_tag = "production"
18     external_id = "unique-execution-id"
19     expand_meta = PromptDeploymentExpandMetaRequest(
20         model_name=True,
21         usage=True,
22         cost=True,
23         finish_reason=True,
24         latency=True,
25         deployment_release_tag=True,
26         prompt_version_id=True
27     )
28     metadata = {
29         "user_id": "123",
30         "session_id": "abc"
31     }

1 MyPromptDeploymentNode.Outputs(
2     text="happily",
3     results=[
4         StringVellumValue(value="h"),
5         StringVellumValue(value="app"),
6         StringVellumValue(value="ily"),
7     ]
8 )

Subworkflow Deployment Node

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.

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}")

API Node

vellum.workflows.nodes.APINode

Used to make HTTP requests to external APIs.

Attributes

url

strRequired

The URL to send the request to.

method

APIRequestMethodRequired

The HTTP method to use for the request.

headers

Optional[Dict[str, Union[str, VellumSecret]]]

The headers to send in the request.

data

Optional[str]

The data to send in the request body.

json

Optional[JsonObject]

The JSON data to send in the request body.

authorization_type

Optional[AuthorizationType]

The type of authorization to use for the API call.

api_key_header_key

Optional[str]

The header key to use for the API key authorization.

bearer_token_value

Optional[str]

The bearer token value to use for the bearer token authorization.

Outputs

status_code

int

The HTTP status code of the response

headers

Dict[str, str]

The response headers

text

str

The raw text response body

json

Optional[Dict[str, Any]]

The parsed JSON response body (if the response is JSON)

Code Execution Node

vellum.workflows.nodes.CodeExecutionNode

Used to execute arbitrary Python code within your workflow. Supports custom package dependencies and any Python or TypeScript runtimes.

Attributes

filepath

strRequired

Path to the Python script file to execute

code_inputs

EntityInputsInterfaceRequired

The inputs for the custom script. Supports:

Strings
Numbers (float)
Arrays
Chat History (List[ChatMessage])
Search Results (List[SearchResult])
JSON objects (Dict[str, Any])
Function Calls
Errors
Secrets

runtime

CodeExecutionRuntimeDefaults to PYTHON_3_12

The runtime to use for the custom script

packages

Optional[Sequence[CodeExecutionPackage]]

The packages to use for the custom script

request_options

Optional[RequestOptions]

The request options to use for the custom script

Outputs

result

_OutputType

The result returned by the executed code, type depends on the node’s generic type parameter

log

str

The execution logs from the code run

1 from vellum.workflows.nodes import CodeExecutionNode
2 from vellum import ChatMessage, CodeExecutionPackage
3 from vellum.workflows.state import BaseState
4 from typing import List, Dict
5 
6 class MyCodeExecutionNode(CodeExecutionNode[BaseState, Dict[str, Any]]):
7     filepath = "./scripts/process_data.py"
8     code_inputs = {
9         "text": "Process this text",
10         "chat_history": [
11             ChatMessage(role="user", content="Hello"),
12             ChatMessage(role="assistant", content="Hi there!")
13         ],
14         "config": {
15             "max_length": 100,
16             "temperature": 0.7
17         }
18     }
19     runtime = "PYTHON_3_11_6"
20     packages = [
21         CodeExecutionPackage(name="pandas", version="2.0.0"),
22         CodeExecutionPackage(name="numpy", version="1.24.0")
23     ]

1 MyCodeExecutionNode.Outputs(
2     result={
3         "processed_text": "PROCESS THIS TEXT",
4         "history_length": 2,
5         "config_used": {
6             "max_length": 100,
7             "temperature": 0.7
8         }
9     },
10     log="INFO: Starting processing...\nINFO: Processing complete"
11 )

Templating Node

vellum.workflows.nodes.TemplatingNode

Used to render text templates using Jinja2 syntax

Attributes

template

strRequired

The Jinja2 template to render

inputs

EntityInputsInterfaceRequired

The values to substitute into the template for the specified variables.

Outputs

result

_OutputType

The rendered template in the appropriate type as specified by the type specified in the Templating Node subclass definition.

Guardrail Node

vellum.workflows.nodes.GuardrailNode

Used to execute a Metric Definition and surface a float output representing the score. This node is commonly used to implement quality checks and guardrails in your workflows.

Metrics are defined in the Vellum UI and can be reused across workflows.

Attributes

metric_definition

Union[UUID, str]Required

Either the Metric Definition’s UUID or its name

metric_inputs

EntityInputsInterfaceRequired

The inputs for the Metric

release_tag

strDefaults to LATEST

The release tag to use for the Metric

request_options

Optional[RequestOptions]

The request options to use for the Metric execution

Outputs

score

float

The score output from the metric execution (between 0 and 1)

Additional outputs may be available depending on the metric definition.

Example Usage

1 from vellum import ChatMessage
2 from vellum.workflows.nodes import GuardrailNode
3 
4 class QualityCheckNode(GuardrailNode):
5     release_tag = "production"
6     metric_definition = "response_quality_metric"  # or UUID("...")
7     # `metric_inputs` align with the custom Metric we've created in the Vellum UI
8     metric_inputs = {
9         "response": "This is the AI response to evaluate",
10         "chat_history": [
11             ChatMessage(role="USER", text="What's the weather?"),
12             ChatMessage(role="ASSISTANT", text="It's sunny today!")
13         ],
14         "evaluation_criteria": {
15             "coherence": True,
16             "relevance": True,
17             "safety": True
18         }
19     }

Basic Output Example

1 QualityCheckNode.Outputs(
2     # `score` is a mandatory field for all Metrics and thus will be available on all Guardrail Nodes
3     score=0.92, 
4     # `details` is a result of our custom Metric definition
5     details={ 
6         "coherence_score": 0.95,
7         "relevance_score": 0.88,
8         "safety_score": 0.93
9     }
10 )

Map Node

vellum.workflows.nodes.MapNode

Used to map over a list of items and execute a Subworkflow for each item. This enables parallel processing of multiple items through the same workflow logic.

Attributes

items

List[Any]Required

The list of items to map over. Each item will be processed by the subworkflow.

subworkflow

Type[BaseWorkflow[WorkflowInputsType, InnerStateType]]Required

The Subworkflow class to execute for each item

concurrency

Optional[int]Defaults to None

The maximum number of concurrent subworkflow executions.

Outputs

The outputs are determined by the subworkflow’s outputs, with each output field becoming a list containing results from all iterations.

Basic Example

1 from vellum.workflows.inputs.base import BaseInputs
2 from vellum.workflows.state import BaseState
3 from vellum.workflows import BaseWorkflow
4 from vellum.workflows.nodes import MapNode, BaseNode
5 from typing import List
6 
7 class MapNodeInputs(BaseInputs):
8     my_list: List[str]
9 
10 class Iteration(BaseNode):
11     item = MapNode.SubworkflowInputs.item
12     index = MapNode.SubworkflowInputs.index
13 
14     class Outputs(BaseNode.Outputs):
15         result: str
16 
17     def run(self) -> Outputs:
18         return self.Outputs(result=self.item + str(self.index))
19     
20 class IterationSubworkflow(BaseWorkflow[MapNode.SubworkflowInputs, BaseState]):
21     graph = Iteration
22 
23     class Outputs(BaseWorkflow.Outputs):
24         result = Iteration.Outputs.result
25 
26 class MyMapNode(MapNode):
27     items = MapNodeInputs.my_list
28     subworkflow = IterationSubworkflow
29 
30     class Outputs(BaseNode.Outputs):
31         result: List[str] # this MUST match the Subworkflow's Outputs.result attribute

Example Outputs

1 MyMapNode.Outputs(
2     result=[
3         "text0",
4         "text1",
5         "text2"
6     ]
7 )

Conditional Node

vellum.workflows.nodes.ConditionalNode

Used to conditionally determine which port to invoke next. This node exists to be backwards compatible with Vellum’s Conditional Node, and for most cases, you should extend BaseNode.Ports directly.

Merge Node

vellum.workflows.nodes.MergeNode

Used to merge the control flow of multiple nodes into a single node. This node exists primarily for backwards compatibility with Vellum’s Merge Node functionality.

For most cases, you should extend from BaseNode.Trigger directly instead of using MergeNode. Read more on Triggers

Attributes

This node doesn’t have any specific attributes as it’s a simple control flow node.

Outputs

This node passes through any outputs from its parent nodes without modification.

Error Node

vellum.workflows.nodes.ErrorNode

Used to raise an error to reject the surrounding Workflow and return custom error messages.

Attributes

error

Union[str, VellumError]Required

The error to raise. Can be either:

A string message
A VellumError object with a message and error code

Outputs

This node doesn’t produce outputs. However, Workflows that invoke this node will error and the error message will appear in the Vellum UI.

1 from vellum.workflows import BaseWorkflow
2 from vellum.workflows.nodes import ErrorNode
3 from vellum.workflows.inputs.base import BaseInputs
4 from vellum.workflows.state import BaseState
5 
6 class Inputs(BaseInputs):
7     threshold: int
8 
9 class StartNode(BaseNode):
10     class Ports(BaseNode.Ports):
11         success = Port.on_if(Inputs.threshold.greater_than(10))
12         fail = Port.on_else()
13 
14 class SuccessNode(BaseNode):
15     class Outputs(BaseNode.Outputs):
16         result = Inputs.threshold
17 
18 class FailNode(ErrorNode):
19     error = "Input threshold was too low"
20 
21 class BasicErrorNodeWorkflow(BaseWorkflow[Inputs, BaseState]):
22     graph = {
23         StartNode.Ports.success >> SuccessNode,
24         StartNode.Ports.fail >> FailNode,
25     }
26     class Outputs(BaseWorkflow.Outputs):
27         final_value = SuccessNode.Outputs.result

Final Output Node

vellum.workflows.nodes.FinalOutputNode

Used to directly reference the output of another node. This provides backward compatibility with Vellum’s Final Output Node functionality. In most cases, you should reference the Workflow Output directly to the output of a particular node, without adding a Final Output Node.

Attributes

value

_OutputTypeRequired

The value to be output. The type is inferred by the type of the node connected to this attribute.

Outputs

value

_OutputType

The output value, with the same type as the input value

1	from vellum.workflows.nodes import APINode
2
3	class MyAPINode(APINode):
4	url = "https://api.example.com/data"
5	method = "POST"
6	headers = {
7	"Authorization": "Bearer ${secrets.API_KEY}",
8	"Content-Type": "application/json"
9	}
10	json = {
11	"query": "example",
12	"filters": {
13	"category": "books",
14	"limit": 10
15	}
16	}

1	from vellum.workflows.nodes import TemplatingNode
2
3	# Notice `str` is used to specify the output type of the Templating Node
4	# ... in fact, `str` is also the default output type for the Templating Node, so this could be omitted
5	class MyTemplatingNode(TemplatingNode[BaseState, str]):
6	template = "The weather in {{ city }} is {{ weather }}."
7
8	inputs = {
9	"city": Inputs.city,
10	"weather": Inputs.weather,
11	}

1	MyTemplatingNode.Outputs(
2	result="""
3	Dear Jane Doe,
4
5	Thank you for your 5 years of service at Acme Corp.
6	Your dedication to Engineering has been invaluable.
7
8	Best regards,
9	HR Team
10	"""
11	)

1	from vellum import (
2	SearchRequestOptionsRequest,
3	SearchWeightsRequest,
4	SearchResultMergingRequest,
5	SearchFiltersRequest,
6	)
7	from vellum.workflows.nodes.displayable import SearchNode
8
9	class MySearchNode(SearchNode):
10	query = "Hello, world"
11	document_index = "my_document_index"
12	options = SearchRequestOptionsRequest(
13	limit=8,
14	weights=SearchWeightsRequest(
15	semantic_similarity=0.8,
16	keywords=0.2,
17	),
18	result_merging=SearchResultMergingRequest(
19	enabled=True
20	),
21	filters=SearchFiltersRequest(
22	external_ids=None,
23	metadata=None,
24	),
25	)
26	chunk_separator = "\n\n#####\n\n"

1	from vellum import SearchResult, SearchResultDocument
2
3	MySearchNode.Outputs(
4	text="Goodbye, world",
5	results=[
6	SearchResult(
7	text="Goodbye, world",
8	score=0.5,
9	keywords=[...],
10	document=SearchResultDocument(
11	id="<id>",
12	label="My Document",
13	external_id="<external-id>",
14	metadata={},
15	),
16	meta=None,
17	),
18	...
19	],
20	)

1	from vellum.workflows.nodes import InlinePromptNode
2	from vellum import (
3	ChatMessagePromptBlock,
4	JinjaPromptBlock,
5	)
6
7	class MyInlinePromptNode(InlinePromptNode):
8	ml_model = "gpt-4o"
9	blocks = [
10	ChatMessagePromptBlock(
11	chat_role="SYSTEM",
12	blocks=[
13	JinjaPromptBlock(
14	template="""\
15	Answer the user's questions based on the context provided below, if you don't know the answer say "Sorry I don't know"
16
17	Context
18	``
19	{{ context }}
20	``
21
22	Limit your answer to 250 words and provide a citation at the end of your answer\
23	""",
24	),
25	],
26	),
27	ChatMessagePromptBlock(
28	chat_role="USER",
29	blocks=[JinjaPromptBlock(template="{{ question }}")],
30	),
31	]
32	prompt_inputs = {"question": "What is the best AI development platform in the world?", "context": "..."}

1	from vellum.workflows.nodes import InlineSubworkflowNode
2	from vellum.workflows.inputs.base import BaseInputs
3	from vellum.workflows.state import BaseState
4	from vellum.workflows import BaseWorkflow
5
6	class SubworkflowInputs(BaseInputs):
7	query: str
8	max_results: int
9
10	class SubworkflowState(BaseState):
11	pass
12
13	class MySubworkflow(BaseWorkflow[SubworkflowInputs, SubworkflowState]):
14	# Define your subworkflow logic here
15	pass
16
17	class MyInlineSubworkflowNode(InlineSubworkflowNode):
18	subworkflow = MySubworkflow
19	subworkflow_inputs = {
20	"query": "search term",
21	"max_results": 10
22	}

1	# Outputs depend on the subworkflow's defined outputs
2	MyInlineSubworkflowNode.Outputs(
3	search_results=["Result 1", "Result 2", "Result 3"],
4	metadata={
5	"total_found": 100,
6	"processing_time": 0.5
7	}
8	)

1	from vellum.workflows.nodes import PromptDeploymentNode
2	from vellum.client import PromptDeploymentExpandMetaRequest, RawPromptExecutionOverridesRequest
3
4	class MyPromptDeploymentNode(PromptDeploymentNode):
5	deployment = "my_prompt_deployment"
6	prompt_inputs = {
7	"question": "What is the meaning of life?",
8	"chat_history": [
9	ChatMessage(role="USER", text="Hello!"),
10	ChatMessage(role="ASSISTANT", text="Hi there!"),
11	],
12	"context": {
13	"source": "philosophy_book",
14	"chapter": 42
15	}
16	}
17	release_tag = "production"
18	external_id = "unique-execution-id"
19	expand_meta = PromptDeploymentExpandMetaRequest(
20	model_name=True,
21	usage=True,
22	cost=True,
23	finish_reason=True,
24	latency=True,
25	deployment_release_tag=True,
26	prompt_version_id=True
27	)
28	metadata = {
29	"user_id": "123",
30	"session_id": "abc"
31	}

1	MyPromptDeploymentNode.Outputs(
2	text="happily",
3	results=[
4	StringVellumValue(value="h"),
5	StringVellumValue(value="app"),
6	StringVellumValue(value="ily"),
7	]
8	)

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}")

1	from vellum.workflows.nodes import CodeExecutionNode
2	from vellum import ChatMessage, CodeExecutionPackage
3	from vellum.workflows.state import BaseState
4	from typing import List, Dict
5
6	class MyCodeExecutionNode(CodeExecutionNode[BaseState, Dict[str, Any]]):
7	filepath = "./scripts/process_data.py"
8	code_inputs = {
9	"text": "Process this text",
10	"chat_history": [
11	ChatMessage(role="user", content="Hello"),
12	ChatMessage(role="assistant", content="Hi there!")
13	],
14	"config": {
15	"max_length": 100,
16	"temperature": 0.7
17	}
18	}
19	runtime = "PYTHON_3_11_6"
20	packages = [
21	CodeExecutionPackage(name="pandas", version="2.0.0"),
22	CodeExecutionPackage(name="numpy", version="1.24.0")
23	]

1	MyCodeExecutionNode.Outputs(
2	result={
3	"processed_text": "PROCESS THIS TEXT",
4	"history_length": 2,
5	"config_used": {
6	"max_length": 100,
7	"temperature": 0.7
8	}
9	},
10	log="INFO: Starting processing...\nINFO: Processing complete"
11	)

1	from vellum import ChatMessage
2	from vellum.workflows.nodes import GuardrailNode
3
4	class QualityCheckNode(GuardrailNode):
5	release_tag = "production"
6	metric_definition = "response_quality_metric" # or UUID("...")
7	# `metric_inputs` align with the custom Metric we've created in the Vellum UI
8	metric_inputs = {
9	"response": "This is the AI response to evaluate",
10	"chat_history": [
11	ChatMessage(role="USER", text="What's the weather?"),
12	ChatMessage(role="ASSISTANT", text="It's sunny today!")
13	],
14	"evaluation_criteria": {
15	"coherence": True,
16	"relevance": True,
17	"safety": True
18	}
19	}

1	QualityCheckNode.Outputs(
2	# `score` is a mandatory field for all Metrics and thus will be available on all Guardrail Nodes
3	score=0.92,
4	# `details` is a result of our custom Metric definition
5	details={
6	"coherence_score": 0.95,
7	"relevance_score": 0.88,
8	"safety_score": 0.93
9	}
10	)

1	from vellum.workflows import BaseWorkflow
2	from vellum.workflows.nodes import BaseNode
3	from vellum.workflows.state import BaseState
4
5
6	class SourceNode(BaseNode):
7	pass
8
9	class MiddleNode(BaseNode):
10	class Ports(BaseNode.Ports):
11	top = Port.on_if(SourceNode.Execution.count.less_than(1))
12	bottom = Port.on_else()
13
14	class TargetNode(BaseNode):
15	pass
16
17	class MyWorkflow(BaseWorkflow):
18	graph = SourceNode >> {
19	# here we use Ports to execute the SourceNode one more time
20	# before proceeding to the TargetNode
21	MiddleNode.Ports.top >> SourceNode,
22	MiddleNode.Ports.bottom >> TargetNode,
23	}
24
25	class Outputs(BaseWorkflow.Outputs):
26	result = TargetNode.Outputs

1	from vellum.workflows.nodes import BaseNode
2
3	class QuickNode(BaseNode):
4	class Outputs(BaseNode.Outputs):
5	prefix = "Hello"
6
7	class SlowNode(BaseNode):
8	class Outputs(BaseNode.Outputs):
9	suffix: str
10	def run(self) -> Outputs:
11	time.sleep(5)
12	return self.Outputs(suffix="World")
13
14	class MergeNode(BaseNode):
15	prefix = QuickNode.Outputs.prefix
16	suffix = SlowNode.Outputs.suffix
17	class Outputs(BaseNode.Outputs):
18	message: str
19	class Trigger(BaseNode.Trigger):
20	merge_strategy = MergeStrategy.AWAIT_ALL
21	def run(self) -> Outputs:
22	return self.Outputs(message=f"{self.prefix} {self.suffix}")

1	from vellum.workflows import BaseWorkflow
2	from vellum.workflows.nodes import ErrorNode
3	from vellum.workflows.inputs.base import BaseInputs
4	from vellum.workflows.state import BaseState
5
6	class Inputs(BaseInputs):
7	threshold: int
8
9	class StartNode(BaseNode):
10	class Ports(BaseNode.Ports):
11	success = Port.on_if(Inputs.threshold.greater_than(10))
12	fail = Port.on_else()
13
14	class SuccessNode(BaseNode):
15	class Outputs(BaseNode.Outputs):
16	result = Inputs.threshold
17
18	class FailNode(ErrorNode):
19	error = "Input threshold was too low"
20
21	class BasicErrorNodeWorkflow(BaseWorkflow[Inputs, BaseState]):
22	graph = {
23	StartNode.Ports.success >> SuccessNode,
24	StartNode.Ports.fail >> FailNode,
25	}
26	class Outputs(BaseWorkflow.Outputs):
27	final_value = SuccessNode.Outputs.result