•【AI大模型应用开发】以LangChain为例：从短期记忆实战，到如何让AI应用保持长期记忆的探索
•【AI大模型应用开发】MemGPT原理与快速上手：这可能是目前管理大模型记忆的最专业的框架和思路
在MetaGPT中，Memory类是智能体的记忆的抽象。当初始化时，Role初始化一个Memory对象作为self.rc.memory属性，它将在之后的_observe中存储每个Message，以便后续的检索。简而言之，Role的记忆是一个含有Message的列表。
get_memories函数就是用来获取记忆的。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;border-left: none;padding: 1em;border-radius: 8px;color: rgba(0, 0, 0, 0.5);background: rgb(247, 247, 247);margin: 2em 8px;">你可以通过前面MetaGPT系列文章中的任一个Demo代码去体验这个函数的使用。
ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 1.2em;font-weight: bold;display: table;margin: 2em auto 1em;padding-right: 1em;padding-left: 1em;border-bottom: 2px solid rgb(15, 76, 129);color: rgb(63, 63, 63);">2. 再深入看源码下图是 MetaGPT 中 Memory 部分的代码文件：
memory.py 是基础的 memory 基类，也是最常用的 Memory 类型。longterm_memory.py 是长时记忆的实现。memory_storage.py 是辅助 longterm_memory.py 来进行记忆的持久化存储和查询的类。brain_memory.py 看描述是具有长时记忆和记忆压缩能力，但是并没有看到应用的地方。
这部分功能在 memory.py 文件中，这是记忆模块的基类。
classMemory(BaseModel):
"""Themostbasicmemory:super-memory"""
storage:list[SerializeAsAny[Message]]=[]
index
efaultDict[str,list[SerializeAsAny[Message]]]=Field(default_factory=lambda:defaultdict(list))
ignore_id:bool=False其 storage 是一个 list 列表，所以是保存在内存中的，关机、重启之后，记忆消失，无法持久化存储。
（1）add：往记忆中添加一条记忆
实现中，还增加了self.index[message.cause_by].append(message)的处理，让使用者可以通过message.cause_by参数快速获取指定动作的记忆。
defadd(self,message:Message):
"""Addanewmessagetostorage,whileupdatingtheindex"""
ifself.ignore_id:
message.id=IGNORED_MESSAGE_ID
ifmessageinself.storage:
return
self.storage.append(message)
ifmessage.cause_by:
self.index[message.cause_by].append(message)（2）add_batch：添加一系列消息
（3）get_by_role：返回指定Role的记忆，判定条件message.role == role
（4）get_by_content：查找带有 content 的记忆，类似关键字检索？判定条件content in message.content
（5）delete_newest：删除最新的一条记忆
（6）delete：删除指定的一条记忆
（7）clear：清空记忆
（8）count：记忆数量
（9）try_remember：查找带有 keyword 的记忆，与get_by_content有点重复？keyword in message.content
（10）get：获取最近的K条记忆，K=0时返回全部的记忆
（11）find_news：找出想观察但上次没观察到的消息？没想象出怎么使用，给大家看下源码理解一下吧：
deffind_news(self,observed:list[Message],k=0)->list[Message]:
"""findnews(previouslyunseenmessages)fromthethemostrecentkmemories,fromallmemorieswhenk=0"""
already_observed=self.get(k)
news:list[Message]=[]
foriinobserved:
ifiinalready_observed:
continue
news.append(i)
returnnews（12）get_by_action 、 get_by_actions：返回指定动作产生的记忆
长时记忆的实现，启动时恢复记忆，变化时更新记忆。
classLongTermMemory(Memory):
"""
TheLong-termmemoryforRoles
-recovermemorywhenitstaruped
-updatememorywhenitchanged
"""
model_config=ConfigDict(arbitrary_types_allowed=True)
memory_storage:MemoryStorage=Field(default_factory=MemoryStorage)
rc:Optional[RoleContext]=None
msg_from_recover:bool=False（1）add：添加一条记忆
defadd(self,message:Message):
super().add(message)
foractioninself.rc.watch:
ifmessage.cause_by==actionandnotself.msg_from_recover:
#currently,onlyaddrole'swatchingmessagestoitsmemory_storage
#andignoreaddingmessagesfromrecoverrepeatedly
self.memory_storage.add(message)首先是调用super也就是基础Memory的add接口，添加记忆到内存中。 然后通过self.memory_storage.add(message)保存记忆。memory_storage的add接口，是调用 FAISS 向量数据库去进行向量化和存储。
defadd(self,message:Message)->bool:
"""addmessageintomemorystorage"""
self.faiss_engine.add_objs([message])
logger.info(f"Role{self.role_id}'smemory_storageaddamessage")（2）recover_memory：恢复记忆
defrecover_memory(self,role_id:str,rc:RoleContext):
self.memory_storage.recover_memory(role_id)
self.rc=rc
ifnotself.memory_storage.is_initialized:
logger.warning(f"ItmaythefirsttimetorunRole{role_id},thelong-termmemoryisempty")
else:
logger.warning(f"Role{role_id}hasexistingmemorystorageandhasrecoveredthem.")
self.msg_from_recover=True
#self.add_batch(messages)#TODOnoneed
self.msg_from_recover=False通过memory_storage来recover_memory，最终也是通过 FAISS 数据库进行记忆查询：
defrecover_memory(self,role_id:str)->list[Message]:
self.role_id=role_id
self.role_mem_path=Path(DATA_PATH/f"role_mem/{self.role_id}/")
self.role_mem_path.mkdir(parents=True,exist_ok=True)
self.cache_dir=self.role_mem_path
ifself.role_mem_path.joinpath("default__vector_store.json").exists():
self.faiss_engine=SimpleEngine.from_index(
index_config=FAISSIndexConfig(persist_path=self.cache_dir),
retriever_configs=[FAISSRetrieverConfig()],
embed_model=self.embedding,
)
else:
self.faiss_engine=SimpleEngine.from_objs(
objs=[],retriever_configs=[FAISSRetrieverConfig()],embed_model=self.embedding
)
self._initialized=True这里恢复的是指定Role的记忆，通过self.role_id来区分。
（3）find_news
与基础Memory不同的是，它首先在基础Memory中找，然后在 long-term memory 中找，将两者的结果结合作为最终结果返回。
asyncdeffind_news(self,observed:list[Message],k=0)->list[Message]:
"""
findnews(previouslyunseenmessages)fromthethemostrecentkmemories,fromallmemorieswhenk=0
1.findtheshort-termmemory(stm)news
2.furthermore,filteroutsimilarmessagesbasedonltm(long-termmemory),getthefinalnews
"""
stm_news=super().find_news(observed,k=k)#shot-termmemorynews
ifnotself.memory_storage.is_initialized:
#memory_storagehasn'tinitialized,usedefault`find_news`togetstm_news
returnstm_news
ltm_news:list[Message]=[]
formeminstm_news:
#filteroutmessagessimilartothoseseenpreviouslyinltm,onlykeepfreshnews
mem_searched=awaitself.memory_storage.search_similar(mem)
iflen(mem_searched)==0:
ltm_news.append(mem)
returnltm_news[-k:]（4）persist
将数据持久化存储。
defpersist(self):
ifself.faiss_engine:
self.faiss_engine.retriever._index.storage_context.persist(self.cache_dir)classBrainMemory(BaseModel):
history
ist[Message]=Field(default_factory=list)
knowledgeist[Message]=Field(default_factory=list)
historical_summary:str=""
last_history_id:str=""
is_dirty:bool=False
last_talk:Optional[str]=None
cacheable:bool=True
llm:Optional[BaseLLM]=Field(default=None,exclude=True)这个Memory没太搞清楚具体是干啥的，但是看到里面是用Redis存储的，并且里面有summary信息的步骤，可以学习一下这种记忆思路。
看源码，它是将历史对话信息用大模型进行总结，然后存储到Redis里面。
asyncdefsummarize(self,llm,max_words=200,keep_language:bool=False,limit:int=-1,**kwargs):
ifisinstance(llm,MetaGPTLLM):
returnawaitself._metagpt_summarize(max_words=max_words)
self.llm=llm
returnawaitself._openai_summarize(llm=llm,max_words=max_words,keep_language=keep_language,limit=limit)
asyncdef_openai_summarize(self,llm,max_words=200,keep_language:bool=False,limit:int=-1):
texts=[self.historical_summary]
forminself.history:
texts.append(m.content)
text="\n".join(texts)
text_length=len(text)
iflimit>0andtext_length<limit:
returntext
summary=awaitself._summarize(text=text,max_words=max_words,keep_language=keep_language,limit=limit)
ifsummary:
awaitself.set_history_summary(history_summary=summary,redis_key=config.redis_key)
returnsummary
raiseValueError(f"texttoolong:{text_length}")它上面还有一层函数 get_title，又将上面Summary的信息抽出了一个标题：
asyncdefget_title(self,llm,max_words=5,**kwargs)->str:
"""Generatetexttitle"""
ifisinstance(llm,MetaGPTLLM):
returnself.history[0].contentifself.historyelse"New"
summary=awaitself.summarize(llm=llm,max_words=500)
language=config.language
command=f"Translatetheabovesummaryintoa{language}titleoflessthan{max_words}words."
summaries=[summary,command]
msg="\n".join(summaries)
logger.debug(f"titleask:{msg}")
response=awaitllm.aask(msg=msg,system_msgs=[],stream=False)
logger.debug(f"titlersp:{response}")
returnresponse这是不是算分层存储了？
总得看下来，MetaGPT的记忆模块其实实现的很简单，主要是在短时记忆上面进行了实现。对于长时记忆，只是简单的实现了将其存入了向量数据库，可以选择持久化的基础能力。对于长上下文这种的记忆能力，并没有特别实现。感觉longterm_memory并没有写完的样子，没看到实际使用。其brain_memory是第一次见，也没看到实际使用，但里面的对历史对话进行总结存储，对总结内容再进行标题化存储的方法值得借鉴。
1.https://docs.deepwisdom.ai/main/zh/guide/tutorials/use_memories.html
2.https://github.com/geekan/MetaGPT
| 欢迎光临 链载Ai (https://www.lianzai.com/) | Powered by Discuz! X3.5 |