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.

Parameters:

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

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

Returns:

Always True.

Return type:

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.

Parameters:

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

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

Returns:

The content attribute of kl.

Return type:

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>)

Bases: 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.

Parameters:

• 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)

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.

Parameters:

• 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.

Returns:

Validated model instance of the appropriate UKFT subclass.

Return type:

BaseUKF

Examples

>>> 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)

Return type:

List[str]

Parameters:

tags (Set[str])

serialize_synonyms(synonyms)

Return type:

List[str]

Parameters:

synonyms (Set[str])

serialize_related(related)

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).

Return type:

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

Parameters:

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

serialize_auths(auths)

Return type:

List[Tuple[str, str]]

Parameters:

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

serialize_triggers(triggers)

Return type:

Dict[str, Dict]

Parameters:

triggers (Dict[str, Callable])

classmethod validate_triggers(value)

Validate and deserialize triggers.

serialize_content_composers(composers)

Return type:

Dict[str, Dict]

Parameters:

composers (Dict[str, Callable])

classmethod validate_content_composers(value)

Validate and deserialize content composers.

serialize_timestamp(dt)

Return type:

str

Parameters:

dt (datetime)

serialize_last_verified(dt)

Return type:

str

Parameters:

dt (datetime)

serialize_expiration(value)

Return type:

int

Parameters:

value (timedelta)

serialize_expiration_timestamp(dt)

Return type:

str

Parameters:

dt (datetime)

classmethod deserialize_tags(value)

Backward compatibility method - validation is now handled by UKFTagsType.

Return type:

Set[str]

Parameters:

value (Iterable[str])

classmethod deserialize_synonyms(value)

Backward compatibility method - validation is now handled by UKFSynonymsType.

Return type:

Set[str]

Parameters:

value (Iterable[str])

classmethod deserialize_related(value)

Backward compatibility method - validation is now handled by UKFRelatedType.

Return type:

Set

Parameters:

value (Iterable)

classmethod deserialize_auths(value)

Backward compatibility method - validation is now handled by UKFAuthsType.

Return type:

Set

Parameters:

value (Iterable)

classmethod deserialize_triggers(value)

Backward compatibility method - deserialize function dictionaries.

Return type:

Dict[str, Callable]

classmethod deserialize_content_composers(value)

Backward compatibility method - deserialize function dictionaries.

Return type:

Dict[str, Callable]

classmethod schema()

Return type:

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.

Returns:

Integer hash value derived from identity fields.

Return type:

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.

Returns:

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

Return type:

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().

Returns:

Integer hash representing the content state.

Return type:

int

property content_hash_str: str

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

Returns:

String representation produced by fmt_hash for

:pyattr:`content_hash`.

Return type:

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.

Returns:

UTC datetime when the item becomes expired.

Return type:

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.

Returns:

Mapping of slot name to set of values.

Return type:

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.

Parameters:

• 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.

Returns:

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

Return type:

bool

has_related(subject_id=None, relation=None, object_id=None, relation_id=None, related_to_id=None)

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`.

Parameters:

• 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.

Returns:

True when a matching relation exists.

Return type:

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).

Returns:

True when the item is inactive.

Return type:

bool

property is_active: bool

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

Returns:

Negation of :pyattr:`is_inactive`.

Return type:

bool

set_composer(name, composer)

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

Parameters:

• name (str) – Name of the composer.

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

Raises:

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

update_composers(composers=None)

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

Parameters:

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

Raises:

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.

Parameters:

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

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

Returns:

Composed or raw textual representation.

Return type:

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.

Parameters:

other (Any) – Object to compare.

Returns:

True when objects represent the same knowledge id.

Return type:

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.

Parameters:

other (BaseUKF) – Other item to compare.

Returns:

True when this item should sort before other.

Return type:

bool

__hash__()

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

Returns:

Integer hash value.

Return type:

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.

Returns:

Readable summary of the object.

Return type:

str

__repr__()

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

Return type:

str

to_dict()

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

Returns:

Dictionary serialization excluding internal fields.

Return type:

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.

Parameters:

• 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.

Returns:

Validated model instance of the appropriate UKFT subclass.

Return type:

BaseUKF

Examples

>>> 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.

Parameters:

• 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.

Returns:

Validated model instance of the appropriate UKFT subclass.

Return type:

BaseUKF

Examples

>>> 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.

Parameters:

• 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.

Returns:

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

Return type:

Any

set(key_path, value)

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

Parameters:

• 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.

Returns:

True if the value was set successfully, False otherwise.

Return type:

bool

unset(key_path)

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

Parameters:

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

Returns:

True if the value was removed successfully, False otherwise.

Return type:

bool

setdef(key_path, value)

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

Parameters:

• 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.

Returns:

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

Return type:

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`.

Parameters:

• 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.

Parameters:

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.

Parameters:

• 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.

Returns:

Individual results or an aggregated boolean depending on aggregate.

Return type:

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.

Parameters:

• 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.

Returns:

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

Return type:

bool

Example

>>> 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.

Parameters:

**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.

Returns:

New instance with the requested updates applied.

Return type:

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`.

Parameters:

**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.

Returns:

Derived instance.

Return type:

BaseUKF

link(kl, dir='object', rel='related', rel_kid=None, relation_resources=None, inv_link=False)

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.

Parameters:

• 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.

Parameters:

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

Returns:

Sequence of object ids.

Return type:

List[str]

sub_ids(rel=None)

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

Parameters:

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

Returns:

Sequence of subject ids.

Return type:

List[str]

grant(user_id, authority)

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

Parameters:

• user_id (int) – User identifier.

• authority (str) – Permission or role string.

revoke(user_id, authority)

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

Parameters:

• user_id (int) – User identifier.

• authority (str) – Permission or role string.

has_authority(user_id, authority)

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

Parameters:

• user_id (int) – User identifier.

• authority (str) – Authority string.

Returns:

True if the authority is granted to the user.

Return type:

bool

add_synonyms(synonyms)

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

Parameters:

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

Return type:

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.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

Return type:

None

remove_synonyms(synonyms)

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

Parameters:

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

Return type:

None

signed(system=False, verified=True, **kwargs)

Sign the knowledge item with default provenance information.

Parameters:

• 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

Returns:

A cloned knowledge item with updated provenance fields.

Return type:

BaseUKF

class ahvn.ukf.UKFTypeRegistry

Bases: 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.

Example

>>> @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.

Parameters:

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

Return type:

Type[BaseUKF]

Returns:

The same class (for use as a decorator).

Raises:

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.

Parameters:

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

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

Return type:

Type[BaseUKF]

Returns:

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

list_types()

List all registered type names.

Return type:

list[str]

Returns:

List of registered type strings.

is_registered(type_name)

Check if a type is registered.

Parameters:

type_name (str) – The type string to check.

Return type:

bool

Returns:

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.

Parameters:

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

Return type:

Type[BaseUKF]

Returns:

The same class (for use as a decorator).

Example

>>> @register_ukft
... class KnowledgeUKFT(BaseUKF):
...     type: str = Field(default="knowledge", frozen=True)

class ahvn.ukf.UKFIdType

Bases: 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.

Parameters:

value – Integer or string representation of an ID.

Returns:

Validated integer ID.

Return type:

int

Raises:

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

Examples

>>> UKFIdType._validate(123)
123
>>> UKFIdType._validate("000123")
123

name = 'id'

class ahvn.ukf.UKFIntegerType

Bases: 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.

Parameters:

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

Returns:

Validated integer value.

Return type:

int

Raises:

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

Examples

>>> UKFIntegerType._validate(123)
123
>>> UKFIntegerType._validate(123.0)
123
>>> UKFIntegerType._validate("123")
123

name = 'int'

class ahvn.ukf.UKFBooleanType

Bases: object

Custom type for boolean fields with validation.

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

Parameters:

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

Returns:

Validated boolean value.

Return type:

bool

Raises:

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

Examples

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

Bases: 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)).

Parameters:

value – String value to validate.

Returns:

Validated string value.

Return type:

str

Raises:

PydanticCustomError – If string exceeds maximum length limit.

Examples

>>> 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()

Return type:

int

class ahvn.ukf.UKFMediumTextType

Bases: 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)).

Parameters:

value – String value to validate.

Returns:

Validated string value.

Return type:

str

Raises:

PydanticCustomError – If string exceeds maximum length limit.

Examples

>>> 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()

Return type:

int

class ahvn.ukf.UKFLongTextType

Bases: 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)).

Parameters:

value – String value to validate.

Returns:

Validated string value.

Return type:

str

Raises:

PydanticCustomError – If string exceeds maximum length limit.

Examples

>>> 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()

Return type:

int

class ahvn.ukf.UKFTimestampType

Bases: 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.

Parameters:

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

Returns:

UTC datetime with microseconds stripped.

Return type:

datetime.datetime

Raises:

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

Examples

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

Bases: timedelta

Custom type for duration fields with validation.

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

Parameters:

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

Returns:

Validated timedelta object.

Return type:

datetime.timedelta

Raises:

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

Examples

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

Bases: 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.

Parameters:

value – Dictionary object or JSON string to validate.

Returns:

Validated dictionary object.

Return type:

dict

Raises:

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

Examples

>>> UKFJsonType._validate({"key": "value"})
{'key': 'value'}
>>> UKFJsonType._validate('{"key": "value"}')
{'key': 'value'}
>>> UKFJsonType._validate(None)
{}

name = 'json'

class ahvn.ukf.UKFTagsType

Bases: 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.

Parameters:

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

Returns:

Set of string tags.

Return type:

set

Raises:

TypeError – If value cannot be iterated over.

Examples

>>> 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.

Parameters:

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

Returns:

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

Return type:

str

ahvn.ukf.tag_v(tag)

Return the value part from a tag string.

Parameters:

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

Returns:

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

Return type:

str

ahvn.ukf.tag_t(tag)

Split a tag string into (slot, value).

Parameters:

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

Returns:

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

Return type:

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.

Parameters:

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

Returns:

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

Return type:

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().

Parameters:

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

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

Returns:

Mapping from lowercase slot name to a set of values.

Return type:

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.

Parameters:

• 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).

Returns:

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

Return type:

bool

Raises:

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

Parameters:

• 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)

ahvn.ukf.has_related(related, subject_id=None, relation=None, object_id=None, relation_id=None, related_to_id=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?).

Parameters:

• 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.

Returns:

True if at least one relation satisfies all provided filters.

Return type:

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.

Parameters:

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

Returns:

New version string with the final numeric component incremented or

appended.

Return type:

str

Examples

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

Subpackages

• ahvn.ukf.templates package
  • Subpackages
    • ahvn.ukf.templates.basic package
      • Submodules

Submodules

• ahvn.ukf.base module
• ahvn.ukf.registry module
  • UKFTypeRegistry
    • UKFTypeRegistry.__init__()
    • UKFTypeRegistry.register()
    • UKFTypeRegistry.get()
    • UKFTypeRegistry.list_types()
    • UKFTypeRegistry.is_registered()
  • register_ukft()
• ahvn.ukf.types module
  • UKFIdType
    • UKFIdType.name
  • UKFIntegerType
    • UKFIntegerType.name
  • UKFBooleanType
    • UKFBooleanType.name
  • UKFShortTextType
    • UKFShortTextType.name
    • UKFShortTextType.max_length()
  • UKFMediumTextType
    • UKFMediumTextType.name
    • UKFMediumTextType.max_length()
  • UKFLongTextType
    • UKFLongTextType.name
    • UKFLongTextType.max_length()
  • UKFTimestampType
    • UKFTimestampType.name
  • UKFDurationType
    • UKFDurationType.name
  • UKFJsonType
    • UKFJsonType.name
  • UKFTagsType
    • UKFTagsType.name
  • UKFAuthsType
    • UKFAuthsType.name
  • UKFSynonymsType
    • UKFSynonymsType.name
  • UKFRelatedType
    • UKFRelatedType.name
  • UKFVectorType
    • UKFVectorType.name
• ahvn.ukf.ukf_utils module
  • valid_tag()
  • tag_s()
  • tag_v()
  • tag_t()
  • ptags()
  • gtags()
  • has_tag()
  • has_related()
  • next_ver()