evofabric.core.factory

evofabric.core.factory#

Factory#

class evofabric.core.factory.ComponentFactory[source]#

Factory class for registering and creating component instances.

create(cls, name: str, /, **kwargs) → BaseComponent[source]#

Create a class instance using the given name and parameters.

Parameters:

name (str) – Component class registration name
kwargs (Any) – Keyword arguments passed to the component class constructor.

Returns:

Created component instance.

Return type:

BaseComponent

Raises:

ValueError – If a component with the specified name is not found.

register(cls, name: str, component_cls: Type[BaseComponent]) → None[source]#

Parameters:

name (str) – The name used during registration.
component_cls (Type[BaseComponent]) – component class to register

Raises:

ValueError – If the name has been registered.

is_registered(cls, name: str) → bool[source]#

Check if the specified name has been registered.

Parameters:: name (str) – Component name to be checked.
Returns:: If the name is already registered, return True, otherwise return False.
Return type:: bool

class evofabric.core.factory.BaseComponent(BaseModel)[source]#

The base class for all component classes, inheriting from pydantic.BaseModel, provides features such as documentation generation and lazy-loaded instances.

Subclasses inheriting from this base class will be automatically registered in ComponentFactory, allowing subsequent creation of instances of the class through the factory.

class evofabric.core.factory.FactoryTypeAdapter[source]#

A type adapter for Pydantic V2 that supports serializing/deserializing dictionaries into instances of BaseComponent and its subclasses. This class identifies the specific component type through the __class_name__ field and uses ComponentFactory for instance creation.

__get_pydantic_core_schema__(cls, source_type, handler) → core_schema.CoreSchema[source]#

Generate a Pydantic core schema to support validation and serialization from dictionary to BaseComponent instances.

Parameters:

source_type (Type) – Decorated primitive type.
handler (GetCoreSchemaHandler) – Pydantic-provided schema processing functions.

Returns:

Return a schema that supports bidirectional conversion between dictionary and BaseComponent.

Return type:

core_schema.CoreSchema

Function Serializer#

class evofabric.core.factory.FunctionSerializerProto[source]#

A protocol class defining methods for serializing and deserializing functions, used to handle Python function handles when storing and reloading DSL files within a module.

serialize(self, obj: Any) → str[source]#

Serialize the given object into a string.

Parameters:: obj (Any) – Object to be serialized.
Returns:: Serialized string.
Return type:: str

deserialize(self, s: str) → Any[source]#

Deserialize the original object from a string.

Parameters:: s (str) – Serialized string.
Returns:: Deserialized object.
Return type:: Any

class evofabric.core.factory.FunctionSerializerCloudPickle[source]#

Function serializer implemented using cloudpickle supports serialization and deserialization of complex Python function objects (including closures, lambdas, etc.).

serialize(self, function: Callable) → str[source]#

Serialize a callable object into a Base64-encoded string.

Parameters:: function (Callable) – Functions or callable objects that need to be serialized.
Returns:: Serialized Base64 string.
Return type:: str

deserialize(self, string: str, required_modules: List[str] | None = None) → Callable[source]#

Deserialize a function object from a Base64-encoded string.

Parameters:

string (str) – Serialized function string.
required_modules (Optional[List[str]]) – List of modules to be imported before deserialization to ensure correct loading of dependent types.

Returns:

Deserialized function object.

Return type:

Callable

evofabric.core.factory.register_deserialize_modules(modules: List[str]) → None[source]#

List of modules to register that may be needed during deserialization. Default includes:

DESERIALIZER_MODULES = [
    "evofabric.logger",
    "evofabric.core.agent",
    "evofabric.core.clients",
    "evofabric.core.factory",
    "evofabric.core.graph",
    "evofabric.core.mem",
    "evofabric.core.multi_agent",
    "evofabric.core.tool",
    "evofabric.core.trace",
    "evofabric.core.typing",
    "evofabric.core.vectorstore"
]

Parameters:: modules (List[str]) – List of module names, each name should be a valid Python module path string.
Returns:: No return value.
Return type:: None

Usage Example:

register_deserialize_modules([
    "evofabric.logger",
    "evofabric.core.agent",
    "evofabric.core.clients",
])

evofabric.core.factory.set_func_serializer(impl: FunctionSerializerProto | None) → None[source]#

Set the global function serializer implementation.

Parameters:: impl (Optional[FunctionSerializerProto]) – An object that implements the FunctionSerializerProto protocol; if it is None, the default is to use FunctionSerializerCloudPickle.
Returns:: No return value.
Return type:: None

evofabric.core.factory.get_func_serializer() → FunctionSerializerProto[source]#

Get the current global function serializer instance.

Returns:: Current function serializer implementation.
Return type:: FunctionSerializerProto

State Schema Serializer#

evofabric.core.factory.dump_schema_annotated_info(schema: Type[BaseModel | Dict]) → Dict[source]#

Convert Pydantic’s BaseModel type or TypedDict type into a dictionary containing annotation information, facilitating serialization and transmission.

Parameters:: schema (Type[Union[BaseModel, Dict]]) – The BaseModel or TypedDict type to be converted.
Returns:: A dictionary containing the type name, type category (BaseModel or TypedDict), and field details.
Return type:: Dict

evofabric.core.factory.load_schema_annotated_info(schema_info: Dict) → Type[BaseModel | Dict][source]#

Based on the dictionary structure containing annotation information, reconstruct the corresponding BaseModel or TypedDict type.

Parameters:: schema_info (Dict) – Type description information generated by dump_schema_annotated_info().
Returns:: The restored BaseModel or TypedDict type.
Return type:: Type[Union[BaseModel, Dict]]

class evofabric.core.factory.StateSchemaSerializable[source]#: Provides serialization and deserialization functionality for the state schema. By inheriting this class, serialization and deserialization methods for the state_schema: type[Union[BaseModel, TypedDict]] property are automatically added to subclasses.

Utils#

evofabric.core.factory.is_typeddict(tp) → bool[source]#

Check whether the given type is TypedDict.

Parameters:: tp (type) – Type to be determined.
Returns:: Return True if it is a TypedDict type, otherwise return False.
Return type:: bool

evofabric.core.factory.is_basemodel(typ) → bool[source]#

Determine whether the given type is a subclass of Pydantic’s BaseModel.

Parameters:: typ (type) – Type to be determined.
Returns:: Returns True if it is a subclass of BaseModel, otherwise returns False.
Return type:: bool

evofabric.core.factory.is_dataclass(typ) → bool[source]#

Determine if the given type is a dataclass in the Pyd 3.7+ standard library or a Pydantic model (determined via __pydantic_config__).

Parameters:: typ (type) – Type to be determined.
Returns:: Return True if it is a dataclass or Pydantic model, otherwise return False.
Return type:: bool

evofabric.core.factory.strip_annotated(tp)[source]#

If the type is wrapped by typing.Annotated, return its original type; otherwise, return it as is.

Parameters:: tp (type) – The types that may be wrapped by Annotated.
Returns:: Primitive unwrapped type.
Return type:: type

evofabric.core.factory.deep_dump(obj: Any) → Any[source]#

Recursively convert all values in the object to dictionary form. Supports BaseModel, dict, list, and tuple types.

Parameters:: obj (Any) – The object to be converted.
Returns:: Converted nested dictionary structure.
Return type:: Any

evofabric.core.factory.fill_defaults(model_or_cls: type[BaseModel] | type[TypedDict], *, extra: Dict[str, Any] | None = None) → Dict[str, Any][source]#

Populate default field values for BaseModel or TypedDict, and optionally merge additional fields.

Parameters:

model_or_cls (type[BaseModel] | type[TypedDict]) – Target model or type.
extra (Dict[str, Any] | None) – Dictionary of optional additional field values.

Returns:

Complete field dictionary containing default values and additional fields.

Return type:

Dict[str, Any]

evofabric.core.factory.safe_get_attr(data, attr, default=MISSING)[source]#

Safely retrieve attribute values from objects like BaseModel or dictionaries.

Parameters:

data (Any) – Data source, can be an object or a dictionary.
attr (str) – Attribute name.
default (Any) – Default value, returns this value when the attribute does not exist.

Returns:

Retrieved attribute value or default value.

Return type:

Any

evofabric.core.factory.safe_set_attr(data, attr, value)[source]#

Safely set property values for objects such as BaseModel or dictionaries.

Parameters:

data (Any) – Data source, can be an object or a dictionary.
attr (str) – Attribute name.
value (Any) – The attribute value to be set.

evofabric.core.factory.safe_convert_to_schema(data, schema)[source]#

Securely convert data to the target schema type (e.g., Pydantic models).

Parameters:

data (Any) – Input data, can be a dictionary, object, or BaseModel instance.
schema (type) – Target schema type, should be a Pydantic model class.

Returns:

Converted schema instance or dictionary.

Return type:

BaseModel | dict

evofabric.core.factory

Contents

evofabric.core.factory#

Factory#

Function Serializer#

State Schema Serializer#

Utils#