FunctionNode
Node for executing custom Python functions within workflows.
FunctionNode allows integration of arbitrary Python functions into the Liman execution graph. Functions can be sync or async.
Attributes
attributespec_type= FunctionNodeSpecattributestate_type= FunctionNodeStateattributeregistry= registryFunctions
func__init____init__(self, /, spec, registry, *, initial_data=None, yaml_path=None, strict=False, default_lang='en', fallback_lang='en') -> NoneInitialize function node with specification and registry.
Args: spec: Function node specification defining the target function registry: Component registry for dependency resolution initial_data: Optional initial data for the component yaml_path: Optional path to the YAML file this node was loaded from strict: Whether to enforce strict validation default_lang: Default language code for localization fallback_lang: Fallback language code when default is unavailable
paramselfparamspecFunctionNodeSpecparamregistryRegistryparaminitial_datadict[str, Any] | None= Noneparamyaml_pathstr | None= Noneparamstrictbool= Falseparamdefault_langstr= 'en'paramfallback_langstr= 'en'Returns
Nonefunccompilecompile(self) -> NoneCompile the function node for execution.
Prepares the node for execution. Currently performs basic validation and sets the compiled flag.
Raises: LimanError: If the node is already compiled
paramselfReturns
Nonefuncset_funcset_func(self, /, func) -> NoneManually set the function for this function node.
Args: func: Callable function to execute during invocation
paramselfparamfuncCallable[..., Any]Returns
Nonefuncinvokeinvoke(self, /, input_, **kwargs) -> AnyExecute the function node with provided input.
Extracts function arguments from input and executes the function. Handles both synchronous and asynchronous functions.
Args: input_: Input data to pass to the function **kwargs: Additional keyword arguments
Returns: Result of function execution
paramselfparaminput_AnyparamkwargsAny= {}Returns
Anyfuncget_new_stateget_new_state(self) -> FunctionNodeStateCreate new state instance for this function node.
Returns: Fresh FunctionNodeState with empty message history
paramselfReturns
FunctionNodeStatefunc_extract_function_args_extract_function_args(self, /, args_dict) -> dict[str, Any]Extract function arguments based on function signature from provided args dict.
Args: args_dict: Dictionary containing all available arguments
Returns: Dictionary with only the arguments that match function signature
paramselfparamargs_dictdict[str, Any] | NoneReturns
dict[str, Any]Last updated on