ahvn.ukf package

ahvn.ukf.default_trigger(kl, **kwargs)[源代码]

Default trigger used as a noop predicate.

The default trigger always returns True and can be used as a safe placeholder when no custom trigger function is supplied.

参数:

  • kl (BaseUKF) -- Knowledge object being evaluated (unused).

  • **kwargs -- Additional context parameters (ignored).

返回:

Always True.

返回类型:

bool

ahvn.ukf.default_composer(kl, **kwargs)[源代码]

Default content composer that returns the raw content string.

This function is intended as a safe default for the content_composers mapping.

参数:

  • kl (BaseUKF) -- Knowledge object to compose (only kl.content is used).

  • **kwargs -- Additional composition parameters (ignored).

返回:

The content attribute of kl.

返回类型:

str

class ahvn.ukf.BaseUKF(*, name, notes='', short_description='', description='', type='general', version='v0.1.0', version_notes='', variant='default', variant_notes='', content='', content_resources=<factory>, content_composers=<factory>, source='unknown', parents=<factory>, creator='unknown', owner='unknown', workspace='unknown', collection='general', tags=<factory>, synonyms=<factory>, triggers=<factory>, priority=0, related=<factory>, auths=<factory>, timefluid=False, timestamp=<factory>, last_verified=<factory>, expiration=-1, inactive_mark=False, metadata=<factory>, profile=<factory>)[源代码]

基类：BaseModel

Base model for knowledge items in the Universal Knowledge Framework (UKF).

This Pydantic model provides a comprehensive structure for representing knowledge units with detailed metadata, content management, provenance tracking, and relationship modeling. Each attribute includes comprehensive descriptions to guide users unfamiliar with UKF in correctly filling knowledge records from source documents and information.

The model is organized into logical sections:

Metadata: Core identification and classification fields (name, type, version, etc.) Content: The actual knowledge and supporting structured data Provenance: Origin tracking and ownership information Retrieval: Search, classification, and access control attributes Relationships: Connections to other knowledge items and permissions Lifecycle: Time-sensitive attributes for expiration and deprecation System: Extensible fields for application-specific data and runtime statistics

All field descriptions include practical examples and clear guidance on appropriate values, making this model self-documenting for users creating knowledge items from documents, conversations, or other information sources.

For detailed attribute guidance, refer to the comprehensive field descriptions in the model definition below.

参数:
type_default: ClassVar[str] = 'general'
tags_default: ClassVar[Set[str]] = {}
name: UKFMediumTextType
notes: UKFMediumTextType
short_description: UKFMediumTextType
description: UKFMediumTextType
type: UKFShortTextType
version: UKFShortTextType
version_notes: UKFMediumTextType
variant: UKFShortTextType
variant_notes: UKFMediumTextType
content: UKFLongTextType
content_resources: UKFJsonType
content_composers: UKFJsonType
source: UKFShortTextType
parents: UKFJsonType
creator: UKFShortTextType
owner: UKFShortTextType
workspace: UKFShortTextType
collection: UKFShortTextType
tags: UKFTagsType
synonyms: UKFSynonymsType
triggers: UKFJsonType
priority: UKFIntegerType
related: UKFRelatedType
auths: UKFAuthsType
timefluid: UKFBooleanType
timestamp: UKFTimestampType
last_verified: UKFTimestampType
expiration: UKFDurationType
inactive_mark: UKFBooleanType
metadata: UKFJsonType
profile: UKFJsonType
id_field: ClassVar[str] = 'id'
external_fields: ClassVar[Tuple[str]] = ('name', 'notes', 'short_description', 'description', 'type', 'version', 'version_notes', 'variant', 'variant_notes', 'content', 'content_resources', 'content_composers', 'source', 'parents', 'creator', 'owner', 'workspace', 'collection', 'tags', 'synonyms', 'triggers', 'priority', 'related', 'auths', 'timefluid', 'timestamp', 'last_verified', 'expiration', 'inactive_mark', 'metadata', 'profile')
internal_fields: ClassVar[Tuple[str]] = ('_id', '_content_hash', '_slots', '_type')
property_fields: ClassVar[Tuple[str]] = ('id', 'id_str', 'content_hash', 'content_hash_str', 'expiration_timestamp', 'is_inactive', 'is_active')
identity_hash_fields: ClassVar[Tuple[str]] = ('type', 'name', 'version', 'variant', 'source', 'creator', 'owner', 'workspace', 'collection', 'tags', 'timefluid')
content_hash_fields: ClassVar[Tuple[str]] = ('content', 'content_resources')
set_fields: ClassVar[Tuple[str]] = ('tags', 'synonyms', 'related', 'auths')
json_func_fields: ClassVar[Tuple[str]] = ('content_composers', 'triggers')
json_data_fields: ClassVar[Tuple[str]] = ('content_resources', 'parents', 'metadata', 'profile')
model_config = {'validate_assignment': True, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

__init__(**data)[源代码]

Initialize BaseUKF and set the _type discriminator.

classmethod model_validate(obj, *, strict=None, from_attributes=None, context=None, polymorphic=True)[源代码]

Validate a pydantic model instance with optional polymorphic deserialization.

This override of Pydantic's model_validate adds polymorphic deserialization support. When polymorphic=True (default), the method checks the 'type' field and returns the appropriate UKFT subclass from the registry.

参数:

  • obj (Any) -- The object to validate (dict, model instance, etc.)

  • strict (Optional[bool]) -- Whether to strictly validate the object

  • from_attributes (Optional[bool]) -- Whether to extract data from object attributes

  • context (Optional[Dict[str, Any]]) -- Additional context for validation

  • polymorphic (bool) -- If True (default), use registry for polymorphic deserialization. If False, use the class on which this method is called.

返回:

Validated model instance of the appropriate UKFT subclass.

返回类型:

BaseUKF

示例

>>> data = {"name": "test", "type": "knowledge", ...}
>>> # Polymorphic - returns KnowledgeUKFT
>>> obj1 = BaseUKF.model_validate(data)
>>> # Non-polymorphic - returns BaseUKF
>>> obj2 = BaseUKF.model_validate(data, polymorphic=False)
serialize_tags(tags)[源代码]
返回类型:

List[str]

参数:

tags (Set[str])

serialize_synonyms(synonyms)[源代码]
返回类型:

List[str]

参数:

synonyms (Set[str])

Notice that the related is reversed: Since Set does not support Dict (relation_resources) as items, the relation tuples should actually be deserialized (str -> json) during serialize_related (export from Python object to JSON).

返回类型:

List[Tuple[int, str, int, Optional[int], Optional[Dict[str, Any]]]]

参数:

related (Set[Tuple[int, str, int, int | None, str | None]])

serialize_auths(auths)[源代码]
返回类型:

List[Tuple[str, str]]

参数:

auths (Set[Tuple[str, str]])

serialize_triggers(triggers)[源代码]
返回类型:

Dict[str, Dict]

参数:

triggers (Dict[str, Callable])

classmethod validate_triggers(value)[源代码]

Validate and deserialize triggers.

serialize_content_composers(composers)[源代码]
返回类型:

Dict[str, Dict]

参数:

composers (Dict[str, Callable])

classmethod validate_content_composers(value)[源代码]

Validate and deserialize content composers.

serialize_timestamp(dt)[源代码]
返回类型:

str

参数:

dt (datetime)

serialize_last_verified(dt)[源代码]
返回类型:

str

参数:

dt (datetime)

serialize_expiration(value)[源代码]
返回类型:

int

参数:

value (timedelta)

serialize_expiration_timestamp(dt)[源代码]
返回类型:

str

参数:

dt (datetime)

classmethod deserialize_tags(value)[源代码]

Backward compatibility method - validation is now handled by UKFTagsType.

返回类型:

Set[str]

参数:

value (Iterable[str])

classmethod deserialize_synonyms(value)[源代码]

Backward compatibility method - validation is now handled by UKFSynonymsType.

返回类型:

Set[str]

参数:

value (Iterable[str])

Backward compatibility method - validation is now handled by UKFRelatedType.

返回类型:

Set

参数:

value (Iterable)

classmethod deserialize_auths(value)[源代码]

Backward compatibility method - validation is now handled by UKFAuthsType.

返回类型:

Set

参数:

value (Iterable)

classmethod deserialize_triggers(value)[源代码]

Backward compatibility method - deserialize function dictionaries.

返回类型:

Dict[str, Callable]

classmethod deserialize_content_composers(value)[源代码]

Backward compatibility method - deserialize function dictionaries.

返回类型:

Dict[str, Callable]

classmethod schema()[源代码]
返回类型:

Dict[str, Any]

property id: int

Compute a deterministic integer identifier for the item.

The identifier is computed by hashing a selection of identity fields (see :pyattr:`identity_hash_fields`) and cached on the instance. The cached value is cleared by _resetinternal_fields() when needed.

返回:

Integer hash value derived from identity fields.

返回类型:

int

property id_str: str

Return the id value formatted as a zero-padded string.

Uses fmt_hash to produce a fixed-width, human-safe string representation of the integer id.

返回:

Zero-padded string version of :pyattr:`id`.

返回类型:

str

property content_hash: int

Compute a hash of the content-related fields for change detection.

The hash covers :pyattr:`content` and :pyattr:`content_resources` and is cached on the instance. Cache is invalidated by _resetinternal_fields().

返回:

Integer hash representing the content state.

返回类型:

int

property content_hash_str: str

Return the content hash formatted as a zero-padded string.

返回:

String representation produced by fmt_hash for

:pyattr:`content_hash`.

返回类型:

str

property expiration_timestamp: datetime

Compute the UTC timestamp when the item expires.

If :pyattr:`expiration` is negative the function returns a distant future sentinel (year 2500) indicating no automatic expiration.

返回:

UTC datetime when the item becomes expired.

返回类型:

datetime.datetime

property slots: Dict[str, Set[str]]

Return a mapping from tag slots to their set of values.

Parses :pyattr:`tags` and groups values by slot name (the left-hand side of a [SLOT:value] tag). The result is cached in the private attribute :pyattr:`_slots` for efficiency.

返回:

Mapping of slot name to set of values.

返回类型:

Dict[str, Set[str]]

has_tag(slot, operator='ANY_OF', value=None)[源代码]

Check whether this item's tags satisfy a named condition for slot.

This is a convenience wrapper around the module-level has_tag() that uses this instance's :pyattr:`tags` set.

参数:

  • slot (str) -- Tag slot to check.

  • operator (TagOperator) -- Operator to apply (see module-level has_tag() for supported operators).

  • value (Optional[Union[Iterable, str, Any]]) -- Values to compare against for non-unary operators.

返回:

True if the condition holds for this item's tags.

返回类型:

bool

Return True if any relation on this item satisfies the provided filters.

This is a thin wrapper around the module-level has_related() that operates on :pyattr:`related`.

参数:

  • subject_id (Optional[Union[int, Iterable[int]]]) -- Filter for subject id(s).

  • relation (Optional[Union[str, Iterable[str]]]) -- Filter for relation name(s).

  • object_id (Optional[Union[int, Iterable[int]]]) -- Filter for object id(s).

  • relation_id (Optional[Union[int, Iterable[int]]]) -- Filter for relation id(s).

  • related_to_id (Optional[Union[int, Iterable[int]]]) -- Matches when either subject or object id is included in this set.

返回:

True when a matching relation exists.

返回类型:

bool

property is_inactive: bool

Return True when the item is considered inactive.

An item is inactive when either :pyattr:`inactive_mark` is True or the current UTC time is past the value returned by :pyattr:`expiration_timestamp` (when expiration is enabled).

返回:

True when the item is inactive.

返回类型:

bool

property is_active: bool

Return True when the item is active (not inactive).

返回:

Negation of :pyattr:`is_inactive`.

返回类型:

bool

set_composer(name, composer)[源代码]

Add or update a content composer in :pyattr:`content_composers`.

参数:

  • name (str) -- Name of the composer.

  • composer (Callable) -- Callable that takes a BaseUKF instance and returns a string representation.

抛出:

ValueError -- If name is empty or composer is not callable.

update_composers(composers=None)[源代码]

Update multiple content composers in :pyattr:`content_composers`.

参数:

composers (Optional[Dict[str, Callable]]) -- Mapping of composer names to callables. If None, no changes are made.

抛出:

ValueError -- If any name is empty or any composer is not callable.

text(composer='default', **kwargs)[源代码]

Return a text representation produced by a composer.

The composer parameter may be a callable or the name of a composer in :pyattr:`content_composers`. If composer is not found the raw :pyattr:`content` string is returned. Any exceptions raised by a composer are logged and the raw content is returned as a fallback.

参数:

  • composer (Optional[Union[str, Callable]]) -- Composer name or callable.

  • **kwargs -- Passed through to the composer callable.

返回:

Composed or raw textual representation.

返回类型:

str

__eq__(other)[源代码]

Equality comparison using :pyattr:`id`.

Two instances are considered equal when they are instances of the same class and their computed :pyattr:`id` values are identical.

参数:

other (Any) -- Object to compare.

返回:

True when objects represent the same knowledge id.

返回类型:

bool

__lt__(other)[源代码]

Ordering comparison used for sorting by priority.

The implementation treats higher :pyattr:`priority` values as "less" to allow reverse-priority ordering when using Python's default ascending sort.

参数:

other (BaseUKF) -- Other item to compare.

返回:

True when this item should sort before other.

返回类型:

bool

__hash__()[源代码]

Return a hash derived from :pyattr:`id`.

返回:

Integer hash value.

返回类型:

int

__str__()[源代码]

Return a compact multi-line human readable representation.

The representation shows a truncated content preview and a short form of the string id.

返回:

Readable summary of the object.

返回类型:

str

__repr__()[源代码]

Return the same value as __str__() for interactive display.

返回类型:

str

to_dict()[源代码]

Serialize the model to a plain dictionary for storage or transport.

返回:

Dictionary serialization excluding internal fields.

返回类型:

Dict[str, Any]

classmethod from_dict(data, *, polymorphic=True)[源代码]

Create a BaseUKF instance from a dictionary.

This method supports polymorphic deserialization: if the type field in the data matches a registered UKFT class in HEAVEN_UR, an instance of that specific subclass will be returned instead of a generic BaseUKF.

参数:

  • data (Dict[str, Any]) -- Dictionary produced by to_dict() or an external source.

  • polymorphic (bool) -- If True (default), use registry to return the appropriate subclass. If False, use the class on which this method is called.

返回:

Validated model instance of the appropriate UKFT subclass.

返回类型:

BaseUKF

示例

>>> data = {"name": "test", "type": "knowledge", ...}
>>> obj = BaseUKF.from_dict(data)  # Returns KnowledgeUKFT
>>> obj2 = BaseUKF.from_dict(data, polymorphic=False)  # Returns BaseUKF
classmethod from_ukf(ukf, *, polymorphic=True, override_type=False)[源代码]

Create a BaseUKF instance from another instance.

This method supports polymorphic deserialization: if the source UKF's type matches a registered UKFT class in HEAVEN_UR, an instance of that specific subclass will be returned.

参数:

  • ukf (BaseUKF) -- Instance of the same or a derived class.

  • polymorphic (bool) -- If True (default), use registry to return the appropriate subclass. If False, use the class on which this method is called, allowing intentional type conversion (e.g., upcasting PromptUKFT to ResourceUKFT). Usually, this is used when loading unknown types from storage.

  • override_type (bool) -- If True and polymorphic=False, override the 'type' field to match the target class. Useful for true downcasting where you want to change the type field. Notice that type is part of the UKF identity, so overriding it will also change the computed id, making this a different knowledge item, which could be undesired in some scenarios. Usually, this is used when defining new sub-type UKFTs. If polymorphic=True, this parameter is ignored.

返回:

Validated model instance of the appropriate UKFT subclass.

返回类型:

BaseUKF

示例

>>> prompt = PromptUKFT(name="test", ...)
>>> # Preserve original type
>>> copy1 = BaseUKF.from_ukf(prompt)
>>> # copy1.type == "prompt" and type(copy1) == PromptUKFT
>>> # Intentional downcast to parent type (keeps original type field)
>>> resource = ResourceUKFT.from_ukf(prompt, polymorphic=False)
>>> # resource.type == "prompt" and type(resource) == ResourceUKFT
>>> # True downcast with type field override
>>> resource2 = ResourceUKFT.from_ukf(prompt, polymorphic=False, override_type=True)
>>> # resource2.type == "resource" and type(resource2) == ResourceUKFT
get(key_path, default=None)[源代码]

Retrieve a nested value from the BaseUKF's content_resources using a dot-separated key path.

参数:

  • key_path (str) -- Dot-separated path to the desired value (e.g., "level1.level2.key").

  • default (Any) -- Value to return if the key path does not exist.

返回:

The value found at the specified key path, or the default if not found.

返回类型:

Any

set(key_path, value)[源代码]

Set a nested value in the BaseUKF's content_resources using a dot-separated key path.

参数:

  • key_path (str) -- Dot-separated path to set the value (e.g., "level1.level2.key").

  • value (Any) -- The value to set at the specified key path.

返回:

True if the value was set successfully, False otherwise.

返回类型:

bool

unset(key_path)[源代码]

Remove a nested value from the BaseUKF's content_resources using a dot-separated key path.

参数:

key_path (str) -- Dot-separated path to the value to remove (e.g., "level1.level2.key").

返回:

True if the value was removed successfully, False otherwise.

返回类型:

bool

setdef(key_path, value)[源代码]

Set a nested value in the BaseUKF's content_resources only if the key path does not already exist.

参数:

  • key_path (str) -- Dot-separated path to set the value (e.g., "level1.level2.key").

  • value (Any) -- The value to set at the specified key path.

返回:

True if the value was set successfully, False if the key path already exists.

返回类型:

bool

set_inactive()[源代码]

Mark the item as inactive by setting :pyattr:`inactive_mark` to True.

unset_inactive()[源代码]

Clear the manual inactive flag by setting :pyattr:`inactive_mark` to False.

set_active()[源代码]

Mark the item as active by setting :pyattr:`inactive_mark` to False.

set_trigger(name, trigger)[源代码]

Add or update a trigger callable in :pyattr:`triggers`.

参数:

  • name (str) -- Name of the trigger.

  • trigger (Callable) -- Callable that takes a BaseUKF instance and returns a boolean.

update_triggers(triggers=None)[源代码]

Merge new trigger callables into the :pyattr:`triggers` mapping.

参数:

triggers (Optional[Dict[str, Callable]]) -- Mapping of name to callable to add or update. If None, no changes are made.

eval_triggers(triggers=None, contexts=None, aggregate='ALL')[源代码]

Evaluate one or more named triggers with optional context.

参数:

  • triggers (List[str], optional) -- Names of triggers to eval_triggers. If None or empty, an empty result is returned (or True when aggregated as described below).

  • contexts (Optional[Dict], optional) -- Mapping from trigger name to a context dict passed as keyword arguments to the trigger.

  • aggregate (Literal["ANY", "ALL", False]) -- If False the function returns a dictionary of individual boolean results. If "ALL" the function returns True only if all evaluated triggers return True. If "ANY" the function returns True when any trigger returns True.

返回:

Individual results or an aggregated boolean depending on aggregate.

返回类型:

Union[Dict[str, bool], bool]

eval_filter(filter=None, **kwargs)[源代码]

Evaluate whether this BaseUKF object satisfies a KLOp expression.

This method evaluates filter conditions in-memory by checking the object's field values against the filter criteria. It supports all KLOp operators and handles both parsed JSON IR expressions and KLOp objects.

参数:

  • filter (Union[Dict[str, Any], KLOp, None]) -- Either a parsed KLOp.expr() dict or None. If None, only kwargs are used.

  • **kwargs -- Additional field constraints that are ANDed with the filter.

返回:

True if the object satisfies all filter conditions, False otherwise.

返回类型:

bool

示例

>>> ukf = BaseUKF(name="test", priority=50, status="active")
>>> ukf.eval_filter(priority=50)  # Simple equality
True
>>> ukf.eval_filter(priority=KLOp.GT(40))  # Comparison operator
True
>>> ukf.eval_filter(priority=KLOp.BETWEEN(0, 100), status="active")  # Combined
True
>>> from ahvn.utils.klop import KLOp
>>> expr = KLOp.expr(priority=KLOp.GT(40))
>>> ukf.eval_filter(expr)  # Using parsed expression
True
clone(**updates)[源代码]

Return a deep-copied model updated with updates.

The returned instance has internal caches reset so identity and content hashes are recomputed on demand.

参数:

**updates -- Field values to override in the cloned instance. For dict/list/set fields, use upd_<field> to update items instead of overwriting. Typical fields that support this are :pyattr:`tags`, :pyattr:`synonyms`, :pyattr:`related`, :pyattr:`auths`, :pyattr:`triggers`, :pyattr:`content_composers`, :pyattr: content_resources, :pyattr:`metadata`, etc.

返回:

New instance with the requested updates applied.

返回类型:

BaseUKF

derive(**updates)[源代码]

Create a derived item with incremented version and provenance.

The derived copy has its :pyattr:`version` bumped using next_ver(), :pyattr:`source` set to "derived" and the original id recorded in :pyattr:`parents`.

参数:

**updates -- Additional field updates to apply to the derived item. For dict/list/set fields, use upd_<field> to update items instead of overwriting. Typical fields that support this are :pyattr:`tags`, :pyattr:`synonyms`, :pyattr:`related`, :pyattr:`auths`, :pyattr:`triggers`, :pyattr:`content_composers`, :pyattr: content_resources, :pyattr:`metadata`, etc.

返回:

Derived instance.

返回类型:

BaseUKF

Add a lightweight relation between this item and kl.

Relation tuples are inserted into :pyattr:`related` using the serialized form for relation_resources. The dir parameter controls whether this item is the subject, object or both.

参数:

  • kl (BaseUKF) -- Target knowledge item.

  • dir (Literal["subject","object","both"]) -- Direction of the relation where "object" means self -> kl.

  • rel (str) -- Relation name.

  • rel_kid (Optional[int]) -- Optional relation id.

  • relation_resources (Optional[Any]) -- Additional relation metadata.

  • inv_link (bool) -- When True also add reciprocal entries on kl.

obj_ids(rel=None)[源代码]

Return object ids for relations where this item is the subject.

参数:

rel (Optional[str]) -- If provided, only relations matching this relation name are included.

返回:

Sequence of object ids.

返回类型:

List[str]

sub_ids(rel=None)[源代码]

Return subject ids for relations where this item is the object.

参数:

rel (Optional[str]) -- Optional relation name filter.

返回:

Sequence of subject ids.

返回类型:

List[str]

grant(user_id, authority)[源代码]

Grant an authority by adding [user:authority] tag.

参数:

  • user_id (int) -- User identifier.

  • authority (str) -- Permission or role string.

revoke(user_id, authority)[源代码]

Remove an authority by removing [user:authority] tag.

参数:

  • user_id (int) -- User identifier.

  • authority (str) -- Permission or role string.

has_authority(user_id, authority)[源代码]

Return True when [user:authority] tag is present.

参数:

  • user_id (int) -- User identifier.

  • authority (str) -- Authority string.

返回:

True if the authority is granted to the user.

返回类型:

bool

add_synonyms(synonyms)[源代码]

Append new synonyms into the :pyattr:`synonyms` set.

参数:

synonyms (Iterable[str]) -- Iterable of synonym strings to add.

返回类型:

None

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

参数:

  • self (BaseModel) -- The BaseModel instance.

  • context (Any) -- The context.

返回类型:

None

remove_synonyms(synonyms)[源代码]

Remove synonyms from the :pyattr:`synonyms` set.

参数:

synonyms (Iterable[str]) -- Iterable of synonym strings to remove.

返回类型:

None

signed(system=False, verified=True, **kwargs)[源代码]

Sign the knowledge item with default provenance information.

参数:

  • system (bool) -- Whether the knowledge is created by the system. Defaults to False.

  • verified (bool) -- Whether the knowledge is verified as accurate. Defaults to True.

  • **kwargs -- Additional provenance fields to override. Specifically the following kwargs are set: - source: set to the provided source, or "system" if system is True, or existing source, or "user" if existing source is "unknown" - creator: set to the provided creator, or "system" if system is True, or existing creator, or current user id from HEAVEN_CM if existing creator is "unknown" - owner: set to the provided owner, or "system" if system is True, or existing owner, or current user id from HEAVEN_CM if existing owner is "unknown" - workspace: set to the provided workspace, or existing workspace - inactive_mark: set to the provided inactive_mark, or not verified - last_verified: set to the provided last_verified, or current UTC time without microseconds if verified is True, or existing last_verified

返回:

A cloned knowledge item with updated provenance fields.

返回类型:

BaseUKF

class ahvn.ukf.UKFTypeRegistry[源代码]

基类：object

Central registry for UKF Template types.

This registry maps UKF type strings (from the type field) to their corresponding UKFT class implementations. It enables polymorphic deserialization where from_dict and from_ukf can return the appropriate subclass based on the type field.

示例

>>> @register_ukft
... class MyUKFT(BaseUKF):
...     type: str = Field(default="my_type", frozen=True)
>>>
>>> data = {"type": "my_type", "name": "test", ...}
>>> ukf = BaseUKF.from_dict(data)  # Returns MyUKFT instance
>>> isinstance(ukf, MyUKFT)
True
__init__()[源代码]
register(ukft_class)[源代码]

Register a UKFT class in the registry.

参数:

ukft_class (Type[BaseUKF]) -- A BaseUKF subclass to register.

返回类型:

Type[BaseUKF]

返回:

The same class (for use as a decorator).

抛出:

ValueError -- If the class doesn't have a type field or if the type is already registered.

get(type_name, default=None)[源代码]

Get a UKFT class by its type name.

参数:

  • type_name (str) -- The type string to look up.

  • default (Type[BaseUKF]) -- Default value to return if type not found.

返回类型:

Type[BaseUKF]

返回:

The registered UKFT class for this type, or the default value if not found.

list_types()[源代码]

List all registered type names.

返回类型:

list[str]

返回:

List of registered type strings.

is_registered(type_name)[源代码]

Check if a type is registered.

参数:

type_name (str) -- The type string to check.

返回类型:

bool

返回:

True if the type is registered, False otherwise.

ahvn.ukf.register_ukft(ukft_class)[源代码]

Decorator to register a UKFT class in the global registry.

This decorator should be applied to all BaseUKF subclasses that represent concrete UKF types. It automatically registers the class in HEAVEN_UR for polymorphic deserialization.

参数:

ukft_class (Type[BaseUKF]) -- A BaseUKF subclass to register.

返回类型:

Type[BaseUKF]

返回:

The same class (for use as a decorator).

示例

>>> @register_ukft
... class KnowledgeUKFT(BaseUKF):
...     type: str = Field(default="knowledge", frozen=True)
class ahvn.ukf.UKFIdType[源代码]

基类：int

Custom type for UKF ID fields with validation and formatting.

Validates and converts values to integer IDs, supporting both raw integers and formatted hash strings with underscores or dashes.

It is recommended to use integer IDs generated from md5hash in hash_utils.

参数:

value -- Integer or string representation of an ID.

返回:

Validated integer ID.

返回类型:

int

抛出:

PydanticCustomError -- If value cannot be converted to a valid integer ID.

示例

>>> UKFIdType._validate(123)
123
>>> UKFIdType._validate("000123")
123
name = 'id'
class ahvn.ukf.UKFIntegerType[源代码]

基类：int

Custom type for integer fields with validation.

Validates and converts values to integers, providing consistent handling of numeric inputs including integers, floats and string representations.

参数:

value -- Integer, float, or string representation of an integer.

返回:

Validated integer value.

返回类型:

int

抛出:

PydanticCustomError -- If value cannot be converted to a valid integer.

示例

>>> UKFIntegerType._validate(123)
123
>>> UKFIntegerType._validate(123.0)
123
>>> UKFIntegerType._validate("123")
123
name = 'int'
class ahvn.ukf.UKFBooleanType[源代码]

基类：object

Custom type for boolean fields with validation.

Validates and converts various representations to boolean values, supporting common string representations and numeric values.

参数:

value -- Boolean, integer, string, or other value to convert.

返回:

Validated boolean value.

返回类型:

bool

抛出:

PydanticCustomError -- If value cannot be converted to a valid boolean.

示例

>>> UKFBooleanType._validate(True)
True
>>> UKFBooleanType._validate("true")
True
>>> UKFBooleanType._validate(1)
True
>>> UKFBooleanType._validate("false")
False
>>> UKFBooleanType._validate(0)
False
name = 'bool'
class ahvn.ukf.UKFShortTextType[源代码]

基类：str

Custom type for short text fields with length validation.

Validates string length against configurable short text limit from config.yaml. Default limit is 255 characters (equivalent to SQL VARCHAR(255)).

参数:

value -- String value to validate.

返回:

Validated string value.

返回类型:

str

抛出:

PydanticCustomError -- If string exceeds maximum length limit.

示例

>>> UKFShortTextType._validate("Hello world")
'Hello world'
>>> UKFShortTextType._validate("a" * 256)  # Assuming 255 char limit
PydanticCustomError: short_text_too_long
name = 'short_text'
classmethod max_length()[源代码]
返回类型:

int

class ahvn.ukf.UKFMediumTextType[源代码]

基类：str

Custom type for medium text fields with length validation.

Validates string length against configurable medium text limit from config.yaml. Default limit is 2047 characters (equivalent to SQL VARCHAR(2047)).

参数:

value -- String value to validate.

返回:

Validated string value.

返回类型:

str

抛出:

PydanticCustomError -- If string exceeds maximum length limit.

示例

>>> UKFMediumTextType._validate("Medium length text")
'Medium length text'
>>> UKFMediumTextType._validate("a" * 2048)  # Assuming 2047 char limit
PydanticCustomError: medium_text_too_long
name = 'medium_text'
classmethod max_length()[源代码]
返回类型:

int

class ahvn.ukf.UKFLongTextType[源代码]

基类：str

Custom type for long text fields with length validation.

Validates string length against configurable long text limit from config.yaml. Default limit is 65535 characters (equivalent to SQL VARCHAR(65535)).

参数:

value -- String value to validate.

返回:

Validated string value.

返回类型:

str

抛出:

PydanticCustomError -- If string exceeds maximum length limit.

示例

>>> UKFLongTextType._validate("Very long text content")
'Very long text content'
>>> UKFLongTextType._validate("a" * 65536)  # Assuming 65535 char limit
PydanticCustomError: long_text_too_long
name = 'long_text'
classmethod max_length()[源代码]
返回类型:

int

class ahvn.ukf.UKFTimestampType[源代码]

基类：datetime

Custom type for datetime fields with UTC conversion and validation.

Validates and normalizes datetime values to UTC timezone with microseconds stripped for consistency. Supports various input formats including ISO strings, timestamps, and datetime objects.

参数:

value -- Datetime, ISO string, timestamp (int/float), or datetime object.

返回:

UTC datetime with microseconds stripped.

返回类型:

datetime.datetime

抛出:

PydanticCustomError -- If value cannot be converted to a valid datetime.

示例

>>> UKFTimestampType._validate("2023-01-01T12:00:00Z")
datetime.datetime(2023, 1, 1, 12, 0, tzinfo=datetime.timezone.utc)
>>> UKFTimestampType._validate(1672574400)  # Unix timestamp
datetime.datetime(2023, 1, 1, 12, 0, tzinfo=datetime.timezone.utc)
name = 'timestamp'
class ahvn.ukf.UKFDurationType[源代码]

基类：timedelta

Custom type for duration fields with validation.

Validates and converts various representations to timedelta objects, supporting ISO 8601 duration strings and numeric seconds.

参数:

value -- Timedelta, ISO 8601 duration string, or numeric seconds.

返回:

Validated timedelta object.

返回类型:

datetime.timedelta

抛出:

PydanticCustomError -- If value cannot be converted to a valid timedelta.

示例

>>> UKFDurationType._validate("P1DT2H")  # 1 day, 2 hours
datetime.timedelta(days=1, hours=2)
>>> UKFDurationType._validate(3600)  # 1 hour in seconds
datetime.timedelta(seconds=3600)
name = 'duration'
class ahvn.ukf.UKFJsonType[源代码]

基类：dict

Custom type for JSON fields with validation and parsing.

Validates and converts JSON data, supporting both dictionary objects and JSON string representations. Uses custom JSON parser for consistency.

参数:

value -- Dictionary object or JSON string to validate.

返回:

Validated dictionary object.

返回类型:

dict

抛出:

PydanticCustomError -- If JSON string cannot be parsed or value is invalid.

示例

>>> UKFJsonType._validate({"key": "value"})
{'key': 'value'}
>>> UKFJsonType._validate('{"key": "value"}')
{'key': 'value'}
>>> UKFJsonType._validate(None)
{}
name = 'json'
class ahvn.ukf.UKFTagsType[源代码]

基类：set

Custom type for tags set with validation and serialization.

Validates and converts various iterable types to a set of string tags. Handles None values gracefully by returning empty set. Also supports auth tuples which are converted to "[user:authority]" tag format.

参数:

value -- Set, list, tuple, or other iterable of tag values. Can also include auth tuples (user, authority) which are converted to tags.

返回:

Set of string tags.

返回类型:

set

抛出:

TypeError -- If value cannot be iterated over.

示例

>>> UKFTagsType._validate(["tag1", "tag2"])
{'tag1', 'tag2'}
>>> UKFTagsType._validate({"tag1", "tag2"})
{'tag1', 'tag2'}
>>> UKFTagsType._validate([("user1", "read")])  # Auth tuple as tag
{'[user1:read]'}
>>> UKFTagsType._validate(None)
set()
name = 'tags'
ahvn.ukf.tag_s(tag)[源代码]

Return the slot (key) from a tag string.

A tag is expected to be in the format "[slot:value]". This helper extracts and returns the slot portion.

参数:

tag (str) -- Tag in the format "[slot:value]".

返回:

The slot part (text before the first ":").

返回类型:

str

ahvn.ukf.tag_v(tag)[源代码]

Return the value part from a tag string.

参数:

tag (str) -- Tag in the format "[slot:value]".

返回:

The value part (text after the first ":").

返回类型:

str

ahvn.ukf.tag_t(tag)[源代码]

Split a tag string into (slot, value).

参数:

tag (str) -- Tag in the format "[slot:value]".

返回:

A sequence of two strings: [slot, value].

返回类型:

List[str]

ahvn.ukf.ptags(**kwargs)[源代码]

Create formatted tag strings from keyword arguments.

Values may be scalars or iterables; iterables are expanded into multiple tags; None values are skipped.

参数:

**kwargs -- Keys will be used as tag slots, values provide the tag values. Example: TYPE=['doc','text'], LANG='en'.

返回:

A set of tag strings in the format "[SLOT:value]".

返回类型:

Set[str]

ahvn.ukf.gtags(tags, **kwargs)[源代码]

Group tags by slot name and collect values.

This converts a flat list/set of tag strings into a mapping from slot to set of values. Additional tags can be provided via the same keyword interface accepted by ptags().

参数:

  • tags (Iterable[str]) -- Iterable of tag strings like "[slot:value]".

  • **kwargs -- Extra tags passed to ptags().

返回:

Mapping from lowercase slot name to a set of values.

返回类型:

Dict[str, Set[str]]

ahvn.ukf.has_tag(tags, slot, operator='ANY_OF', value=None)[源代码]

Check whether a collection of tags satisfies a condition for a slot.

The function supports a variety of operators to express membership and set-based conditions. See the implementation for the full list of supported operators; common ones include ANY_OF, ALL_OF, and unary tests like HAS_NONE.

参数:

  • tags (Iterable[str]) -- Iterable of tag strings, e.g. "[type:doc]".

  • slot (str) -- Slot name to check (the left-hand part of a tag).

  • operator (TagOperator) -- Operator describing the condition. Can be a textual operator (see module-level TagOperator) or a numeric requirement (int or float are treated specially).

  • value (Optional[Union[Iterable, str, Any]]) -- Value(s) to compare against. Required for non-unary operators.

Operators:

"EXACT" or "==": Slot values exactly match the provided values (sets equal). "NONE_OF": No slot values match the provided values. "ANY_OF": At least one slot value matches the provided values. "ANY_IF_EXISTS": If the slot exists, at least one provided value matches; if slot missing, returns True. "ONE_OF": Exactly one slot value matches the provided values. "MANY_OF": At least two slot values match the provided values. "ALL_OF": All provided values are present in the slot values. "ALL_IN" or "IN": All slot values are included in the provided values. "HAS_NONE": Unary — slot has no values. "HAS_ANY": Unary — slot has at least one value. "HAS_ONE": Unary — slot has exactly one value. "HAS_MANY": Unary — slot has at least two values. int: Numeric operator meaning "at least N matching values". float: Jaccard similarity threshold (intersection/union >= threshold).

返回:

True when the condition holds for the provided tags, otherwise False.

返回类型:

bool

抛出:

ValueError -- If an unary operator is given a non-None value or when an unsupported operator is provided.

参数:

  • tags (Iterable[str])

  • slot (str)

  • operator (Literal['EXACT', 'NONE_OF', 'ANY_OF', 'ANY_IF_EXISTS', 'ONE_OF', 'MANY_OF', 'ALL_OF', 'ALL_IN', 'HAS_NONE', 'HAS_ANY', 'HAS_ONE', 'HAS_MANY'] | int | float)

  • value (Iterable | str | Any | None)

Return True when the given related tuples contain a matching relation.

Each relation tuple is expected to be in the form (subject_id, relation, object_id, relation_id?, relation_resources?).

参数:

  • related (Iterable[Tuple[int, str, int, Optional[int], Optional[Dict]]]) -- Iterable of relation tuples to search.

  • subject_id (Optional[Union[int, Iterable[int]]]) -- Filter for subject id(s).

  • relation (Optional[Union[str, Iterable[str]]]) -- Filter for relation name(s).

  • object_id (Optional[Union[int, Iterable[int]]]) -- Filter for object id(s).

  • relation_id (Optional[Union[int, Iterable[int]]]) -- Filter for relation id(s).

  • related_to_id (Optional[Union[int, Iterable[int]]]) -- If provided, the function matches relations where either the subject or object id is in this set.

返回:

True if at least one relation satisfies all provided filters.

返回类型:

bool

ahvn.ukf.next_ver(version)[源代码]

Return the next version string by incrementing the last numeric part.

If the last component of version is numeric it will be incremented by one. Otherwise a new numeric component "1" will be appended.

参数:

version (str) -- Current version string (for example "v1.2.3" or "v1.2.beta").

返回:

New version string with the final numeric component incremented or

appended.

返回类型:

str

示例

>>> next_ver('v1.2.3')
'v1.2.4'
>>> next_ver('v1.2.beta')
'v1.2.beta.1'

Subpackages

Submodules