1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
from datetime import datetime
from uuid import UUID
from personal_agent.database.database import Database
from personal_agent.memory.models import LongTermMemory, MemoryCategory
class LongTermMemoryStore:
def __init__(self, database: Database) -> None:
self.database = database
def add(self, memory: LongTermMemory) -> None:
created_at = memory.created_at.isoformat().replace("+00:00", "Z")
with self.database.connect() as connection:
connection.execute(
"""
INSERT INTO memories (id, content, category, created_at, importance)
VALUES (?, ?, ?, ?, ?)
""",
(
str(memory.id),
memory.content,
memory.category.value,
created_at,
memory.importance,
),
)
connection.execute(
"""
INSERT INTO memories_fts(rowid, content)
SELECT rowid, content
FROM memories
WHERE id = ?
""",
(str(memory.id),),
)
def get(self, memory_id: UUID) -> LongTermMemory | None:
with self.database.connect() as connection:
row = connection.execute(
"""
SELECT id, content, category, created_at, importance
FROM memories
WHERE id = ?
""",
(str(memory_id),),
).fetchone()
if row is None:
return None
return self._from_row(row)
def list(self) -> list[LongTermMemory]:
with self.database.connect() as connection:
rows = connection.execute(
"""
SELECT id, content, category, created_at, importance
FROM memories
ORDER BY id
"""
).fetchall()
return [self._from_row(row) for row in rows]
def delete(self, memory_id: UUID) -> None:
with self.database.connect() as connection:
connection.execute(
"""
DELETE FROM memories_fts
WHERE rowid = (
SELECT rowid
FROM memories
WHERE id = ?
)
""",
(str(memory_id),),
)
connection.execute(
"DELETE FROM memories WHERE id = ?",
(str(memory_id),),
)
def search(
self,
query: str,
category: MemoryCategory | None = None,
limit: int = 10,
) -> list[LongTermMemory]:
if limit <= 0:
raise ValueError("Search limit must be positive")
if not query.strip():
raise ValueError("Search query must not be empty")
terms = query.split()
fts_query = " OR ".join(
f'"{term.replace(chr(34), chr(34) * 2)}"' for term in terms
)
sql = """
SELECT m.id, m.content, m.category, m.created_at, m.importance
FROM memories_fts as f
JOIN memories AS m ON m.rowid = f.rowid
WHERE memories_fts MATCH ?
"""
parameters: list[str | int] = [fts_query]
if category is not None:
sql += " AND m.category = ?"
parameters.append(category.value)
sql += " ORDER BY bm25(memories_fts) LIMIT ?"
parameters.append(limit)
with self.database.connect() as connection:
rows = connection.execute(sql, parameters).fetchall()
return [self._from_row(row) for row in rows]
@staticmethod
def _from_row(row: tuple[str, str, str, str, int]) -> LongTermMemory:
memory_id, content, category, created_at, importance = row
return LongTermMemory(
id=UUID(memory_id),
content=content,
category=MemoryCategory(category),
created_at=datetime.fromisoformat(created_at),
importance=importance,
)
|
