47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
"""Structured conversational path helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
from .models import MemoryEvent, MemoryPath, MemoryPathState
|
|
from .utils import stable_digest, utc_now_iso
|
|
|
|
|
|
def create_path(path_id: str, *, event_ids: tuple[str, ...] = (), metadata: dict | None = None) -> MemoryPath:
|
|
return MemoryPath(path_id=path_id, event_ids=event_ids, metadata=dict(metadata or {}))
|
|
|
|
|
|
def branch_path(parent: MemoryPath, path_id: str, *, event_ids: tuple[str, ...] = ()) -> MemoryPath:
|
|
return MemoryPath(
|
|
path_id=path_id,
|
|
parent_path_id=parent.path_id,
|
|
event_ids=event_ids,
|
|
metadata={"branched_from": parent.path_id},
|
|
)
|
|
|
|
|
|
def merge_path(path: MemoryPath, target_path_id: str) -> MemoryPath:
|
|
return replace(path, state=MemoryPathState.MERGED, merged_into=target_path_id, updated_at=utc_now_iso())
|
|
|
|
|
|
def abandon_path(path: MemoryPath, reason: str) -> MemoryPath:
|
|
return replace(path, state=MemoryPathState.ABANDONED, abandoned_reason=reason, updated_at=utc_now_iso())
|
|
|
|
|
|
def compact_path(path: MemoryPath, summary_node_id: str) -> MemoryPath:
|
|
return replace(path, state=MemoryPathState.COMPACTED, compacted_summary_id=summary_node_id, updated_at=utc_now_iso())
|
|
|
|
|
|
def path_event(path: MemoryPath, kind: str, *, metadata: dict | None = None) -> MemoryEvent:
|
|
event_id = f"path-event:{stable_digest([path.path_id, kind, path.event_ids, metadata or {}])}"
|
|
return MemoryEvent(
|
|
event_id=event_id,
|
|
kind=kind,
|
|
metadata={
|
|
"path_id": path.path_id,
|
|
"parent_path_id": path.parent_path_id,
|
|
"path_state": path.state.value,
|
|
**dict(metadata or {}),
|
|
},
|
|
)
|