FunctionalFlowStep

LLMFlow module for the FunctionalFlowstep class that can be used to run a given function within a flow.

FunctionalFlowStep

Bases: BaseFlowStep

Represents a functional flow step that runs a function. The function must take a dictionary of strings as input and return a string(like regular flow steps)

Parameters:

Name	Type	Description	Default
name	str	The name of the flow step.	required
fn	Callable	The function to run.	required
required_keys	list[str]	A list of required keys.	required
output_key	str	The key to use for the output.	required
callbacks	list[Callback]	List of callback instances. Defaults to None.	None

Attributes:

Name	Type	Description
required_keys	set[str]	The keys required for the flow step to run.
fn	Callable[[dict[str, str]], str]	The function to be run.

Source code in llmflows/flows/functional_flowstep.py

class FunctionalFlowStep(BaseFlowStep):
    """
    Represents a functional flow step that runs a function. The function must take
        a dictionary of strings as input and return a string(like regular flow steps)

    Args:
        name (str): The name of the flow step.
        fn (Callable): The function to run.
        required_keys (list[str]): A list of required keys.
        output_key (str): The key to use for the output.
        callbacks (list[Callback], optional): List of callback instances. Defaults to
            None.

    Attributes:
        required_keys (set[str]): The keys required for the flow step to run.
        fn (Callable[[dict[str, str]], str]): The function to be run.
    """

    def __init__(
        self,
        name: str,
        flowstep_fn: Callable[..., str],
        output_key: str,
        callbacks: Union[list[BaseCallback], None] = None,
    ):
        super().__init__(name, output_key, callbacks)
        self.flowstep_fn = flowstep_fn
        self.required_keys = inspect.getfullargspec(self.flowstep_fn).args

    def generate(
        self, inputs: dict[str, Any]
    ) -> tuple[Any, Union[dict, None], Union[dict, None]]:
        """
        Executes the function with the provided inputs.

        Args:
            inputs (dict[str, Any]): Input parameters as a dictionary.

        Returns:
            The result of the function call, followed by two None values (for call
                data and config, which are not applicable in this case).
        """
        # Get the argument names of fn
        fn_args = inspect.getfullargspec(self.flowstep_fn).args

        # Create a new dictionary from inputs containing only the keys that fn requires
        filtered_inputs = {
            key: value for key, value in inputs.items() if key in fn_args
        }

        result = self.flowstep_fn(**filtered_inputs)

        if not isinstance(result, str):
            raise TypeError(
                f"Return value must be of type str, but got {type(result).__name__}"
            )

        return result, None, None

generate(inputs)

Executes the function with the provided inputs.

Parameters:

Name	Type	Description	Default
inputs	dict[str, Any]	Input parameters as a dictionary.	required

Returns:

Type	Description
tuple[Any, Union[dict, None], Union[dict, None]]	The result of the function call, followed by two None values (for call data and config, which are not applicable in this case).

Source code in llmflows/flows/functional_flowstep.py

def generate(
    self, inputs: dict[str, Any]
) -> tuple[Any, Union[dict, None], Union[dict, None]]:
    """
    Executes the function with the provided inputs.

    Args:
        inputs (dict[str, Any]): Input parameters as a dictionary.

    Returns:
        The result of the function call, followed by two None values (for call
            data and config, which are not applicable in this case).
    """
    # Get the argument names of fn
    fn_args = inspect.getfullargspec(self.flowstep_fn).args

    # Create a new dictionary from inputs containing only the keys that fn requires
    filtered_inputs = {
        key: value for key, value in inputs.items() if key in fn_args
    }

    result = self.flowstep_fn(**filtered_inputs)

    if not isinstance(result, str):
        raise TypeError(
            f"Return value must be of type str, but got {type(result).__name__}"
        )

    return result, None, None