Serializing/Deserializing Workflows

[pjirsa]

Shouldn't I be able to create a workflow from the json output generated by Workflow.to_json() by using Workflow.from_json()?

My sample code is producing the following error:

Exception has occurred: TypeError
Workflow.__init__() missing 2 required positional arguments: 'start_executor' and 'runner_context'
  File "C:\Users\pjirsa\Dev\agent-framework\python\samples\getting_started\workflows\_start-here\step1_executors_and_edges.py", line 126, in main
    workflow2 = Workflow.from_json(wf_json)
  File "C:\Users\pjirsa\Dev\agent-framework\python\samples\getting_started\workflows\_start-here\step1_executors_and_edges.py", line 141, in <module>
    asyncio.run(main())
    ~~~~~~~~~~~^^^^^^^^
TypeError: Workflow.__init__() missing 2 required positional arguments: 'start_executor' and 'runner_context'

Here is the code I'm using:

async def main():
    """Build and run a simple 2-step workflow using the fluent builder API."""

    upper_case = UpperCase(id="upper_case_executor")

    # Build the workflow using a fluent pattern:
    # 1) add_edge(from_node, to_node) defines a directed edge upper_case -> reverse_text
    # 2) set_start_executor(node) declares the entry point
    # 3) build() finalizes and returns an immutable Workflow object
    workflow = WorkflowBuilder().add_edge(upper_case, reverse_text).set_start_executor(upper_case).build()

    wf_json = workflow.to_json()
    # Run the workflow by sending the initial message to the start node.
    # The run(...) call returns an event collection; its get_outputs() method
    # retrieves the outputs yielded by any terminal nodes.
    events = await workflow.run("hello world")
    print(events.get_outputs())
    # Summarize the final run state (e.g., IDLE)
    print("Workflow result:", events.get_final_state())

    workflow2 = Workflow.from_json(wf_json)
    events2 = await workflow2.run("hello json")

    print("Workflow2 result:", events2.get_final_state())

    """
    Sample Output:

    ['DLROW OLLEH']
    Final state: WorkflowRunState.IDLE
    """

[moonbox3]

Hi @pjirsa,

The to_json() and from_json() methods on Workflow are not designed as a public serialization API for workflow persistence.

Why it fails:

to_json() serializes the workflow's structural definition (executors, edges, start node) for observability/tracing purposes. The from_json() is inherited from DictConvertible which does a naive cls(**data), but Workflow.__init__ requires live Python objects (Executor, RunnerContext) that cannot be round-tripped through JSON.

Alternatives:

1. Workflow state persistence: use the checkpointing API:

   workflow = WorkflowBuilder().with_checkpointing(storage).build()

2. Declarative definitions from config (coming soon - in PR): we will soon have support for declarative workflows in Python. At that time, one can use agent_framework.declarative to load workflows from YAML.
3. Workflow introspection only: to_json() / to_dict() work correctly for logging/debugging the structure.

An improvement on our side: the inherited from_json() should be overridden to raise NotImplementedError with guidance, since the current behavior is misleading.

[pjirsa]

@moonbox3 Thanks for the reply. I did notice after I posted that the from_json() and from_dict() functions were coming from a base class.
My project aims to allow users to dynamically build workflows using the Agent Framework. So, I'm eagerly awaiting option 2 😺