gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

better：优化工作记忆

Browse Source
This commit is contained in:
SengokuCola
2025-06-02 23:05:15 +08:00
parent 5e47efe987
commit 258512e11f
5 changed files with 34 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/chat/focus_chat/info/workingmemory_info.py
View File

@@ -18,30 +18,28 @@ class WorkingMemoryInfo(InfoBase):
self.data["talking_message"] = message
def set_working_memory(self, working_memory: List[str]) -> None:
"""设置工作记忆
"""设置工作记忆列表
Args:
working_memory (str): 工作记忆内容
working_memory (List[str]): 工作记忆内容列表
"""
self.data["working_memory"] = working_memory
def add_working_memory(self, working_memory: str) -> None:
"""添加工作记忆
"""添加一条工作记忆
Args:
working_memory (str): 工作记忆内容
working_memory (str): 工作记忆内容，格式为"记忆要点:xxx"
"""
working_memory_list = self.data.get("working_memory", [])
# print(f"working_memory_list: {working_memory_list}")
working_memory_list.append(working_memory)
# print(f"working_memory_list: {working_memory_list}")
self.data["working_memory"] = working_memory_list
def get_working_memory(self) -> List[str]:
"""获取工作记忆
"""获取所有工作记忆
Returns:
List[str]: 工作记忆内容
List[str]: 工作记忆内容列表，每条记忆格式为"记忆要点:xxx"
"""
return self.data.get("working_memory", [])
@@ -53,33 +51,32 @@ class WorkingMemoryInfo(InfoBase):
"""
return self.type
def get_data(self) -> Dict[str, str]:
def get_data(self) -> Dict[str, List[str]]:
"""获取所有信息数据
Returns:
Dict[str, str]: 包含所有信息数据的字典
Dict[str, List[str]]: 包含所有信息数据的字典
"""
return self.data
def get_info(self, key: str) -> Optional[str]:
def get_info(self, key: str) -> Optional[List[str]]:
"""获取特定属性的信息
Args:
key: 要获取的属性键名
Returns:
Optional[str]: 属性值，如果键不存在则返回 None
Optional[List[str]]: 属性值，如果键不存在则返回 None
"""
return self.data.get(key)
def get_processed_info(self) -> Dict[str, str]:
def get_processed_info(self) -> str:
"""获取处理后的信息
Returns:
Dict[str, str]: 处理后的信息数据
str: 处理后的信息数据，所有记忆要点按行拼接
"""
all_memory = self.get_working_memory()
# print(f"all_memory: {all_memory}")
memory_str = ""
for memory in all_memory:
memory_str += f"{memory}\n"

19
src/chat/focus_chat/info_processors/working_memory_processor.py
View File

@@ -45,7 +45,6 @@ def init_prompt():
"selected_memory_ids": ["id1", "id2", ...],
"new_memory": "true" or "false",
"merge_memory": [["id1", "id2"], ["id3", "id4"],...]
}}
```
"""
@@ -104,11 +103,10 @@ class WorkingMemoryProcessor(BaseProcessor):
all_memory = working_memory.get_all_memories()
memory_prompts = []
for memory in all_memory:
# memory_content = memory.data
memory_summary = memory.summary
memory_id = memory.id
memory_brief = memory_summary.get("brief")
memory_keypoints = memory_summary.get("key_points", [])
memory_points = memory_summary.get("points", [])
memory_single_prompt = f"记忆id:{memory_id},记忆摘要:{memory_brief}\n"
memory_prompts.append(memory_single_prompt)
@@ -122,11 +120,11 @@ class WorkingMemoryProcessor(BaseProcessor):
memory_str=memory_choose_str,
)
print(f"prompt: {prompt}")
# 调用LLM处理记忆
content = ""
try:
# logger.debug(f"{self.log_prefix} 处理工作记忆的prompt: {prompt}")
content, _ = await self.llm_model.generate_response_async(prompt=prompt)
if not content:
logger.warning(f"{self.log_prefix} LLM返回空结果，处理工作记忆失败。")
@@ -159,13 +157,12 @@ class WorkingMemoryProcessor(BaseProcessor):
for memory_id in selected_memory_ids:
memory = await working_memory.retrieve_memory(memory_id)
if memory:
# memory_content = memory.data
memory_summary = memory.summary
memory_id = memory.id
memory_brief = memory_summary.get("brief")
memory_keypoints = memory_summary.get("key_points", [])
for keypoint in memory_keypoints:
memory_str += f"记忆要点:{keypoint}\n"
memory_points = memory_summary.get("points", [])
for point in memory_points:
memory_str += f"记忆要点:{point}\n"
working_memory_info = WorkingMemoryInfo()
if memory_str:
@@ -216,9 +213,7 @@ class WorkingMemoryProcessor(BaseProcessor):
merged_memory = await working_memory.merge_memory(memory_id1, memory_id2)
logger.debug(f"{self.log_prefix} 异步合并记忆成功: {memory_id1} 和 {memory_id2}...")
logger.debug(f"{self.log_prefix} 合并后的记忆梗概: {merged_memory.summary.get('brief')}")
logger.debug(f"{self.log_prefix} 合并后的记忆详情: {merged_memory.summary.get('detailed')}")
logger.debug(f"{self.log_prefix} 合并后的记忆要点: {merged_memory.summary.get('key_points')}")
logger.debug(f"{self.log_prefix} 合并后的记忆事件: {merged_memory.summary.get('events')}")
logger.debug(f"{self.log_prefix} 合并后的记忆要点: {merged_memory.summary.get('points')}")
except Exception as e:
logger.error(f"{self.log_prefix} 异步合并记忆失败: {e}")

28
src/chat/focus_chat/working_memory/memory_item.py
View File

@@ -7,14 +7,14 @@ import string
class MemoryItem:
"""记忆项类，用于存储单个记忆的所有相关信息"""
def __init__(self, data: Any, from_source: str = "", tags: Optional[List[str]] = None):
def __init__(self, data: Any, from_source: str = "", brief: str = ""):
"""
初始化记忆项
Args:
data: 记忆数据
from_source: 数据来源
tags: 数据标签列表
brief: 记忆内容主题
"""
# 生成可读ID：时间戳_随机字符串
timestamp = int(time.time())
@@ -23,11 +23,10 @@ class MemoryItem:
self.data = data
self.data_type = type(data)
self.from_source = from_source
self.tags = set(tags) if tags else set()
self.brief = brief
self.timestamp = time.time()
# 修改summary的结构说明，用于存储可能的总结信息
# summary结构：{
# "brief": "记忆内容主题",
# "detailed": "记忆内容概括",
# "keypoints": ["关键概念1", "关键概念2"],
# "events": ["事件1", "事件2"]
@@ -47,23 +46,6 @@ class MemoryItem:
# 格式: [(操作类型, 时间戳, 当时精简次数, 当时强度), ...]
self.history = [("create", self.timestamp, self.compress_count, self.memory_strength)]
def add_tag(self, tag: str) -> None:
"""添加标签"""
self.tags.add(tag)
def remove_tag(self, tag: str) -> None:
"""移除标签"""
if tag in self.tags:
self.tags.remove(tag)
def has_tag(self, tag: str) -> bool:
"""检查是否有特定标签"""
return tag in self.tags
def has_all_tags(self, tags: List[str]) -> bool:
"""检查是否有所有指定的标签"""
return all(tag in self.tags for tag in tags)
def matches_source(self, source: str) -> bool:
"""检查来源是否匹配"""
return self.from_source == source
@@ -103,9 +85,9 @@ class MemoryItem:
current_time = time.time()
self.history.append((operation_type, current_time, self.compress_count, self.memory_strength))
def to_tuple(self) -> Tuple[Any, str, Set[str], float, str]:
def to_tuple(self) -> Tuple[Any, str, float, str]:
"""转换为元组格式（为了兼容性）"""
return (self.data, self.from_source, self.tags, self.timestamp, self.id)
return (self.data, self.from_source, self.timestamp, self.id)
def is_memory_valid(self) -> bool:
"""检查记忆是否有效（强度是否大于等于1）"""

45
src/chat/focus_chat/working_memory/memory_manager.py
View File

@@ -71,14 +71,13 @@ class MemoryManager:
return memory_item.id
async def push_with_summary(self, data: T, from_source: str = "", tags: Optional[List[str]] = None) -> MemoryItem:
async def push_with_summary(self, data: T, from_source: str = "") -> MemoryItem:
"""
推送一段有类型的信息到工作记忆中，并自动生成总结
Args:
data: 要存储的数据
from_source: 数据来源
tags: 数据标签列表
Returns:
包含原始数据和总结信息的字典
@@ -88,11 +87,8 @@ class MemoryManager:
# 先生成总结
summary = await self.summarize_memory_item(data)
# 准备标签
memory_tags = list(tags) if tags else []
# 创建记忆项
memory_item = MemoryItem(data, from_source, memory_tags)
memory_item = MemoryItem(data, from_source, brief=summary.get("brief", ""))
# 将总结信息保存到记忆项中
memory_item.set_summary(summary)
@@ -103,7 +99,7 @@ class MemoryManager:
return memory_item
else:
# 非字符串类型，直接创建并推送记忆项
memory_item = MemoryItem(data, from_source, tags)
memory_item = MemoryItem(data, from_source)
self.push_item(memory_item)
return memory_item
@@ -136,7 +132,6 @@ class MemoryManager:
self,
data_type: Optional[Type] = None,
source: Optional[str] = None,
tags: Optional[List[str]] = None,
start_time: Optional[float] = None,
end_time: Optional[float] = None,
memory_id: Optional[str] = None,
@@ -150,7 +145,6 @@ class MemoryManager:
Args:
data_type: 要查找的数据类型
source: 数据来源
tags: 必须包含的标签列表
start_time: 开始时间戳
end_time: 结束时间戳
memory_id: 特定记忆项ID
@@ -191,10 +185,6 @@ class MemoryManager:
if source is not None and not item.matches_source(source):
continue
# 检查标签是否匹配
if tags is not None and not item.has_all_tags(tags):
continue
# 检查时间范围
if start_time is not None and item.timestamp < start_time:
continue
@@ -491,9 +481,6 @@ class MemoryManager:
if not memory_item1 or not memory_item2:
raise ValueError("无法找到指定的记忆项")
content1 = memory_item1.data
content2 = memory_item2.data
# 获取记忆的摘要信息（如果有）
summary1 = memory_item1.summary
summary2 = memory_item2.summary
@@ -515,31 +502,22 @@ class MemoryManager:
prompt += f"记忆2主题：{summary2['brief']}\n"
prompt += "记忆2关键要点：\n" + "\n".join([f"- {point}" for point in summary2.get("points", [])]) + "\n\n"
# 添加记忆原始内容
prompt += f"""
记忆1原始内容：
{content1}
记忆2原始内容：
{content2}
prompt += """
请按以下JSON格式输出合并结果：
```json
{{
"content": "合并后的记忆内容文本（尽可能保留原信息，但去除重复）",
{
"brief": "合并后的主题（20字以内）",
"points": [
"合并后的要点",
"合并后的要点"
]
}}
}
```
请确保输出是有效的JSON格式，不要添加任何额外的说明或解释。
"""
# 默认合并结果
default_merged = {
"content": f"{content1}\n\n{content2}",
"brief": f"合并：{summary1['brief']} + {summary2['brief']}",
"points": [],
}
@@ -579,10 +557,6 @@ class MemoryManager:
logger.error(f"修复后的JSON不是字典类型: {type(merged_data)}")
merged_data = default_merged
# 确保所有必要字段都存在且类型正确
if "content" not in merged_data or not isinstance(merged_data["content"], str):
merged_data["content"] = default_merged["content"]
if "brief" not in merged_data or not isinstance(merged_data["brief"], str):
merged_data["brief"] = default_merged["brief"]
@@ -605,9 +579,6 @@ class MemoryManager:
merged_data = default_merged
# 创建新的记忆项
# 合并记忆项的标签
merged_tags = memory_item1.tags.union(memory_item2.tags)
# 取两个记忆项中更强的来源
merged_source = (
memory_item1.from_source
@@ -615,8 +586,8 @@ class MemoryManager:
else memory_item2.from_source
)
# 创建新的记忆项
merged_memory = MemoryItem(data=merged_data["content"], from_source=merged_source, tags=list(merged_tags))
# 创建新的记忆项，使用空字符串作为data
merged_memory = MemoryItem(data="", from_source=merged_source, brief=merged_data["brief"])
# 设置合并后的摘要
summary = {

5
src/chat/focus_chat/working_memory/working_memory.py
View File

@@ -51,19 +51,18 @@ class WorkingMemory:
except Exception as e:
print(f"自动衰减记忆时出错: {str(e)}")
async def add_memory(self, content: Any, from_source: str = "", tags: Optional[List[str]] = None):
async def add_memory(self, content: Any, from_source: str = ""):
"""
添加一段记忆到指定聊天
Args:
content: 记忆内容
from_source: 数据来源
tags: 数据标签列表
Returns:
包含记忆信息的字典
"""
memory = await self.memory_manager.push_with_summary(content, from_source, tags)
memory = await self.memory_manager.push_with_summary(content, from_source)
if len(self.memory_manager.get_all_items()) > self.max_memories_per_chat:
self.remove_earliest_memory()