Serializing / deserializing a workflow to write to DB

[vaibhav-46]

I have been using Dynamiq for a while and it's been pretty useful to build micro-agents. One thing that I tried to implement but failed, is a way to create a workflow, serialize it to store it in the DB and then be able to deserialize it to get back the same state.

A very simple example :

 workflow = Workflow()

 # Initialize shared connections
 firecrawl_connection = Firecrawl()

    # Initial URL crawling and market extraction
    scraper = FirecrawlTool(connection=firecrawl_connection)
    context_extractor = (
        OpenAI(
            model="gpt-4o",
            prompt=Prompt(
                messages=[
                    Message(
                        role=MessageRole.USER,
                        content=(
                           "Write a summary on the performance of each player"
                        ),
                    )
                ]
            ),
        )
        .inputs(content=scraper.outputs.content)
        .depends_on(scraper)
    )
    workflow.add_nodes([scraper, context_extractor])
    workflow_dict = workflow.model_dump(serialize_as_any=True)
    workflow = Workflow.model_validate(workflow_dict)

This keeps failing because the NodeType is not retained when it's being re-created from the workflow_dict. Is there a way that I might not have found to get this to work? I do see the to_dict function, but I am not sure if there is a way to restore the workflow from the output of the to_dict function.

[acoola]

@vaibhav-46 Thanks for the great question.
The simplest solution is to use Dynamiq’s YAML serialization rather than converting the workflow to a dict because the dict-based approach doesn't properly handle all nodes/connections/clients.

from dynamiq.serializers.loaders.yaml import WorkflowYAMLLoader

WF_YAML_FILE_PATH = "wf.yaml"

# Create workflow 
workflow = Workflow()
workflow.add_nodes([...])

# Save the workflow to a YAML file.
workflow.to_yaml_file(file_path=WF_YAML_FILE_PATH)

# Later, load the YAML back and re-create the workflow.
wf_data = WorkflowYAMLLoader.load(file_path=WF_YAML_FILE_PATH, init_components=True)
workflow_from_yaml = Workflow.from_yaml_file_data(file_data=wf_data)