Coverage for open_webui/models/memories.py: 45%
166 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1"""Long-term memory storage for per-user context recall."""
3from __future__ import annotations
5import time
6import uuid
7from typing import Literal
9from open_webui.internal.db import Base, get_async_db_context
10from pydantic import BaseModel, ConfigDict
11from sqlalchemy import JSON, BigInteger, Column, Index, String, Text, delete, select
12from sqlalchemy.ext.asyncio import AsyncSession
15class Memory(Base): # user memory store
16 """Stores user-created memory entries linked to a vector collection."""
18 __tablename__ = 'memory'
19 __table_args__ = (Index('ix_memory_id_user_id', 'id', 'user_id'),)
21 id = Column(String, primary_key=True, unique=True)
22 user_id = Column(String, index=True)
23 type = Column(String, default='context', server_default='context', index=True)
24 path = Column(Text, nullable=True)
25 content = Column(Text) # free-form text learned from conversation
26 meta = Column(JSON, nullable=True)
27 updated_at = Column(BigInteger) # epoch seconds
28 created_at = Column(BigInteger) # epoch seconds
31class MemoryModel(BaseModel):
32 """Pydantic mirror of the Memory table row."""
34 id: str
35 user_id: str
36 type: Literal['user', 'context'] = 'context'
37 path: str | None = None
38 content: str
39 meta: dict | None = None
40 updated_at: int # timestamp in epoch
41 created_at: int # timestamp in epoch
42 model_config = ConfigDict(from_attributes=True) # allows ORM mapping
45class MemoriesTable:
46 @staticmethod
47 def normalize_memory_type(memory_type: str | None = None) -> str:
48 return 'user' if memory_type == 'user' else 'context'
50 async def insert_new_memory(
51 self,
52 user_id: str,
53 content: str,
54 memory_type: str | None = None,
55 path: str | None = None,
56 meta: dict | None = None,
57 db: AsyncSession | None = None,
58 ) -> MemoryModel | None:
59 """Persist a new memory entry and return the created model."""
60 async with get_async_db_context(db) as db:
61 now = int(time.time())
62 record = Memory(
63 id=str(uuid.uuid4()),
64 user_id=user_id,
65 type=self.normalize_memory_type(memory_type),
66 path=path,
67 content=content,
68 meta=meta,
69 created_at=now,
70 updated_at=now,
71 )
72 db.add(record)
73 await db.commit()
74 return MemoryModel.model_validate(record) if record else None
76 async def update_memory_by_id_and_user_id(
77 self,
78 id: str,
79 user_id: str,
80 content: str | None,
81 memory_type: str | None = None,
82 path: str | None = None,
83 update_path: bool = False,
84 meta: dict | None = None,
85 db: AsyncSession | None = None,
86 ) -> MemoryModel | None:
87 async with get_async_db_context(db) as db:
88 try:
89 memory = await db.get(Memory, id)
90 if not memory or memory.user_id != user_id:
91 return None
93 if content is not None:
94 memory.content = content
95 if memory_type is not None:
96 memory.type = self.normalize_memory_type(memory_type)
97 if update_path:
98 memory.path = path
99 if meta is not None:
100 memory.meta = {**(memory.meta or {}), **meta}
101 memory.updated_at = int(time.time())
103 await db.commit()
104 return MemoryModel.model_validate(memory)
105 except Exception:
106 return None
108 async def get_memories(self, db: AsyncSession | None = None) -> list[MemoryModel]:
109 async with get_async_db_context(db) as db:
110 try:
111 result = await db.execute(select(Memory))
112 memories = result.scalars().all()
113 return [MemoryModel.model_validate(memory) for memory in memories]
114 except Exception:
115 return None
117 async def get_memories_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> list[MemoryModel]:
118 async with get_async_db_context(db) as db:
119 try:
120 result = await db.execute(select(Memory).filter_by(user_id=user_id))
121 memories = result.scalars().all()
122 return [MemoryModel.model_validate(memory) for memory in memories]
123 except Exception:
124 return None
126 async def get_memory_by_id(self, id: str, db: AsyncSession | None = None) -> MemoryModel | None:
127 async with get_async_db_context(db) as db:
128 try:
129 memory = await db.get(Memory, id)
130 return MemoryModel.model_validate(memory) if memory else None
131 except Exception:
132 return None
134 async def delete_memory_by_id(self, id: str, db: AsyncSession | None = None) -> bool:
135 async with get_async_db_context(db) as db:
136 try:
137 await db.execute(delete(Memory).filter_by(id=id))
138 await db.commit()
140 return True
142 except Exception:
143 return False
145 async def delete_memories_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> bool:
146 async with get_async_db_context(db) as db:
147 try:
148 await db.execute(delete(Memory).filter_by(user_id=user_id))
149 await db.commit()
151 return True
152 except Exception:
153 return False
155 async def delete_memory_by_id_and_user_id(self, id: str, user_id: str, db: AsyncSession | None = None) -> bool:
156 async with get_async_db_context(db) as db:
157 try:
158 memory = await db.get(Memory, id)
159 if not memory or memory.user_id != user_id:
160 return False
162 await db.delete(memory)
163 await db.commit()
164 return True
165 except Exception:
166 return False
168 async def apply_memory_operations(
169 self,
170 user_id: str,
171 operations: list[dict],
172 db: AsyncSession | None = None,
173 ) -> list[dict]:
174 now = int(time.time())
175 results: list[dict] = []
177 async with get_async_db_context(db) as db:
178 for operation in operations: 178 ↛ 263line 178 didn't jump to line 263 because the loop on line 178 didn't complete
179 action = operation.get('action')
181 if action == 'add':
182 content = operation.get('content', '').strip()
183 memory_type = self.normalize_memory_type(operation.get('type'))
184 path = operation.get('path')
185 result = await db.execute(
186 select(Memory).filter_by(user_id=user_id, content=content, type=memory_type, path=path)
187 )
188 existing = result.scalars().first()
189 if existing:
190 results.append(
191 {
192 'action': action,
193 'status': 'skipped',
194 'memory': MemoryModel.model_validate(existing),
195 'reason': 'duplicate',
196 }
197 )
198 continue
200 memory = Memory(
201 id=str(uuid.uuid4()),
202 user_id=user_id,
203 type=memory_type,
204 path=path,
205 content=content,
206 meta=operation.get('meta'),
207 created_at=now,
208 updated_at=now,
209 )
210 db.add(memory)
211 await db.flush()
212 results.append(
213 {'action': action, 'status': 'created', 'memory': MemoryModel.model_validate(memory)}
214 )
216 elif action == 'replace': 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 memory_id = operation.get('id')
218 content = operation.get('content', '').strip()
219 memory = await db.get(Memory, memory_id)
220 if not memory or memory.user_id != user_id:
221 raise ValueError(f'Memory not found: {memory_id}')
223 memory.content = content
224 if operation.get('type') is not None:
225 memory.type = self.normalize_memory_type(operation.get('type'))
226 if 'path' in operation:
227 memory.path = operation.get('path')
228 if operation.get('meta') is not None:
229 memory.meta = {**(memory.meta or {}), **operation.get('meta')}
230 memory.updated_at = now
231 await db.flush()
232 results.append(
233 {'action': action, 'status': 'updated', 'memory': MemoryModel.model_validate(memory)}
234 )
236 elif action == 'move': 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true
237 memory_id = operation.get('id')
238 memory = await db.get(Memory, memory_id)
239 if not memory or memory.user_id != user_id:
240 raise ValueError(f'Memory not found: {memory_id}')
242 memory.path = operation.get('path')
243 if operation.get('meta') is not None:
244 memory.meta = {**(memory.meta or {}), **operation.get('meta')}
245 memory.updated_at = now
246 await db.flush()
247 results.append(
248 {'action': action, 'status': 'updated', 'memory': MemoryModel.model_validate(memory)}
249 )
251 elif action == 'remove': 251 ↛ 261line 251 didn't jump to line 261 because the condition on line 251 was always true
252 memory_id = operation.get('id')
253 memory = await db.get(Memory, memory_id)
254 if not memory or memory.user_id != user_id:
255 raise ValueError(f'Memory not found: {memory_id}')
257 await db.delete(memory)
258 results.append({'action': action, 'status': 'deleted', 'id': memory_id})
260 else:
261 raise ValueError(f'Unsupported memory operation: {action}')
263 await db.commit()
265 return results
268Memories = MemoriesTable() # user memory registry