Code Execution Node

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 )

from vellum.workflows.nodes import CodeExecutionNode
from vellum import ChatMessage, CodeExecutionPackage
from vellum.workflows.state import BaseState
from typing import List, Dict
class MyCodeExecutionNode(CodeExecutionNode[BaseState, Dict[str, Any]]):
    filepath = "./scripts/process_data.py"
    code_inputs = {
        "text": "Process this text",
        "chat_history": [
            ChatMessage(role="user", content="Hello"),
            ChatMessage(role="assistant", content="Hi there!")
        ],
        "config": {
            "max_length": 100,
            "temperature": 0.7
        }
    }
    runtime = "PYTHON_3_11_6"
    packages = [
        CodeExecutionPackage(name="pandas", version="2.0.0"),
        CodeExecutionPackage(name="numpy", version="1.24.0")
    ]

MyCodeExecutionNode.Outputs(
    result={
        "processed_text": "PROCESS THIS TEXT",
        "history_length": 2,
        "config_used": {
            "max_length": 100,
            "temperature": 0.7
        }
    },
    log="INFO: Starting processing...\nINFO: Processing complete"
)

vellum.workflows.nodes.CodeExecutionNode

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

Important: Your code file must contain a main() function with parameters that match the names of the node’s inputs. Without this function, you’ll get a NameError: name 'main' is not defined error.

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	)