Coverage for src/lilbee/app/memory.py: 100%
52 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-08 09:20 +0000
1"""Use-case orchestration for long-term chat memory, shared by every surface.
3Surfaces (TUI, CLI, MCP, REST, Python API) call these functions rather than
4constructing ``MemoryRow`` objects or building owner predicates themselves, so
5embedding, id/timestamp assignment, and scoping live in one place.
6"""
8from __future__ import annotations
10import uuid
11from collections.abc import Callable
12from dataclasses import dataclass
13from datetime import UTC, datetime
15from lilbee.app.services import get_services
16from lilbee.core.config import cfg
17from lilbee.core.llm_json import json_reply_format
18from lilbee.core.vectors import Vector
19from lilbee.data.store import (
20 LOCAL_OWNER,
21 MemoryKind,
22 MemoryRow,
23 MemorySource,
24 agent_recall_predicate,
25 escape_sql_string,
26 human_recall_predicate,
27 local_owner_predicate,
28)
29from lilbee.providers.base import aux_options
32def make_memory_row(
33 text: str,
34 embed: Callable[[str], Vector],
35 *,
36 owner: str = LOCAL_OWNER,
37 kind: MemoryKind = MemoryKind.FACT,
38 source: MemorySource = MemorySource.MANUAL,
39 shared: bool = False,
40) -> MemoryRow:
41 """Build a fully populated ``MemoryRow`` with a fresh id, timestamps, and
42 an embedded vector. The single id/timestamp/embedding assignment point, so
43 callers supply only their own ``embed`` and store the result.
44 """
45 now = datetime.now(UTC).isoformat()
46 return MemoryRow(
47 id=uuid.uuid4().hex,
48 owner=owner,
49 shared=shared,
50 kind=kind,
51 source=source,
52 text=text,
53 # MemoryRow serializes to JSON on write, which has no ndarray encoding.
54 vector=embed(text).tolist(),
55 created_at=now,
56 updated_at=now,
57 )
60MEMORY_DISABLED_HINT = (
61 "Memory is off. Turn it on with /set memory_enabled true in the TUI, "
62 "settings_set via MCP, or memory_enabled = true in config.toml."
63)
66def memory_enabled() -> bool:
67 """True when the memory subsystem is switched on (off by default)."""
68 return cfg.memory_enabled
71def remember(
72 text: str,
73 *,
74 owner: str = LOCAL_OWNER,
75 kind: MemoryKind = MemoryKind.FACT,
76 source: MemorySource = MemorySource.MANUAL,
77 shared: bool = False,
78) -> str:
79 """Embed *text* and store it as a memory; returns the stored id."""
80 services = get_services()
81 record = make_memory_row(
82 text,
83 services.embedder.embed,
84 owner=owner,
85 kind=kind,
86 source=source,
87 shared=shared,
88 )
89 return services.store.add_memory(record)
92def recall(query: str, owner: str = LOCAL_OWNER, *, top_k: int | None = None) -> list[MemoryRow]:
93 """Recall facts for *owner*.
95 For the human this includes memories an agent shared (so shared agent
96 knowledge informs answers); for an agent it includes the human's shared
97 facts. The management list (:func:`list_memories`) stays narrower.
98 """
99 services = get_services()
100 predicate = human_recall_predicate() if owner == LOCAL_OWNER else agent_recall_predicate(owner)
101 return services.store.search_memories(
102 services.embedder.embed_query(query),
103 owner_predicate=predicate,
104 top_k=cfg.memory_top_k if top_k is None else top_k,
105 max_distance=cfg.memory_max_distance,
106 )
109def list_memories(owner: str = LOCAL_OWNER) -> list[MemoryRow]:
110 """List the memories *owner* owns and can manage (any kind), newest first.
112 Strictly owner-scoped (unlike :func:`recall`): the management surface shows
113 only rows the caller can delete or re-flag, so it never lists agent-shared
114 memories the human cannot act on.
115 """
116 predicate = (
117 local_owner_predicate() if owner == LOCAL_OWNER else f"owner = '{escape_sql_string(owner)}'"
118 )
119 return get_services().store.get_memories(owner_predicate=predicate)
122def forget(memory_id: str, *, owner: str = LOCAL_OWNER) -> bool:
123 """Delete *owner*'s memory by id; returns True when it existed and was owned.
125 Defaults to the local human's namespace (TUI/CLI/REST/Python API); MCP passes
126 the calling agent's owner so an agent can only delete its own memories.
127 """
128 return get_services().store.delete_memory(memory_id, owner=owner)
131def set_memory_shared(memory_id: str, *, shared: bool, owner: str = LOCAL_OWNER) -> bool:
132 """Set *owner*'s memory shared-with-agents flag; returns True when found and owned."""
133 return get_services().store.update_memory(memory_id, shared=shared, owner=owner)
136def auto_extract_enabled() -> bool:
137 """True when auto-extraction is on (requires the master gate too)."""
138 return cfg.memory_enabled and cfg.memory_auto_extract
141@dataclass(frozen=True, slots=True)
142class SavedMemory:
143 """A memory created by auto-extraction: its stored id, kind, and text."""
145 id: str
146 kind: MemoryKind
147 text: str
150def auto_extract(question: str, answer: str) -> list[SavedMemory]:
151 """Extract durable memories from a chat turn and store them.
153 Returns one :class:`SavedMemory` per stored memory. Stored memories are
154 ``source=EXTRACTED`` and are recalled like any other; the user manages them
155 in ``/memories``. A no-op (returns ``[]``) unless both the master gate and
156 ``memory_auto_extract`` are on.
157 """
158 from lilbee.retrieval.query.memory_extract import MEMORY_EXTRACT_MAX_TOKENS, extract_memories
160 if not auto_extract_enabled():
161 return []
162 services = get_services()
164 def _chat_text(messages: list[dict[str, str]], **_kwargs: object) -> str:
165 return services.provider.chat(
166 messages,
167 stream=False,
168 options=aux_options(MEMORY_EXTRACT_MAX_TOKENS, response_format=json_reply_format()),
169 ).text
171 extracted = extract_memories(question, answer, _chat_text)
172 stored: list[SavedMemory] = []
173 for memory in extracted:
174 memory_id = remember(memory.text, kind=memory.kind, source=MemorySource.EXTRACTED)
175 stored.append(SavedMemory(id=memory_id, kind=memory.kind, text=memory.text))
176 return stored