Graph Export and Reload

Graph Export and Reload#

We support exporting the nodes, edges, entry nodes, and other information added to GraphBuilder into a json file via dump(). Subsequently, the configuration file can be loaded via load() in other scenarios to reconstruct GraphBuilder.

Note

Nodes in the graph and the components they depend on use pydantic for serialization and deserialization, and all components in the graph must be subclasses of BaseComponent.

The functions in the figure (such as the router for conditional edges, the multi_input_merge_strategy for multi-input nodes, etc.) depend on the serialization and deserialization functionality of cloudpickle.

Component Serialization/Deserialization#

All components in EvoFabric inherit BaseComponent. Additionally, we have separately defined serialization and deserialization methods for non-serializable input parameters.

After inheriting the BaseComponent class, the following will be supported:

  • Supports dynamic class creation via create() (the class is automatically registered with the factory during inheritance).

  • Support for serializing and overloading classes through model_dump_json() and model_validate().

Additionally, when the input parameter type of a component is defined as a virtual base class but the actual expected input is a subclass of this virtual base class, pydantic cannot automatically instantiate the subclass, instead using the subclass’s parameters to instantiate the base class, causing an error. Therefore, for such parameters, you can use Annotated to declare the annotation as FactoryTypeAdapter, which will save the class name to the __class_name__ field during model_dump and call ComponentFactory to create an instance of the corresponding class during reloading (note that this subclass must also inherit from BaseComponent).

Note

Subclasses also need to inherit BaseComponent to support factory instantiation.

Function serialization/deserialization#

The framework provides a serializer/deserializer based on cloudpickle, which can be obtained via get_func_serializer().

During the graph’s serialization/deserialization process, this instance will be used to serialize and deserialize parameters of function type. If you wish to save and reload the graph, ensure all custom functions support pickle.

If some modules are missing during deserialization, you can first register the missing modules through register_deserialize_modules().

from evofabric.core.factory import get_func_serializer

def custom_function():
    ...


serialized = get_func_serializer().serialize(custom_function)

deserialized_function = get_func_serializer().deserialize(serialized)

Note

If you need to replace the current function serialization/deserialization method, you can register an instance that conforms to the FunctionSerializerProto interface via register_deserialize_modules.

Using custom components in the diagram#

When custom components are placed in the diagram, the following requirements must be followed to support saving and reloading the diagram:

  • Inherit BaseComponent to obtain factory support and BaseModel features.

  • Use the pydantic style to define class input parameters.

  • Explicitly declare the serialization and deserialization methods for function types and other non-serializable parameters. (For the pickle serialization of functions, you can use get_func_serializer() to obtain the function serializer/deserializer.)

Example:

import json
from typing import Annotated, Any, Callable
from pydantic import Field, field_serializer, field_validator
from evofabric.core.clients import ChatClientBase, OpenAIChatClient
from evofabric.core.factory import BaseComponent, FactoryTypeAdapter, get_func_serializer


class CustomNode(BaseComponent):

    client: Annotated[ChatClientBase, FactoryTypeAdapter, Field(description="a parameter need FactoryTypeAdapter")]

    custom_router: Callable = Field(description="a router that need specific (de)serialize method")

    @field_serializer("custom_router")
    def serialize_stream_parser(self, _value: Callable) -> str:
        return get_func_serializer().serialize(_value)

    @field_validator('custom_router', mode='before')
    @classmethod
    def deserialize_stream_parser(cls, v: Any) -> Callable:
        if callable(v):
            return v
        return get_func_serializer().deserialize(v)

    async def __call__(self, *args, **kwargs):
        ...


def custom_router():
    # do something
    ...


node = CustomNode(client=OpenAIChatClient(model="your-model-name"), custom_router=custom_router)

# dump agent to json string
node_config = node.model_dump_json()

# reload node from json string
reload_node = CustomNode.model_validate(json.loads(node_config))
Contents