ahvn.ukf.base module¶
Universal Knowledge Framework (UKF) base module.
This module provides the core BaseUKF model and a small set of
utility functions used throughout the UKF implementation:
Tag helpers: parsing, grouping and membership checks (
tag_s,tag_v,tag_t,ptags,gtags,has_tag)Relation helpers: check relation entries (
has_related)Versioning helper:
next_verDefault no-op trigger/composer helpers used as safe defaults
The data model is implemented with Pydantic and is intended to represent a piece of knowledge with rich metadata, provenance, tagging and simple relationship modelling. Most public functions and methods in this module use Google-style docstrings describing arguments, return values and raised exceptions.
- ahvn.ukf.base.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.
- ahvn.ukf.base.default_composer(kl, **kwargs)[源代码]
Default content composer that returns the raw
contentstring.This function is intended as a safe default for the
content_composersmapping.
- class ahvn.ukf.base.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>)[源代码]
基类:
BaseModelBase 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.
- 参数:
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)
- 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
- 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')
- 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')
- 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 objectfrom_attributes (
Optional[bool]) -- Whether to extract data from object attributescontext (
Optional[Dict[str,Any]]) -- Additional context for validationpolymorphic (
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.
- 返回类型:
示例
>>> 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_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).
- classmethod validate_triggers(value)[源代码]
Validate and deserialize triggers.
- serialize_content_composers(composers)[源代码]
- classmethod validate_content_composers(value)[源代码]
Validate and deserialize content composers.
- classmethod deserialize_tags(value)[源代码]
Backward compatibility method - validation is now handled by UKFTagsType.
- classmethod deserialize_synonyms(value)[源代码]
Backward compatibility method - validation is now handled by UKFSynonymsType.
- classmethod deserialize_related(value)[源代码]
Backward compatibility method - validation is now handled by UKFRelatedType.
- classmethod deserialize_auths(value)[源代码]
Backward compatibility method - validation is now handled by UKFAuthsType.
- classmethod deserialize_triggers(value)[源代码]
Backward compatibility method - deserialize function dictionaries.
- classmethod deserialize_content_composers(value)[源代码]
Backward compatibility method - deserialize function dictionaries.
- 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.
- 返回类型:
- property id_str: str
Return the
idvalue 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`.
- 返回类型:
- 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.
- 返回类型:
- property content_hash_str: str
Return the content hash formatted as a zero-padded string.
- 返回:
- String representation produced by fmt_hash for
- 返回类型:
- 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.
- 返回类型:
- 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.
- 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.
- 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`.- 参数:
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.
- 返回类型:
- 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.
- 返回类型:
- property is_active: bool
Return True when the item is active (not inactive).
- 返回:
Negation of :pyattr:`is_inactive`.
- 返回类型:
- set_composer(name, composer)[源代码]
Add or update a content composer in :pyattr:`content_composers`.
- 参数:
- 抛出:
ValueError -- If
nameis empty orcomposeris 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
composerparameter may be a callable or the name of a composer in :pyattr:`content_composers`. Ifcomposeris 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.
- __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.
- __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.
- __hash__()[源代码]
Return a hash derived from :pyattr:`id`.
- 返回:
Integer hash value.
- 返回类型:
- __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.
- 返回类型:
- to_dict()[源代码]
Serialize the model to a plain dictionary for storage or transport.
- classmethod from_dict(data, *, polymorphic=True)[源代码]
Create a
BaseUKFinstance 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.
- 参数:
- 返回:
Validated model instance of the appropriate UKFT subclass.
- 返回类型:
示例
>>> 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
BaseUKFinstance 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.
- 返回类型:
示例
>>> 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.
- set(key_path, value)[源代码]
Set a nested value in the BaseUKF's content_resources using a dot-separated key path.
- unset(key_path)[源代码]
Remove a nested value from the BaseUKF's content_resources using a dot-separated key path.
- setdef(key_path, value)[源代码]
Set a nested value in the BaseUKF's content_resources only if the key path does not already exist.
- 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`.
- 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
Truewhen 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
Falsethe function returns a dictionary of individual boolean results. If"ALL"the function returnsTrueonly if all evaluated triggers returnTrue. If"ANY"the function returnsTruewhen any trigger returnsTrue.
- 返回:
Individual results or an aggregated boolean depending on
aggregate.- 返回类型:
- 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.
- 参数:
- 返回:
True if the object satisfies all filter conditions, False otherwise.
- 返回类型:
示例
>>> 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.
- 返回类型:
- 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.
- 返回类型:
- 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. Thedirparameter 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"meansself -> 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.
- sub_ids(rel=None)[源代码]
Return subject ids for relations where this item is the object.
- grant(user_id, authority)[源代码]
Grant an authority by adding [user:authority] tag.
- revoke(user_id, authority)[源代码]
Remove an authority by removing [user:authority] tag.
- has_authority(user_id, authority)[源代码]
Return True when [user:authority] tag is present.
- add_synonyms(synonyms)[源代码]
Append new synonyms into the :pyattr:`synonyms` set.
- 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.
- remove_synonyms(synonyms)[源代码]
Remove synonyms from the :pyattr:`synonyms` set.
- 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.
- 返回类型: