evofabric.core.factory

evofabric.core.factory#

Factory#

class evofabric.core.factory.ComponentFactory[源代码]#

用于注册和创建组件实例的工厂类。

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

根据给定的名称和参数创建一个类实例。

参数:

name (str) -- 组件类的注册名称。
kwargs (Any) -- 传递给组件类构造函数的关键字参数。

返回:

创建的组件实例。

返回类型:

BaseComponent

抛出:

ValueError -- 如果未找到指定名称的组件。

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

将一个类注册到工厂中。

参数:

name (str) -- 注册时使用的名称。
component_cls (Type[BaseComponent]) -- 要注册的组件类。

抛出:

ValueError -- 如果该名称已被注册。

is_registered(cls, name: str) → bool[源代码]#

检查指定名称是否已注册。

参数:: name (str) -- 要检查的组件名称。
返回:: 如果名称已注册返回 True，否则返回 False。
返回类型:: bool

class evofabric.core.factory.BaseComponent(BaseModel)[源代码]#

所有组件类的基类，继承自 pydantic.BaseModel，提供文档生成、懒加载实例等功能。

继承了该基类的子类，会自动注册到 ComponentFactory 中，允许后续通过工厂创建该类的实例。

class evofabric.core.factory.FactoryTypeAdapter[源代码]#

一个用于 Pydantic V2 的类型适配器，支持将字典序列化/反序列化为 BaseComponent 及其子类实例。该类通过 __class_name__ 字段识别具体组件类型，并借助 ComponentFactory 进行实例创建。

__get_pydantic_core_schema__(cls, source_type, handler) → core_schema.CoreSchema[源代码]#

生成 Pydantic 核心 schema，用于支持从字典到 BaseComponent 实例的验证和序列化。

参数:

source_type (Type) -- 被装饰的原始类型。
handler (GetCoreSchemaHandler) -- Pydantic 提供的 schema 处理函数。

返回:

返回一个支持字典与 BaseComponent 之间相互转换的 schema。

返回类型:

core_schema.CoreSchema

Function Serializer#

class evofabric.core.factory.FunctionSerializerProto[源代码]#

定义了函数序列化与反序列化方法的协议类，用于在模块中存储和重新加载 DSL 文件时处理 Python 函数句柄。

serialize(self, obj: Any) → str[源代码]#

将给定对象序列化为字符串。

参数:: obj (Any) -- 需要序列化的对象。
返回:: 序列化后的字符串。
返回类型:: str

deserialize(self, s: str) → Any[源代码]#

从字符串中反序列化出原始对象。

参数:: s (str) -- 序列化后的字符串。
返回:: 反序列化后的对象。
返回类型:: Any

class evofabric.core.factory.FunctionSerializerCloudPickle[源代码]#

使用 cloudpickle 实现的函数序列化器，支持对复杂 Python 函数对象（包括闭包、lambda 等）进行序列化与反序列化。

serialize(self, function: Callable) → str[源代码]#

将一个可调用对象序列化为 Base64 编码的字符串。

参数:: function (Callable) -- 需要序列化的函数或可调用对象。
返回:: 序列化后的 Base64 字符串。
返回类型:: str

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

从 Base64 编码的字符串中反序列化出函数对象。

参数:

string (str) -- 已序列化的函数字符串。
required_modules (Optional[List[str]]) -- 反序列化前需要导入的模块列表，用于确保依赖类型正确加载。

返回:

反序列化后的函数对象。

返回类型:

Callable

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

注册反序列化过程中可能需要用到的模块列表。默认包含了：

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"
]

参数:: modules (List[str]) -- 模块名称列表，每个名称应为合法的 Python 模块路径字符串。
返回:: 无返回值。
返回类型:: None

使用示例：

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

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

设置全局使用的函数序列化器实现。

参数:: impl (Optional[FunctionSerializerProto]) -- 一个实现了 FunctionSerializerProto 协议的对象；若为 None，则默认使用 FunctionSerializerCloudPickle。
返回:: 无返回值。
返回类型:: None

evofabric.core.factory.get_func_serializer() → FunctionSerializerProto[源代码]#

获取当前全局使用的函数序列化器实例。

返回:: 当前的函数序列化器实现。
返回类型:: FunctionSerializerProto

State Schema Serializer#

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

将 Pydantic 的 BaseModel 类型或 TypedDict 类型转换为包含注解信息的字典结构，便于序列化和传输。

参数:: schema (Type[Union[BaseModel, Dict]]) -- 要转换的 BaseModel 或 TypedDict 类型。
返回:: 包含类型名称、类型类别（BaseModel 或 TypedDict）及字段详细信息的字典。
返回类型:: Dict

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

根据包含注解信息的字典结构，还原出对应的 BaseModel 或 TypedDict 类型。

参数:: schema_info (Dict) -- 由 dump_schema_annotated_info() 生成的类型描述信息。
返回:: 还原后的 BaseModel 或 TypedDict 类型。
返回类型:: Type[Union[BaseModel, Dict]]

class evofabric.core.factory.StateSchemaSerializable[源代码]#: 提供对状态 schema 的序列化与反序列化功能，通过继承该类，可以自动为子类添加对 state_schema: type[Union[BaseModel, TypedDict]] 属性的序列化、反序列化方法。

Utils#

evofabric.core.factory.is_typeddict(tp) → bool[源代码]#

判断给定类型是否为 TypedDict。

参数:: tp (type) -- 待判断的类型。
返回:: 如果是 TypedDict 类型则返回 True，否则返回 False。
返回类型:: bool

evofabric.core.factory.is_basemodel(typ) → bool[源代码]#

判断给定类型是否为 Pydantic 的 BaseModel 子类。

参数:: typ (type) -- 待判断的类型。
返回:: 如果是 BaseModel 的子类则返回 True，否则返回 False。
返回类型:: bool

evofabric.core.factory.is_dataclass(typ) → bool[源代码]#

判断给定类型是否为 Pyd 3.7+ 标准库中的 dataclass 或 Pydantic 模型（通过 __pydantic_config__ 判断）。

参数:: typ (type) -- 待判断的类型。
返回:: 如果是 dataclass 或 Pydantic 模型则返回 True，否则返回 False。
返回类型:: bool

evofabric.core.factory.strip_annotated(tp)[源代码]#

若类型被 typing.Annotated 包装，则返回其原始类型；否则原样返回。

参数:: tp (type) -- 可能被 Annotated 包装的类型。
返回:: 原始未包装的类型。
返回类型:: type

evofabric.core.factory.deep_dump(obj: Any) → Any[源代码]#

递归地将对象中所有值转换为字典形式。支持 BaseModel、dict、list 和 tuple 类型。

参数:: obj (Any) -- 需要转换的对象。
返回:: 转换后的嵌套字典结构。
返回类型:: Any

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

为 BaseModel 或 TypedDict 填充默认字段值，并可选地合并额外字段。

参数:

model_or_cls (type[BaseModel] | type[TypedDict]) -- 目标模型或类型。
extra (Dict[str, Any] | None) -- 可选的额外字段值字典。

返回:

包含默认值和额外字段的完整字段字典。

返回类型:

Dict[str, Any]

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

安全地从 BaseModel 等对象或字典中获取属性值。

参数:

data (Any) -- 数据源，可以是对象或字典。
attr (str) -- 属性名称。
default (Any) -- 默认值，当属性不存在时返回该值。

返回:

获取到的属性值或默认值。

返回类型:

Any

evofabric.core.factory.safe_set_attr(data, attr, value)[源代码]#

安全地为 BaseModel 等对象或字典设置属性值。

参数:

data (Any) -- 数据源，可以是对象或字典。
attr (str) -- 属性名称。
value (Any) -- 要设置的属性值。

evofabric.core.factory.safe_convert_to_schema(data, schema)[源代码]#

将数据安全地转换为目标 schema 类型（如 Pydantic 模型）。

参数:

data (Any) -- 输入数据，可以是字典、对象或 BaseModel 实例。
schema (type) -- 目标 schema 类型，应为 Pydantic 模型类。

返回:

转换后的 schema 实例或字典。

返回类型:

BaseModel | dict

evofabric.core.factory

目录

evofabric.core.factory#

Factory#

Function Serializer#

State Schema Serializer#

Utils#