Coverage for src / kemi / procedures.py: 100%

1"""Procedural memory helpers for kemi. 

2 

3Procedural memories represent *how-to* knowledge — step-by-step instructions, 

4workflows, recipes, or standard operating procedures. They are distinct from 

5*episodic* (event-based) and *semantic* (fact-based) memories. 

6 

7Use procedural memory when you want an agent to recall reusable action sequences 

8rather than isolated facts or past events. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import TYPE_CHECKING, Any 

14 

15from kemi.models import LifecycleState, MemoryType 

16 

17if TYPE_CHECKING: 

18 from kemi.core import Memory 

19 from kemi.models import MemoryObject 

20 

21 

22def remember_procedure( 

23 memory: "Memory", 

24 *, 

25 user_id: str | None = None, 

26 name: str, 

27 steps: list[str], 

28 metadata: dict[str, Any] | None = None, 

29 namespace: str = "default", 

30 importance: float = 0.7, 

31 agent_id: str | None = None, 

32 session_id: str | None = None, 

33) -> str: 

34 """Store a step-by-step procedure as a procedural memory. 

35 

36 Joins *steps* into a concise content string, tags the memory with 

37 ``["procedure", <name>]``, and sets ``memory_type=MemoryType.PROCEDURAL``. 

38 

39 Args: 

40 memory: A kemi :class:`~kemi.core.Memory` instance. 

41 user_id: User to associate the procedure with. If ``None``, a generic 

42 system user (``"_system"``) is used. 

43 name: Short identifier for the procedure (e.g. ``"onboarding_flow"``). 

44 steps: Ordered list of action strings. 

45 metadata: Optional extra metadata dict. 

46 namespace: Memory namespace. 

47 importance: Importance score (0.0-1.0). Defaults to 0.7 because 

48 procedures are usually high-value reusable knowledge. 

49 agent_id: Optional agent identifier. 

50 session_id: Optional session identifier. 

51 

52 Returns: 

53 The memory ID of the stored procedure. 

54 

55 Example: 

56 >>> procedure_id = remember_procedure( 

57 ... memory, 

58 ... user_id="alice", 

59 ... name="password_reset", 

60 ... steps=[ 

61 ... "Ask the user for their email address", 

62 ... "Send a reset link to the verified email", 

63 ... "Confirm the reset was initiated", 

64 ... ], 

65 ... ) 

66 """ 

67 if not steps: 

68 raise ValueError("steps cannot be empty — there is nothing to remember") 

69 if not name or not name.strip(): 

70 raise ValueError("name cannot be empty") 

71 

72 uid = user_id if user_id is not None else "_system" 

73 # Join steps into a concise, readable block 

74 step_lines = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(steps)) 

75 content = f"Procedure: {name}\n{step_lines}" 

76 

77 merged_meta = metadata.copy() if metadata else {} 

78 merged_meta["procedure_name"] = name 

79 merged_meta["step_count"] = len(steps) 

80 

81 return memory.remember( 

82 user_id=uid, 

83 content=content, 

84 memory_type=MemoryType.PROCEDURAL, 

85 tags=["procedure", name], 

86 metadata=merged_meta, 

87 namespace=namespace, 

88 importance=importance, 

89 agent_id=agent_id, 

90 session_id=session_id, 

91 ) 

92 

93 

94def recall_procedures( 

95 memory: "Memory", 

96 query: str, 

97 *, 

98 user_id: str | None = None, 

99 namespace: str = "default", 

100 top_k: int = 10, 

101 lifecycle_filter: list[LifecycleState] | None = None, 

102 session_id: str | None = None, 

103) -> list["MemoryObject"]: 

104 """Recall procedural memories relevant to a query. 

105 

106 Performs a semantic search and filters the results to only 

107 ``memory_type=PROCEDURAL``. 

108 

109 Args: 

110 memory: A kemi :class:`~kemi.core.Memory` instance. 

111 query: Natural-language query (e.g. ``"how do I reset a password?"``). 

112 user_id: If provided, scope the search to this user. If ``None``, 

113 the search is scoped to the generic ``"_system"`` user. 

114 namespace: Memory namespace. 

115 top_k: Maximum number of procedures to return. 

116 lifecycle_filter: Optional lifecycle states to include. Defaults to 

117 ``[ACTIVE, DECAYING, ARCHIVED]``. 

118 session_id: Optional session ID to scope the search. 

119 

120 Returns: 

121 List of :class:`~kemi.models.MemoryObject` instances with 

122 ``memory_type=PROCEDURAL``, sorted by relevance score. 

123 

124 Example: 

125 >>> results = recall_procedures( 

126 ... memory, 

127 ... "password reset", 

128 ... user_id="alice", 

129 ... top_k=3, 

130 ... ) 

131 >>> for proc in results: 

132 ... print(proc.content) 

133 """ 

134 if not query or not query.strip(): 

135 raise ValueError("query cannot be empty — what procedure should kemi search for?") 

136 if top_k < 1: 

137 raise ValueError(f"top_k must be at least 1, got {top_k}") 

138 

139 uid = user_id if user_id is not None else "_system" 

140 

141 if lifecycle_filter is None: 

142 lifecycle_filter = [ 

143 LifecycleState.ACTIVE, 

144 LifecycleState.DECAYING, 

145 LifecycleState.ARCHIVED, 

146 ] 

147 

148 results = memory.recall( 

149 user_id=uid, 

150 query=query, 

151 top_k=top_k * 3, # fetch extra to survive the post-hoc type filter 

152 lifecycle_filter=lifecycle_filter, 

153 namespace=namespace, 

154 session_id=session_id, 

155 ) 

156 

157 procedures = [m for m in results if m.memory_type == MemoryType.PROCEDURAL] 

158 return procedures[:top_k]