@@ -165,6 +165,125 @@ def retrieve_context(query: str, max_tokens: int = 8000) -> str:
165165 return truncate_to_token_limit(context, max_tokens)
166166```
167167
168+ ## 完整实现示例
169+
170+ ``` python
171+ # memory_system.py
172+ from dataclasses import dataclass
173+ from enum import Enum
174+ from typing import Optional
175+ import sqlite3
176+ import time
177+ import pickle
178+
179+ class MemoryType (Enum ):
180+ WORKING = " working"
181+ SESSION = " session"
182+ PERSISTENT = " persistent"
183+
184+ @dataclass
185+ class Memory :
186+ id : Optional[int ]
187+ memory_type: MemoryType
188+ content: str
189+ embedding: list[float ]
190+ created_at: float
191+ workspace_id: str
192+
193+ class MemorySystem :
194+ def __init__ (self , db_path : str ):
195+ self .conn = sqlite3.connect(db_path)
196+ self ._init_db()
197+
198+ def _init_db (self ):
199+ self .conn.execute("""
200+ CREATE TABLE IF NOT EXISTS memories (
201+ id INTEGER PRIMARY KEY,
202+ memory_type TEXT NOT NULL,
203+ content TEXT NOT NULL,
204+ embedding BLOB,
205+ created_at REAL NOT NULL,
206+ workspace_id TEXT NOT NULL
207+ )
208+ """ )
209+ self .conn.commit()
210+
211+ def save_memory (
212+ self ,
213+ memory_type : MemoryType,
214+ content : str ,
215+ embedding : list[float ],
216+ workspace_id : str
217+ ) -> int :
218+ """ 保存记忆"""
219+ cursor = self .conn.execute("""
220+ INSERT INTO memories (memory_type, content, embedding, created_at, workspace_id)
221+ VALUES (?, ?, ?, ?, ?)
222+ """ , (
223+ memory_type.value,
224+ content,
225+ pickle.dumps(embedding),
226+ time.time(),
227+ workspace_id
228+ ))
229+ self .conn.commit()
230+ return cursor.lastrowid
231+
232+ def retrieve (
233+ self ,
234+ query_embedding : list[float ],
235+ workspace_id : str ,
236+ memory_types : list[MemoryType] = None ,
237+ top_k : int = 5
238+ ) -> list[Memory]:
239+ """ 检索记忆"""
240+ if memory_types is None :
241+ memory_types = [MemoryType.PERSISTENT ]
242+
243+ type_values = [t.value for t in memory_types]
244+ placeholders = " ," .join(" ?" * len (type_values))
245+
246+ cursor = self .conn.execute(f """
247+ SELECT id, memory_type, content, embedding, created_at
248+ FROM memories
249+ WHERE workspace_id = ? AND memory_type IN ( { placeholders} )
250+ ORDER BY created_at DESC
251+ LIMIT ?
252+ """ , [workspace_id] + type_values + [top_k])
253+
254+ return [
255+ Memory(
256+ id = row[0 ],
257+ memory_type = MemoryType(row[1 ]),
258+ content = row[2 ],
259+ embedding = pickle.loads(row[3 ]) if row[3 ] else [],
260+ created_at = row[4 ],
261+ workspace_id = workspace_id
262+ )
263+ for row in cursor
264+ ]
265+
266+ def build_context (self , query : str , workspace_id : str ) -> str :
267+ """ 构建完整上下文"""
268+ query_embedding = self ._simple_embed(query)
269+
270+ # 检索持久记忆 + 会话记忆
271+ persistent = self .retrieve(query_embedding, workspace_id, [MemoryType.PERSISTENT ], top_k = 3 )
272+ session = self .retrieve(query_embedding, workspace_id, [MemoryType.SESSION ], top_k = 5 )
273+
274+ parts = [" ## 相关记忆" ]
275+ for m in persistent + session:
276+ parts.append(f " - [ { m.memory_type.value} ] { m.content[:200 ]} " )
277+
278+ return " \n " .join(parts)
279+
280+ def _simple_embed (self , text : str ) -> list[float ]:
281+ """ 简化版 embedding"""
282+ import hashlib
283+ h = hashlib.sha256(text.encode()).digest()
284+ return [b / 255.0 for b in h[:32 ]]
285+ ```
286+
168287## LoopForge 的记忆设计
169288
170289作为本地优先的 Agent OS,LoopForge 实现了完整的多层记忆系统:
@@ -238,6 +357,6 @@ Agent 的记忆不是"记住所有事情",而是**在对的时刻想起对的
238357
239358** 相关链接**
240359
241- - [ LoopForge 记忆模块源码] ( ../explanation/memory-subsystem.md )
242- - [ 向量检索最佳实践 ] ( ../how-to/vector-search .md )
360+ - [ LoopForge 记忆模块源码(rexos-memory) ] ( https://github.com/rexleimo/LoopForge/tree/main/meos/crates/rexos-memory )
361+ - [ LoopForge 概念与记忆模型 ] ( ../explanation/concepts .md )
243362- [ Harness 教程:长任务执行] ( ../tutorials/harness-long-task.md )
0 commit comments