I'm also getting this error when trying to execute...
# ask-the-community
f
I'm also getting this error when trying to execute a workflow. I've checked the types, they seem to be consistent.
Copy code
TypeTransformerFailedError: Type of Val 'typing.Dict' is not an instance of typing.Dict[str, str]
e
Hello! I am from the community and happy to help debug. If you can paste a minimal working example that reproduces the error, I can work through it with you.
f
Thank you @Evan Sadler
Copy code
class FlyteRemote:
    def __init__(self):
        ...

    def _register_execute_workflow(
        self,
        wf: flytekit.core.workflow.WorkflowBase,
        inputs: Dict,
    ) -> flytekit.remote.executions.FlyteWorkflowExecution:

        self.remote.register_workflow(wf, self.serialization_settings)

        fetched_wf = self.remote.fetch_workflow(
            name=wf.name,
            version=self.version,
        )

        execution_response = self.remote.execute(
            entity=fetched_wf,
            inputs=inputs,
            version=self.version,
            image_config=self.image_config,
        )

        return execution_response

    def run_regular_simulation_wf(
            self,
            inputs: Dict,
            ) -> flytekit.remote.executions.FlyteWorkflowExecution:

        wf_response = self._register_execute_workflow(wf=regular_simulation_cpu_wf, inputs=inputs)
Copy code
@workflow
def regular_simulation_cpu_wf(
    param1: Challenge,
    param2: List[int],
    param3: str,
    param4: str,
    param5: module.Model,
    param6: module.Explorer,
    param7: module.Landscape,
    param8: int,
    param9: str,
):

    return simulation_cpu_task(
        param1=param1,
        param2=param2,
        param3=param3,
        param4=param4,
        param5=param5,
        param6=param6,
        param7=param7,
        param8=param8,
        param9=param9,
    )
e
This looks like a type issue with the Flyte workflow and task. I am noticing there is no typing for the output of the workflow, but I need to see •
simulation_cpu_task
to be able to judge more
Also using
pyflyte run
from the command line might help isolate if the issues faster. Because you can nest workflows, you can create one for testing that wraps
regular_simulation_cpu_wf
and pass in defaults. Alternatively you can call
regular_simulation_cpu_wf
from inside a notebook too like normal python.
Copy code
@workflow
def test_regular_simulation_cpu_wf():
   regular_simulation_cpu_wf([pass in values])
f
@Evan Sadler the
regular_simulation_cpu_wf
is a one-task workflow, so the parameters of that workflow is the same as the parameters of the task. Below is the snippet of the
simulation_cpu_task
Copy code
@task(requests=Resources(cpu="1", mem="1Gi"))
def simulation_cpu_task(
    param1: Challenge,
    param2: List[int],
    param3: str,
    param4: str,
    param5: module.Model,
    param6: module.Explorer,
    param7: module.Landscape,
    param8: int,
    param9: str,
) -> List[Campaign]:
    """CPU simulation task"""
    simulation = module.Simulation(
        param1=param1,
        param2=param2,
        param3=param3,
        param4=param4,
        param5=param5,
        param6=param6,
        param7=param7,
        param8=param8,
        param9=param9,
    )
    experiments = simulation.run_sim()
    return experiments
Thank you very much. I'm currently testing out your suggestions to see if they work. I will revert.
106 Views