Coverage for src/lilbee/app/memory.py: 100%
51 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +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)
31def make_memory_row(
32 text: str,
33 embed: Callable[[str], Vector],
34 *,
35 owner: str = LOCAL_OWNER,
36 kind: MemoryKind = MemoryKind.FACT,
37 source: MemorySource = MemorySource.MANUAL,
38 shared: bool = False,
39) -> MemoryRow:
40 """Build a fully populated ``MemoryRow`` with a fresh id, timestamps, and
41 an embedded vector. The single id/timestamp/embedding assignment point, so
42 callers supply only their own ``embed`` and store the result.
43 """
44 now = datetime.now(UTC).isoformat()
45 return MemoryRow(
46 id=uuid.uuid4().hex,
47 owner=owner,
48 shared=shared,
49 kind=kind,
50 source=source,
51 text=text,
52 # MemoryRow serializes to JSON on write, which has no ndarray encoding.
53 vector=embed(text).tolist(),
54 created_at=now,
55 updated_at=now,
56 )
59MEMORY_DISABLED_HINT = (
60 "Memory is off. Turn it on with /set memory_enabled true in the TUI, "
61 "settings_set via MCP, or memory_enabled = true in config.toml."
62)
65def memory_enabled() -> bool:
66 """True when the memory subsystem is switched on (off by default)."""
67 return cfg.memory_enabled
70def remember(
71 text: str,
72 *,
73 owner: str = LOCAL_OWNER,
74 kind: MemoryKind = MemoryKind.FACT,
75 source: MemorySource = MemorySource.MANUAL,
76 shared: bool = False,
77) -> str:
78 """Embed *text* and store it as a memory; returns the stored id."""
79 services = get_services()
80 record = make_memory_row(
81 text,
82 services.embedder.embed,
83 owner=owner,
84 kind=kind,
85 source=source,
86 shared=shared,
87 )
88 return services.store.add_memory(record)
91def recall(query: str, owner: str = LOCAL_OWNER, *, top_k: int | None = None) -> list[MemoryRow]:
92 """Recall facts for *owner*.
94 For the human this includes memories an agent shared (so shared agent
95 knowledge informs answers); for an agent it includes the human's shared
96 facts. The management list (:func:`list_memories`) stays narrower.
97 """
98 services = get_services()
99 predicate = human_recall_predicate() if owner == LOCAL_OWNER else agent_recall_predicate(owner)
100 return services.store.search_memories(
101 services.embedder.embed_query(query),
102 owner_predicate=predicate,
103 top_k=cfg.memory_top_k if top_k is None else top_k,
104 max_distance=cfg.memory_max_distance,
105 )
108def list_memories(owner: str = LOCAL_OWNER) -> list[MemoryRow]:
109 """List the memories *owner* owns and can manage (any kind), newest first.
111 Strictly owner-scoped (unlike :func:`recall`): the management surface shows
112 only rows the caller can delete or re-flag, so it never lists agent-shared
113 memories the human cannot act on.
114 """
115 predicate = (
116 local_owner_predicate() if owner == LOCAL_OWNER else f"owner = '{escape_sql_string(owner)}'"
117 )
118 return get_services().store.get_memories(owner_predicate=predicate)
121def forget(memory_id: str, *, owner: str = LOCAL_OWNER) -> bool:
122 """Delete *owner*'s memory by id; returns True when it existed and was owned.
124 Defaults to the local human's namespace (TUI/CLI/REST/Python API); MCP passes
125 the calling agent's owner so an agent can only delete its own memories.
126 """
127 return get_services().store.delete_memory(memory_id, owner=owner)
130def set_memory_shared(memory_id: str, *, shared: bool, owner: str = LOCAL_OWNER) -> bool:
131 """Set *owner*'s memory shared-with-agents flag; returns True when found and owned."""
132 return get_services().store.update_memory(memory_id, shared=shared, owner=owner)
135def auto_extract_enabled() -> bool:
136 """True when auto-extraction is on (requires the master gate too)."""
137 return cfg.memory_enabled and cfg.memory_auto_extract
140@dataclass(frozen=True, slots=True)
141class SavedMemory:
142 """A memory created by auto-extraction: its stored id, kind, and text."""
144 id: str
145 kind: MemoryKind
146 text: str
149def auto_extract(question: str, answer: str) -> list[SavedMemory]:
150 """Extract durable memories from a chat turn and store them.
152 Returns one :class:`SavedMemory` per stored memory. Stored memories are
153 ``source=EXTRACTED`` and are recalled like any other; the user manages them
154 in ``/memories``. A no-op (returns ``[]``) unless both the master gate and
155 ``memory_auto_extract`` are on.
156 """
157 from lilbee.retrieval.query.memory_extract import extract_memories
159 if not auto_extract_enabled():
160 return []
161 services = get_services()
163 def _chat_text(messages: list[dict[str, str]], **_kwargs: object) -> str:
164 return services.provider.chat(
165 messages, stream=False, options={"response_format": json_reply_format()}
166 ).text
168 extracted = extract_memories(question, answer, _chat_text)
169 stored: list[SavedMemory] = []
170 for memory in extracted:
171 memory_id = remember(memory.text, kind=memory.kind, source=MemorySource.EXTRACTED)
172 stored.append(SavedMemory(id=memory_id, kind=memory.kind, text=memory.text))
173 return stored