mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-09-14 23:21:04 +00:00
refactor(personal): standardize memory operations and improve time handling
- Rename and restructure memory loading operations - Enhance time extraction and formatting capabilities- Implement more robust memory filtering and ranking logic - Refactor observation extraction methods for better reusability - Improve logging and error handling in memory operations
This commit is contained in:
jinli.yl
parent
8600015632
commit
527e04140c
25 changed files with 905 additions and 1001 deletions
|
|
@ -22,7 +22,7 @@ flow:
|
|||
input_schema:
|
||||
query:
|
||||
type: "str"
|
||||
description: "current query"
|
||||
description: "user query"
|
||||
required: true
|
||||
|
||||
summary_task_memory:
|
||||
|
|
@ -34,16 +34,6 @@ flow:
|
|||
description: "A list of conversation trajectory information, including message content and score. This field does not need to be filled in, the system will complete it automatically."
|
||||
required: false
|
||||
|
||||
vector_store:
|
||||
flow_content: vector_store_action_op
|
||||
description: "directly operate the vector store."
|
||||
input_schema:
|
||||
action:
|
||||
type: "str"
|
||||
description: "vector store operations"
|
||||
required: true
|
||||
enum: [ copy, delete, delete_ids, dump, load ]
|
||||
|
||||
retrieve_task_memory_simple:
|
||||
flow_content: build_query_op >> recall_vector_store_op >> merge_memory_op
|
||||
description: "Retrieve the most relevant top_k memory experience from historical memory based on the query to help solve tasks better now"
|
||||
|
|
@ -62,6 +52,39 @@ flow:
|
|||
description: "A list of conversation trajectory information, including message content and score. This field does not need to be filled in, the system will complete it automatically."
|
||||
required: false
|
||||
|
||||
vector_store:
|
||||
flow_content: vector_store_action_op
|
||||
description: "directly operate the vector store."
|
||||
input_schema:
|
||||
action:
|
||||
type: "str"
|
||||
description: "vector store operations"
|
||||
required: true
|
||||
enum: [ copy, delete, delete_ids, dump, load ]
|
||||
|
||||
retrieve_personal_memory:
|
||||
flow_content: set_query_op >> (extract_time_op | (retrieve_memory_op >> semantic_rank_op)) >> fuse_rerank_op
|
||||
description: "Retrieve the most relevant memories from historical memory based on the query to help answer better now."
|
||||
input_schema:
|
||||
query:
|
||||
type: "str"
|
||||
description: "user query"
|
||||
required: true
|
||||
|
||||
consolidate_personal_memory:
|
||||
flow_content: info_filter_op >> (get_observation_op | get_observation_with_time_op | load_today_memory_op) >> contra_repeat_op >> update_vector_store_op
|
||||
description: "summary user's observation memory"
|
||||
input_schema:
|
||||
messages:
|
||||
type: "list"
|
||||
description: "A list of conversation messages information. This field does not need to be filled in, the system will complete it automatically."
|
||||
required: false
|
||||
|
||||
# reconsolidate_personal_memory:
|
||||
# flow_content: load_not_reflected_memory_op >> get_reflection_subject_op >> update_insight_op >> long_contra_repeat_op >> update_vector_store_op
|
||||
# description: "Consolidate personal memories by generating topic insights, updating values, resolving conflicts, and updating vector store"
|
||||
|
||||
|
||||
llm:
|
||||
default:
|
||||
backend: openai_compatible
|
||||
|
|
|
|||
|
|
@ -1,103 +0,0 @@
|
|||
# demo config.yaml
|
||||
|
||||
http_service:
|
||||
host: "0.0.0.0"
|
||||
port: 8001
|
||||
timeout_keep_alive: 600
|
||||
limit_concurrency: 64
|
||||
|
||||
thread_pool:
|
||||
max_workers: 64
|
||||
|
||||
api:
|
||||
retriever: build_query_op->recall_vector_store_op->rerank_experience_op->rewrite_experience_op
|
||||
summarizer: trajectory_preprocess_op->[success_extraction_op|failure_extraction_op|comparative_extraction_op]->experience_validation_op->experience_deduplication_op->update_vector_store_op
|
||||
vector_store: vector_store_action_op
|
||||
|
||||
op:
|
||||
# retriever ops
|
||||
build_query_op:
|
||||
backend: build_query_op
|
||||
vector_store: default
|
||||
|
||||
recall_vector_store_op:
|
||||
backend: recall_vector_store_op
|
||||
vector_store: default
|
||||
|
||||
rerank_experience_op:
|
||||
backend: rerank_experience_op
|
||||
llm: default
|
||||
params:
|
||||
enable_llm_rerank: true
|
||||
enable_score_filter: false
|
||||
top_k: 5
|
||||
|
||||
rewrite_experience_op:
|
||||
backend: rewrite_experience_op
|
||||
llm: default
|
||||
params:
|
||||
enable_llm_rewrite: true
|
||||
|
||||
#summarizer ops
|
||||
trajectory_preprocess_op:
|
||||
backend: trajectory_preprocess_op
|
||||
params:
|
||||
success_threshold: 1.0
|
||||
|
||||
success_extraction_op:
|
||||
backend: success_extraction_op
|
||||
llm: default
|
||||
|
||||
failure_extraction_op:
|
||||
backend: failure_extraction_op
|
||||
llm: default
|
||||
|
||||
comparative_extraction_op:
|
||||
backend: comparative_extraction_op
|
||||
llm: default
|
||||
params:
|
||||
enable_soft_comparison: true
|
||||
|
||||
experience_validation_op:
|
||||
backend: experience_validation_op
|
||||
llm: default
|
||||
params:
|
||||
validation_threshold: 0.5
|
||||
|
||||
experience_deduplication_op:
|
||||
backend: experience_deduplication_op
|
||||
vector_store: default
|
||||
params:
|
||||
similarity_threshold: 0.5
|
||||
|
||||
experience_storage_op:
|
||||
backend: experience_storage_op
|
||||
vector_store: default
|
||||
|
||||
vector_store_action_op:
|
||||
backend: vector_store_action_op
|
||||
vector_store: default
|
||||
|
||||
update_vector_store_op:
|
||||
backend: update_vector_store_op
|
||||
vector_store: default
|
||||
|
||||
|
||||
llm:
|
||||
default:
|
||||
backend: openai_compatible
|
||||
model_name: qwen3-32b
|
||||
params:
|
||||
temperature: 0.6
|
||||
|
||||
embedding_model:
|
||||
default:
|
||||
backend: openai_compatible
|
||||
model_name: text-embedding-v4
|
||||
params:
|
||||
dimensions: 1024
|
||||
|
||||
vector_store:
|
||||
default:
|
||||
backend: local_file
|
||||
embedding_model: default
|
||||
|
|
@ -1,179 +0,0 @@
|
|||
global:
|
||||
language: en
|
||||
thread_pool_max_workers: 5
|
||||
enable_ranker: false
|
||||
enable_today_contra_repeat: true
|
||||
enable_long_contra_repeat: false
|
||||
output_memory_max_count: 20
|
||||
|
||||
memory_chat:
|
||||
cli_memory_chat:
|
||||
class: core.chat.cli_memory_chat
|
||||
memory_service: memoryscope_service
|
||||
generation_model: generation_model
|
||||
stream: true
|
||||
|
||||
memory_service:
|
||||
memoryscope_service:
|
||||
class: core.service.memory_scope_service
|
||||
human_name: user
|
||||
assistant_name: AI
|
||||
memory_operations:
|
||||
read_message:
|
||||
class: core.operation.frontend_operation
|
||||
workflow: read_message
|
||||
description: "read short memory"
|
||||
|
||||
retrieve_memory:
|
||||
class: core.operation.frontend_operation
|
||||
workflow: set_query,[extract_time|retrieve_obs_ins,semantic_rank],fuse_rerank
|
||||
description: "retrieve long-term memory"
|
||||
|
||||
list_memory:
|
||||
class: core.operation.frontend_operation
|
||||
workflow: set_query,retrieve_top_memory,print_memory
|
||||
description: "read all long-term memory of the user, use `refresh_time=5` to refresh screen every 5 seconds."
|
||||
|
||||
delete_memory:
|
||||
class: core.operation.frontend_operation
|
||||
workflow: set_query,retrieve_all_memory,delete_memory
|
||||
description: "delete a single long-term memory"
|
||||
|
||||
delete_all:
|
||||
class: core.operation.frontend_operation
|
||||
workflow: set_query,retrieve_all_memory,delete_all
|
||||
description: "delete all long-term memory"
|
||||
|
||||
add_memory:
|
||||
class: core.operation.frontend_operation
|
||||
workflow: add_memory
|
||||
description: "add a single observation"
|
||||
|
||||
consolidate_memory:
|
||||
class: core.operation.consolidate_memory_op
|
||||
workflow: info_filter,[get_observation|get_observation_with_time|load_today_memory],contra_repeat,store_memory
|
||||
description: "summary user's observation memory, run backend."
|
||||
interval_time: 1
|
||||
|
||||
reflect_and_reconsolidate:
|
||||
class: core.operation.backend_operation
|
||||
workflow: load_obs_and_insight,get_reflection_subject,update_insight,long_contra_repeat,store_memory
|
||||
description: "summary user's insight memory, run backend."
|
||||
interval_time: 15
|
||||
|
||||
worker:
|
||||
dummy:
|
||||
class: core.worker.dummy_worker
|
||||
generation_model: generation_model
|
||||
embedding_model: embedding_model
|
||||
rank_model: rank_model
|
||||
read_message:
|
||||
class: core.worker.frontend.read_message_worker
|
||||
set_query:
|
||||
class: core.worker.frontend.set_query_worker
|
||||
retrieve_obs_ins:
|
||||
class: core.worker.frontend.retrieve_memory_worker
|
||||
retrieve_obs_top_k: 100
|
||||
retrieve_ins_top_k: 100
|
||||
extract_time:
|
||||
class: core.worker.frontend.extract_time_worker
|
||||
generation_model: generation_model
|
||||
semantic_rank:
|
||||
class: core.worker.frontend.semantic_rank_worker
|
||||
rank_model: rank_model
|
||||
fuse_rerank:
|
||||
class: core.worker.frontend.fuse_rerank_worker
|
||||
fuse_score_threshold: 0.01
|
||||
fuse_ratio_dict:
|
||||
conversation: 0.5
|
||||
observation: 1
|
||||
obs_customized: 1.2
|
||||
insight: 2.0
|
||||
fuse_time_ratio: 2.0
|
||||
retrieve_top_memory:
|
||||
class: core.worker.frontend.retrieve_memory_worker
|
||||
retrieve_obs_top_k: 100
|
||||
retrieve_ins_top_k: 100
|
||||
retrieve_expired_top_k: 100
|
||||
print_memory:
|
||||
class: core.worker.frontend.print_memory_worker
|
||||
retrieve_all_memory:
|
||||
class: core.worker.frontend.retrieve_memory_worker
|
||||
retrieve_obs_top_k: 1000
|
||||
retrieve_ins_top_k: 1000
|
||||
retrieve_expired_top_k: 1000
|
||||
delete_memory:
|
||||
class: core.worker.backend.update_memory_worker
|
||||
method: delete_memory
|
||||
delete_all:
|
||||
class: core.worker.backend.update_memory_worker
|
||||
method: delete_all
|
||||
add_memory:
|
||||
class: core.worker.backend.update_memory_worker
|
||||
method: from_query
|
||||
info_filter:
|
||||
class: core.worker.backend.info_filter_worker
|
||||
generation_model: generation_model
|
||||
load_today_memory:
|
||||
class: core.worker.backend.load_memory_worker
|
||||
retrieve_today_top_k: 100
|
||||
get_observation:
|
||||
class: core.worker.backend.get_observation_worker
|
||||
generation_model: generation_model
|
||||
get_observation_with_time:
|
||||
class: core.worker.backend.get_observation_with_time_worker
|
||||
generation_model: generation_model
|
||||
contra_repeat:
|
||||
class: core.worker.backend.contra_repeat_worker
|
||||
generation_model: generation_model
|
||||
store_memory:
|
||||
class: core.worker.backend.update_memory_worker
|
||||
method: from_memory_key
|
||||
memory_key: all
|
||||
load_obs_and_insight:
|
||||
class: core.worker.backend.load_memory_worker
|
||||
retrieve_not_reflected_top_k: 100
|
||||
retrieve_not_updated_top_k: 100
|
||||
retrieve_insight_top_k: 100
|
||||
get_reflection_subject:
|
||||
class: core.worker.backend.get_reflection_subject_worker
|
||||
generation_model: generation_model
|
||||
reflect_obs_cnt_threshold: 6
|
||||
update_insight:
|
||||
class: core.worker.backend.update_insight_worker
|
||||
generation_model: generation_model
|
||||
rank_model: rank_model
|
||||
embedding_model: embedding_model
|
||||
update_insight_threshold: 0.01
|
||||
enable_parallel: false
|
||||
long_contra_repeat:
|
||||
class: core.worker.backend.long_contra_repeat_worker
|
||||
generation_model: generation_model
|
||||
long_contra_repeat_threshold: 0.5
|
||||
|
||||
model:
|
||||
generation_model:
|
||||
class: core.models.llama_index_generation_model
|
||||
module_name: openai_generation
|
||||
model_name: gpt-4o
|
||||
max_tokens: 2000
|
||||
temperature: 0.01
|
||||
embedding_model:
|
||||
class: core.models.llama_index_embedding_model
|
||||
module_name: openai_embedding
|
||||
model_name: text-embedding-3-small
|
||||
rank_model:
|
||||
class: core.models.llama_index_rank_model
|
||||
module_name: dashscope_rank
|
||||
model_name: gte-rerank
|
||||
top_n: 500
|
||||
|
||||
memory_store:
|
||||
class: core.storage.llama_index_es_memory_store
|
||||
embedding_model: embedding_model
|
||||
index_name: memory_index
|
||||
es_url: http://localhost:9200
|
||||
retrieve_mode: dense
|
||||
|
||||
monitor:
|
||||
class: core.storage.dummy_monitor
|
||||
|
|
@ -40,7 +40,9 @@ class ExtractTimeOp(BaseLLMOp):
|
|||
# Identify if the query contains datetime keywords
|
||||
contain_datetime = DatetimeHandler.has_time_word(query, self.language)
|
||||
if not contain_datetime:
|
||||
logger.info(f"contain_datetime={contain_datetime}")
|
||||
logger.info(f"Query contains no datetime keywords: {contain_datetime}")
|
||||
# Set empty time dict for downstream operations
|
||||
self.context[EXTRACT_TIME_DICT] = {}
|
||||
return
|
||||
|
||||
# Prepare the prompt with necessary contextual details
|
||||
|
|
@ -50,27 +52,46 @@ class ExtractTimeOp(BaseLLMOp):
|
|||
# Create message with system and few-shot examples
|
||||
system_prompt = self.prompt_format(prompt_name="extract_time_system")
|
||||
few_shot = self.prompt_format(prompt_name="extract_time_few_shot")
|
||||
user_prompt = self.prompt_format(prompt_name="extract_time_user_query", query=query,
|
||||
query_time_str=query_time_str)
|
||||
user_prompt = self.prompt_format(prompt_name="extract_time_user_query",
|
||||
query=query, query_time_str=query_time_str)
|
||||
|
||||
full_prompt = f"{system_prompt}\n\n{few_shot}\n\n{user_prompt}"
|
||||
logger.info(f"extract_time_prompt={full_prompt}")
|
||||
logger.info(f"Extracting time from query: {query[:100]}...")
|
||||
|
||||
# Invoke the LLM to generate a response
|
||||
response = self.llm.chat([Message(role=Role.USER, content=full_prompt)])
|
||||
|
||||
# Handle empty or unsuccessful responses
|
||||
if not response or not response.content:
|
||||
logger.warning("LLM returned empty response for time extraction")
|
||||
self.context[EXTRACT_TIME_DICT] = {}
|
||||
return
|
||||
|
||||
response_text = response.content
|
||||
|
||||
# Extract time information from the LLM's response using regex
|
||||
# Extract and parse time information from the LLM's response
|
||||
extract_time_dict = self._parse_time_from_response(response_text)
|
||||
|
||||
logger.info(f"Extracted time information: {extract_time_dict}")
|
||||
self.context[EXTRACT_TIME_DICT] = extract_time_dict
|
||||
|
||||
def _parse_time_from_response(self, response_text: str) -> Dict[str, str]:
|
||||
"""
|
||||
Parse time information from LLM response using regex.
|
||||
|
||||
Args:
|
||||
response_text: Raw LLM response content
|
||||
|
||||
Returns:
|
||||
Dictionary of extracted time information
|
||||
"""
|
||||
extract_time_dict: Dict[str, str] = {}
|
||||
matches = re.findall(self.EXTRACT_TIME_PATTERN, response_text)
|
||||
key_map: dict = DATATIME_KEY_MAP[DatetimeHandler.language_transform]
|
||||
|
||||
for key, value in matches:
|
||||
if key in key_map.keys():
|
||||
extract_time_dict[key_map[key]] = value
|
||||
|
||||
logger.info(f"response_text={response_text} matches={matches} filters={extract_time_dict}")
|
||||
self.context[EXTRACT_TIME_DICT] = extract_time_dict
|
||||
logger.debug(f"Time extraction - Response: {response_text[:200]}... Matches: {matches}")
|
||||
return extract_time_dict
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ from loguru import logger
|
|||
|
||||
from reme_ai.constants.common_constants import EXTRACT_TIME_DICT
|
||||
from reme_ai.schema.memory import BaseMemory
|
||||
from reme_ai.utils.datetime_handler import DatetimeHandler
|
||||
|
||||
|
||||
@C.register_op()
|
||||
|
|
@ -55,12 +54,17 @@ class FuseRerankOp(BaseLLMOp):
|
|||
2. Reranks memories based on a combination of their original score, type,
|
||||
and temporal alignment with extracted events/messages.
|
||||
3. Selects the top-K reranked memories according to the predefined threshold.
|
||||
4. Optionally infuses inferred time information into the content of selected memories.
|
||||
5. Logs reranking details and formats the final list of memories for output.
|
||||
4. Formats the final list of memories for output.
|
||||
5. Sets both response.answer and response.metadata["memory_list"]
|
||||
"""
|
||||
# Get operation parameters
|
||||
fuse_score_threshold = self.op_params.get("fuse_score_threshold", 0.1)
|
||||
fuse_ratio_dict = self.op_params.get("fuse_ratio_dict", {})
|
||||
fuse_ratio_dict = self.op_params.get("fuse_ratio_dict", {
|
||||
"conversation": 0.5,
|
||||
"observation": 1,
|
||||
"obs_customized": 1.2,
|
||||
"insight": 2.0
|
||||
})
|
||||
fuse_time_ratio = self.op_params.get("fuse_time_ratio", 2.0)
|
||||
output_memory_max_count = self.op_params.get("output_memory_max_count", 5)
|
||||
|
||||
|
|
@ -70,14 +74,42 @@ class FuseRerankOp(BaseLLMOp):
|
|||
|
||||
# Check if memories are available; warn and return if not
|
||||
if not memory_list:
|
||||
logger.warning("Memory list is empty.")
|
||||
logger.warning("No memories available for fuse reranking")
|
||||
self.context.response.answer = ""
|
||||
self.context.response.metadata["memory_list"] = []
|
||||
return
|
||||
|
||||
logger.info(f"Fuse reranking {len(memory_list)} memories")
|
||||
logger.info(f"Fuse reranking {len(memory_list)} memories with time dict: {bool(extract_time_dict)}")
|
||||
|
||||
# Perform reranking based on score, type, and time relevance
|
||||
reranked_memories = self._apply_fuse_reranking(
|
||||
memory_list, extract_time_dict, fuse_score_threshold,
|
||||
fuse_ratio_dict, fuse_time_ratio
|
||||
)
|
||||
|
||||
# Sort and select top-k memories
|
||||
reranked_memories = sorted(reranked_memories,
|
||||
key=lambda x: x.score or 0.0,
|
||||
reverse=True)[:output_memory_max_count]
|
||||
|
||||
logger.info(f"Final reranked memories: {len(reranked_memories)}")
|
||||
|
||||
# Format memories for output
|
||||
formatted_memories = self._format_memories_for_output(reranked_memories)
|
||||
|
||||
# Store results in context - both answer and metadata as required
|
||||
self.context.response.metadata["memory_list"] = reranked_memories
|
||||
self.context.response.answer = "\n".join(formatted_memories)
|
||||
|
||||
def _apply_fuse_reranking(self,
|
||||
memory_list: List[BaseMemory],
|
||||
extract_time_dict: Dict[str, str],
|
||||
fuse_score_threshold: float,
|
||||
fuse_ratio_dict: Dict[str, float],
|
||||
fuse_time_ratio: float) -> List[BaseMemory]:
|
||||
"""Apply fuse reranking logic to memories"""
|
||||
reranked_memories = []
|
||||
|
||||
for memory in memory_list:
|
||||
# Skip memories below the fuse score threshold
|
||||
memory_score = memory.score or 0.0
|
||||
|
|
@ -87,42 +119,62 @@ class FuseRerankOp(BaseLLMOp):
|
|||
# Calculate type-based adjustment factor
|
||||
memory_type = memory.metadata.get("memory_type", "default")
|
||||
if memory_type not in fuse_ratio_dict:
|
||||
logger.warning(f"{memory_type} factor is not configured!")
|
||||
logger.debug(f"Memory type '{memory_type}' not in fuse_ratio_dict, using default 0.1")
|
||||
type_ratio: float = fuse_ratio_dict.get(memory_type, 0.1)
|
||||
|
||||
# Determine time relevance adjustment factor
|
||||
match_event_flag, match_msg_flag = self.match_memory_time(
|
||||
extract_time_dict=extract_time_dict, memory=memory)
|
||||
match_event_flag, match_msg_flag = self.match_memory_time(extract_time_dict, memory)
|
||||
time_ratio: float = fuse_time_ratio if match_event_flag or match_msg_flag else 1.0
|
||||
|
||||
# Apply reranking score adjustments
|
||||
original_score = memory_score
|
||||
memory.score = memory_score * type_ratio * time_ratio
|
||||
|
||||
logger.debug(f"Memory reranked: {original_score:.3f} -> {memory.score:.3f} "
|
||||
f"(type={type_ratio}, time={time_ratio})")
|
||||
|
||||
reranked_memories.append(memory)
|
||||
|
||||
# Sort and select top-k memories
|
||||
reranked_memories = sorted(reranked_memories,
|
||||
key=lambda x: x.score or 0.0,
|
||||
reverse=True)[:output_memory_max_count]
|
||||
return reranked_memories
|
||||
|
||||
# Build result
|
||||
def _format_memories_for_output(self, memories: List[BaseMemory]) -> List[str]:
|
||||
"""Format memories for final output"""
|
||||
formatted_memories = []
|
||||
for memory in reranked_memories:
|
||||
# Log reranking details including flags for event and message matches
|
||||
logger.info(f"Rerank Stage: Content={memory.content}, Score={memory.score}, "
|
||||
f"Event Flag={memory.metadata.get('match_event_flag', '0')}, "
|
||||
f"Message Flag={memory.metadata.get('match_msg_flag', '0')}")
|
||||
|
||||
for memory in memories:
|
||||
# Log reranking details
|
||||
logger.info(f"Final memory: Score={memory.score:.3f}, "
|
||||
f"Event={memory.metadata.get('match_event_flag', '0')}, "
|
||||
f"Msg={memory.metadata.get('match_msg_flag', '0')}, "
|
||||
f"Content={memory.content[:50]}...")
|
||||
|
||||
# Format memory with timestamp if available
|
||||
if hasattr(memory, 'timestamp') and memory.timestamp:
|
||||
dt_handler = DatetimeHandler(memory.timestamp)
|
||||
datetime_str = dt_handler.datetime_format("%Y-%m-%d %H:%M:%S")
|
||||
weekday = dt_handler.get_dt_info_dict(self.language)["weekday"]
|
||||
formatted_content = f"[{datetime_str} {weekday}] {memory.content}"
|
||||
else:
|
||||
formatted_content = memory.content
|
||||
|
||||
formatted_content = self._format_memory_with_timestamp(memory, self.language)
|
||||
formatted_memories.append(formatted_content)
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["memory_list"] = reranked_memories
|
||||
self.context.response.answer = "\n".join(formatted_memories)
|
||||
return formatted_memories
|
||||
|
||||
@staticmethod
|
||||
def _format_memory_with_timestamp(memory, language: str = "en") -> str:
|
||||
"""
|
||||
Format memory content with timestamp if available.
|
||||
|
||||
Args:
|
||||
memory: Memory object
|
||||
language: Language for formatting
|
||||
|
||||
Returns:
|
||||
Formatted memory content string
|
||||
"""
|
||||
try:
|
||||
if hasattr(memory, 'timestamp') and memory.timestamp:
|
||||
from reme_ai.utils.datetime_handler import DatetimeHandler
|
||||
dt_handler = DatetimeHandler(memory.timestamp)
|
||||
datetime_str = dt_handler.datetime_format("%Y-%m-%d %H:%M:%S")
|
||||
weekday = dt_handler.get_dt_info_dict(language)["weekday"]
|
||||
return f"[{datetime_str} {weekday}] {memory.content}"
|
||||
else:
|
||||
return memory.content
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to format memory with timestamp: {e}")
|
||||
return memory.content
|
||||
|
|
|
|||
|
|
@ -62,3 +62,70 @@ class PrintMemoryOp(BaseOp):
|
|||
formatted_memories.append(memory_text)
|
||||
|
||||
return "\n".join(formatted_memories)
|
||||
|
||||
@staticmethod
|
||||
def format_memories_for_output(memories: List) -> str:
|
||||
"""
|
||||
Format memory list for output string.
|
||||
|
||||
Args:
|
||||
memories: List of memory objects
|
||||
|
||||
Returns:
|
||||
Formatted string
|
||||
"""
|
||||
if not memories:
|
||||
return ""
|
||||
|
||||
formatted_parts = []
|
||||
for i, memory in enumerate(memories, 1):
|
||||
when_to_use = getattr(memory, 'when_to_use', '') or memory.get('when_to_use', '')
|
||||
content = getattr(memory, 'content', '') or memory.get('content', '')
|
||||
|
||||
part = f"Memory {i}:\n"
|
||||
if when_to_use:
|
||||
part += f"When to use: {when_to_use}\n"
|
||||
if content:
|
||||
part += f"Content: {content}\n"
|
||||
|
||||
formatted_parts.append(part)
|
||||
|
||||
return "\n".join(formatted_parts)
|
||||
|
||||
@staticmethod
|
||||
def format_memories_for_simple_output(memories: List) -> str:
|
||||
"""
|
||||
Format memory list for simple flow output.
|
||||
|
||||
Args:
|
||||
memories: List of memory objects
|
||||
|
||||
Returns:
|
||||
Formatted string suitable for response.answer
|
||||
"""
|
||||
if not memories:
|
||||
return "No relevant memories found."
|
||||
|
||||
content_parts = ["Previous Memory"]
|
||||
|
||||
for memory in memories:
|
||||
# Safely get field values
|
||||
when_to_use = getattr(memory, 'when_to_use', '') or memory.get('when_to_use', '')
|
||||
content = getattr(memory, 'content', '') or memory.get('content', '')
|
||||
|
||||
# Skip memories with empty content
|
||||
if not content:
|
||||
continue
|
||||
|
||||
# Format individual memory
|
||||
memory_text = f"- when_to_use: {when_to_use}\n content: {content}"
|
||||
content_parts.append(memory_text)
|
||||
|
||||
# If no valid memories, return empty message
|
||||
if len(content_parts) == 1: # Only title
|
||||
return "No relevant memories with valid content found."
|
||||
|
||||
content_parts.append("\nPlease consider the helpful parts from these in answering the question, "
|
||||
"to make the response more comprehensive and substantial.")
|
||||
|
||||
return "\n".join(content_parts)
|
||||
|
|
|
|||
|
|
@ -1,37 +1,14 @@
|
|||
import json
|
||||
import re
|
||||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
from loguru import logger
|
||||
|
||||
from reme_ai.schema import Message, Role
|
||||
from reme_ai.schema.memory import BaseMemory
|
||||
|
||||
|
||||
def _parse_ranking_response(response: str) -> List[dict]:
|
||||
"""Parse LLM ranking response"""
|
||||
import json
|
||||
import re
|
||||
|
||||
try:
|
||||
# Try to extract JSON blocks
|
||||
json_pattern = r'```json\s*([\s\S]*?)\s*```'
|
||||
json_blocks = re.findall(json_pattern, response)
|
||||
|
||||
if json_blocks:
|
||||
parsed = json.loads(json_blocks[0])
|
||||
if isinstance(parsed, dict) and "rankings" in parsed:
|
||||
return parsed["rankings"]
|
||||
|
||||
# Fallback: try to parse the entire response as JSON
|
||||
parsed = json.loads(response)
|
||||
if isinstance(parsed, dict) and "rankings" in parsed:
|
||||
return parsed["rankings"]
|
||||
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("Failed to parse ranking response as JSON")
|
||||
|
||||
return []
|
||||
|
||||
|
||||
@C.register_op()
|
||||
class SemanticRankOp(BaseLLMOp):
|
||||
"""
|
||||
|
|
@ -55,8 +32,8 @@ class SemanticRankOp(BaseLLMOp):
|
|||
If no memories are retrieved or if the ranking fails,
|
||||
appropriate warnings are logged.
|
||||
"""
|
||||
# Get memory list from context
|
||||
memory_list: List[BaseMemory] = self.context.response.metadata.get("memory_list", [])
|
||||
# Get memory list from context - previous op guarantees this exists
|
||||
memory_list: List[BaseMemory] = self.context.response.metadata["memory_list"]
|
||||
query: str = self.context.query
|
||||
|
||||
# Get parameters from op_params
|
||||
|
|
@ -67,27 +44,30 @@ class SemanticRankOp(BaseLLMOp):
|
|||
logger.warning("Memory list is empty!")
|
||||
return
|
||||
|
||||
logger.info(f"Semantic ranking {len(memory_list)} memories for query: {query[:100]}...")
|
||||
|
||||
if not enable_ranker or len(memory_list) <= output_memory_max_count:
|
||||
# Use original scores if ranker is disabled or memory count is small
|
||||
logger.warning("Using original scores instead of semantic ranking!")
|
||||
logger.info("Skipping semantic ranking - using original scores")
|
||||
else:
|
||||
# Remove duplicates based on content
|
||||
memory_dict = {memory.content.strip(): memory for memory in memory_list if memory.content.strip()}
|
||||
memory_list = list(memory_dict.values())
|
||||
logger.info(f"After deduplication: {len(memory_list)} memories")
|
||||
|
||||
# Perform semantic ranking using LLM
|
||||
ranked_memories = self._semantic_rank_memories(query, memory_list)
|
||||
if ranked_memories:
|
||||
memory_list = ranked_memories
|
||||
|
||||
# Sort by score (assuming score is available in BaseMemory)
|
||||
# Sort by score
|
||||
memory_list = sorted(memory_list, key=lambda m: getattr(m, 'score', 0.0), reverse=True)
|
||||
|
||||
# Log ranked memories
|
||||
logger.info(f"Semantic rank stage: query={query}")
|
||||
for i, memory in enumerate(memory_list):
|
||||
# Log top ranked memories
|
||||
logger.info(f"Semantic ranking completed for query: {query[:50]}...")
|
||||
for i, memory in enumerate(memory_list[:5]): # Log top 5
|
||||
score = getattr(memory, 'score', 0.0)
|
||||
logger.info(f"Rank stage: Memory {i + 1}: Content={memory.content[:100]}..., Score={score}")
|
||||
logger.info(f"Top {i + 1}: Score={score:.3f}, Content={memory.content[:80]}...")
|
||||
|
||||
# Save ranked memories back to context
|
||||
self.context.response.metadata["memory_list"] = memory_list
|
||||
|
|
@ -99,12 +79,11 @@ class SemanticRankOp(BaseLLMOp):
|
|||
if not memories:
|
||||
return memories
|
||||
|
||||
try:
|
||||
# Format memories for ranking
|
||||
formatted_memories = self._format_memories_for_ranking(memories)
|
||||
# Format memories for ranking
|
||||
formatted_memories = SemanticRankOp.format_memories_for_llm_ranking(memories)
|
||||
|
||||
# Create prompt for semantic ranking
|
||||
prompt = f"""Given the query: "{query}"
|
||||
# Create prompt for semantic ranking
|
||||
prompt = f"""Given the query: "{query}"
|
||||
|
||||
Please rank the following memories by their semantic relevance to the query.
|
||||
Rate each memory on a scale of 0.0 to 1.0 where 1.0 is most relevant.
|
||||
|
|
@ -115,46 +94,72 @@ Memories:
|
|||
Please respond in JSON format:
|
||||
{{"rankings": [{{"index": 0, "score": 0.8}}, {{"index": 1, "score": 0.6}}, ...]}}"""
|
||||
|
||||
# Get LLM response
|
||||
from flowllm.schema.message import Message
|
||||
from flowllm.enumeration.role import Role
|
||||
response = self.llm.chat([Message(role=Role.USER, content=prompt)])
|
||||
|
||||
response = self.llm.chat([Message(role=Role.USER, content=prompt)])
|
||||
if not response or not response.content:
|
||||
logger.warning("LLM ranking failed, using original order")
|
||||
return memories
|
||||
|
||||
if not response or not response.content:
|
||||
logger.warning("LLM ranking failed, using original order")
|
||||
return memories
|
||||
# Parse and apply ranking results
|
||||
rankings = SemanticRankOp.parse_llm_ranking_response(response.content)
|
||||
|
||||
# Parse ranking results
|
||||
rankings = _parse_ranking_response(response.content)
|
||||
|
||||
if rankings:
|
||||
# Apply scores to memories
|
||||
for ranking in rankings:
|
||||
idx = ranking.get("index", -1)
|
||||
score = ranking.get("score", 0.0)
|
||||
if 0 <= idx < len(memories):
|
||||
# Set score on memory object
|
||||
if hasattr(memories[idx], 'score'):
|
||||
memories[idx].score = score
|
||||
else:
|
||||
# Add score as metadata if score attribute doesn't exist
|
||||
if not hasattr(memories[idx], 'metadata'):
|
||||
memories[idx].metadata = {}
|
||||
memories[idx].metadata['semantic_score'] = score
|
||||
|
||||
logger.info(f"Successfully applied semantic rankings to {len(rankings)} memories")
|
||||
else:
|
||||
logger.warning("Failed to parse ranking results")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in semantic ranking: {e}")
|
||||
if rankings:
|
||||
applied_count = SemanticRankOp.apply_semantic_scores_to_memories(memories, rankings)
|
||||
logger.info(f"Successfully applied semantic rankings to {applied_count} memories")
|
||||
else:
|
||||
logger.warning("Failed to parse ranking results")
|
||||
|
||||
return memories
|
||||
|
||||
@staticmethod
|
||||
def _format_memories_for_ranking(memories: List[BaseMemory]) -> str:
|
||||
"""Format memories for LLM ranking"""
|
||||
def parse_llm_ranking_response(response: str) -> List[dict]:
|
||||
"""Parse LLM ranking response to extract rankings."""
|
||||
try:
|
||||
# Try to extract JSON blocks
|
||||
json_pattern = r'```json\s*([\s\S]*?)\s*```'
|
||||
json_blocks = re.findall(json_pattern, response)
|
||||
|
||||
if json_blocks:
|
||||
parsed = json.loads(json_blocks[0])
|
||||
if isinstance(parsed, dict) and "rankings" in parsed:
|
||||
return parsed["rankings"]
|
||||
|
||||
# Fallback: try to parse the entire response as JSON
|
||||
parsed = json.loads(response)
|
||||
if isinstance(parsed, dict) and "rankings" in parsed:
|
||||
return parsed["rankings"]
|
||||
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("Failed to parse ranking response as JSON")
|
||||
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def apply_semantic_scores_to_memories(memories: List, rankings: List[dict]) -> int:
|
||||
"""Apply semantic ranking scores to memory objects."""
|
||||
applied_count = 0
|
||||
|
||||
for ranking in rankings:
|
||||
idx = ranking.get("index", -1)
|
||||
score = ranking.get("score", 0.0)
|
||||
|
||||
if 0 <= idx < len(memories):
|
||||
# Set score on memory object
|
||||
if hasattr(memories[idx], 'score'):
|
||||
memories[idx].score = score
|
||||
applied_count += 1
|
||||
else:
|
||||
# Add score as metadata if score attribute doesn't exist
|
||||
if not hasattr(memories[idx], 'metadata'):
|
||||
memories[idx].metadata = {}
|
||||
memories[idx].metadata['semantic_score'] = score
|
||||
applied_count += 1
|
||||
|
||||
return applied_count
|
||||
|
||||
@staticmethod
|
||||
def format_memories_for_llm_ranking(memories: List) -> str:
|
||||
"""Format memories for LLM ranking input."""
|
||||
formatted_memories = []
|
||||
|
||||
for i, memory in enumerate(memories):
|
||||
|
|
|
|||
|
|
@ -13,52 +13,25 @@ class SetQueryOp(BaseOp):
|
|||
The `SetQueryOp` class is responsible for setting a query and its associated timestamp
|
||||
into the context, utilizing either provided parameters or details from the context.
|
||||
"""
|
||||
file_path: str = __file__
|
||||
|
||||
def execute(self):
|
||||
"""
|
||||
Executes the operation's primary function, which involves determining the query and its
|
||||
timestamp, then storing these values within the context.
|
||||
|
||||
If 'query' exists in context, it is used directly. Otherwise, extracts query from
|
||||
messages or other context parameters.
|
||||
Input requirement: self.context.query must exist (flow input requirement)
|
||||
"""
|
||||
query = "" # Default query value
|
||||
timestamp = int(datetime.datetime.now().timestamp()) # Current timestamp as default
|
||||
# Flow guarantees query exists - use it directly
|
||||
query: str = self.context.query
|
||||
timestamp: int = int(datetime.datetime.now().timestamp())
|
||||
|
||||
try:
|
||||
# Check if query already exists in context
|
||||
if hasattr(self.context, 'query') and self.context.query:
|
||||
query = str(self.context.query).strip()
|
||||
logger.info(f"Using existing query from context: {query}")
|
||||
# Set timestamp if provided in op_params
|
||||
_timestamp = self.op_params.get("timestamp")
|
||||
if _timestamp and isinstance(_timestamp, int):
|
||||
timestamp = _timestamp
|
||||
|
||||
# Check for query in op_params
|
||||
elif "query" in self.op_params:
|
||||
query = self.op_params["query"]
|
||||
if not query:
|
||||
query = ""
|
||||
query = query.strip()
|
||||
logger.info(f"Using query from op_params: {query}")
|
||||
# Store the query and its timestamp in the context
|
||||
query_with_ts: Tuple[str, int] = (query, timestamp)
|
||||
self.context[QUERY_WITH_TS] = query_with_ts
|
||||
|
||||
# Check for messages in context
|
||||
elif hasattr(self.context, 'messages') and self.context.messages:
|
||||
# Use the last message content as query
|
||||
last_message = self.context.messages[-1]
|
||||
query = last_message.content.strip() if hasattr(last_message, 'content') else ""
|
||||
logger.info(f"Using query from last message: {query}")
|
||||
|
||||
# Set timestamp if provided in op_params
|
||||
_timestamp = self.op_params.get("timestamp")
|
||||
if _timestamp and isinstance(_timestamp, int):
|
||||
timestamp = _timestamp
|
||||
|
||||
# Store the determined query and its timestamp in the context
|
||||
query_with_ts: Tuple[str, int] = (query, timestamp)
|
||||
self.context[QUERY_WITH_TS] = query_with_ts
|
||||
|
||||
logger.info(f"Set query with timestamp: query='{query}', timestamp={timestamp}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in SetQueryOp execution: {e}")
|
||||
# Fallback: set empty query with current timestamp
|
||||
self.context[QUERY_WITH_TS] = ("", timestamp)
|
||||
logger.info(f"Set query with timestamp: query='{query}', timestamp={timestamp}")
|
||||
|
|
|
|||
|
|
@ -2,6 +2,8 @@ from flowllm import C, BaseLLMOp
|
|||
from flowllm.utils.llm_utils import merge_messages_content
|
||||
from loguru import logger
|
||||
|
||||
from reme_ai.schema import Message, Role
|
||||
|
||||
|
||||
@C.register_op()
|
||||
class BuildQueryOp(BaseLLMOp):
|
||||
|
|
@ -14,7 +16,9 @@ class BuildQueryOp(BaseLLMOp):
|
|||
elif "messages" in self.context:
|
||||
if self.op_params.get("enable_llm_build", True):
|
||||
execution_process = merge_messages_content(self.context.messages)
|
||||
query = self.prompt_format(prompt_name="query_build", execution_process=execution_process)
|
||||
prompt = self.prompt_format(prompt_name="query_build", execution_process=execution_process)
|
||||
message = self.llm.chat(messages=[Message(role=Role.USER, content=prompt)])
|
||||
query = message.content
|
||||
|
||||
else:
|
||||
context_parts = []
|
||||
|
|
|
|||
|
|
@ -87,7 +87,8 @@ class RewriteMemoryOp(BaseLLMOp):
|
|||
logger.error(f"Error in context rewriting: {e}")
|
||||
return context_content
|
||||
|
||||
def _format_memories_for_context(self, memories: List[BaseMemory]) -> str:
|
||||
@staticmethod
|
||||
def _format_memories_for_context(memories: List[BaseMemory]) -> str:
|
||||
"""Format memories for context generation"""
|
||||
formatted_memories = []
|
||||
|
||||
|
|
@ -100,7 +101,8 @@ class RewriteMemoryOp(BaseLLMOp):
|
|||
|
||||
return "\n".join(formatted_memories)
|
||||
|
||||
def _extract_context(self, messages: List[Message]) -> str:
|
||||
@staticmethod
|
||||
def _extract_context(messages: List[Message]) -> str:
|
||||
"""Extract relevant context from messages"""
|
||||
if not messages:
|
||||
return ""
|
||||
|
|
@ -119,7 +121,8 @@ class RewriteMemoryOp(BaseLLMOp):
|
|||
|
||||
return "\n\n".join(context_parts)
|
||||
|
||||
def _parse_json_response(self, response: str, key: str) -> str:
|
||||
@staticmethod
|
||||
def _parse_json_response(response: str, key: str) -> str:
|
||||
"""Parse JSON response to extract specific key"""
|
||||
try:
|
||||
# Try to extract JSON blocks
|
||||
|
|
|
|||
|
|
@ -100,39 +100,6 @@ class PersonalMemory(BaseMemory):
|
|||
metadata=node.metadata.get("metadata"))
|
||||
|
||||
|
||||
class PersonalTopicMemory(PersonalMemory):
|
||||
memory_type: str = Field(default="personal_topic")
|
||||
|
||||
def to_vector_node(self) -> VectorNode:
|
||||
return VectorNode(unique_id=self.memory_id,
|
||||
workspace_id=self.workspace_id,
|
||||
content=self.when_to_use,
|
||||
metadata={
|
||||
"memory_type": self.memory_type,
|
||||
"content": self.content,
|
||||
"target": self.target,
|
||||
"score": self.score,
|
||||
"created_time": self.created_time,
|
||||
"modified_time": self.modified_time,
|
||||
"author": self.author,
|
||||
"metadata": self.metadata,
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def from_vector_node(cls, node: VectorNode) -> "PersonalTopicMemory":
|
||||
return cls(workspace_id=node.workspace_id,
|
||||
memory_id=node.unique_id,
|
||||
memory_type=node.metadata.get("memory_type"),
|
||||
when_to_use=node.content,
|
||||
content=node.metadata.get("content"),
|
||||
target=node.metadata.get("target", ""),
|
||||
score=node.metadata.get("score"),
|
||||
created_time=node.metadata.get("created_time"),
|
||||
modified_time=node.metadata.get("modified_time"),
|
||||
author=node.metadata.get("author"),
|
||||
metadata=node.metadata.get("metadata"))
|
||||
|
||||
|
||||
def vector_node_to_memory(node: VectorNode) -> BaseMemory:
|
||||
memory_type = node.metadata.get("memory_type")
|
||||
if memory_type == "task":
|
||||
|
|
@ -141,9 +108,6 @@ def vector_node_to_memory(node: VectorNode) -> BaseMemory:
|
|||
elif memory_type == "personal":
|
||||
return PersonalMemory.from_vector_node(node)
|
||||
|
||||
elif memory_type == "personal_topic":
|
||||
return PersonalTopicMemory.from_vector_node(node)
|
||||
|
||||
else:
|
||||
raise RuntimeError(f"memory_type={memory_type} not supported!")
|
||||
|
||||
|
|
@ -156,9 +120,6 @@ def dict_to_experience(memory_dict: dict):
|
|||
elif memory_type == "personal":
|
||||
return PersonalMemory(**memory_dict)
|
||||
|
||||
elif memory_type == "personal_topic":
|
||||
return PersonalTopicMemory(**memory_dict)
|
||||
|
||||
else:
|
||||
raise RuntimeError(f"memory_type={memory_type} not supported!")
|
||||
|
||||
|
|
|
|||
|
|
@ -3,17 +3,6 @@ from .get_observation_op import GetObservationOp
|
|||
from .get_observation_with_time_op import GetObservationWithTimeOp
|
||||
from .get_reflection_subject_op import GetReflectionSubjectOp
|
||||
from .info_filter_op import InfoFilterOp
|
||||
from .load_memory_op import LoadMemoryOp
|
||||
from .load_today_memory_op import LoadTodayMemoryOp
|
||||
from .long_contra_repeat_op import LongContraRepeatOp
|
||||
from .update_insight_op import UpdateInsightOp
|
||||
|
||||
__all__ = [
|
||||
"ContraRepeatOp",
|
||||
"GetObservationWithTimeOp",
|
||||
"GetObservationOp",
|
||||
"GetReflectionSubjectOp",
|
||||
"InfoFilterOp",
|
||||
"LoadMemoryOp",
|
||||
"LongContraRepeatOp",
|
||||
"UpdateInsightOp"
|
||||
]
|
||||
from .update_insight_op import UpdateInsightOp
|
||||
|
|
@ -1,4 +1,6 @@
|
|||
from typing import List
|
||||
import json
|
||||
import re
|
||||
from typing import List, Tuple
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
from flowllm.enumeration.role import Role
|
||||
|
|
@ -30,11 +32,17 @@ class ContraRepeatOp(BaseLLMOp):
|
|||
3. Parses the model's response to detect contradictions or redundancies
|
||||
4. Filters and returns the processed memories
|
||||
"""
|
||||
# Get memory list from context
|
||||
memory_list: List[BaseMemory] = self.context.response.metadata.get("memory_list", [])
|
||||
# Get memory list from context - standardized key
|
||||
memory_list: List[BaseMemory] = []
|
||||
memory_list.extend(self.context.observation_memories)
|
||||
memory_list.extend(self.context.observation_memories_with_time)
|
||||
memory_list.extend(self.context.today_memories)
|
||||
|
||||
self.context.response.metadata["memory_list"] = memory_list
|
||||
|
||||
if not memory_list:
|
||||
logger.info("memory_list is empty!")
|
||||
self.context.response.metadata["deleted_memory_ids"] = []
|
||||
return
|
||||
|
||||
# Get operation parameters
|
||||
|
|
@ -43,14 +51,16 @@ class ContraRepeatOp(BaseLLMOp):
|
|||
|
||||
if not enable_contra_repeat:
|
||||
logger.warning("contra_repeat is not enabled!")
|
||||
self.context.response.metadata["deleted_memory_ids"] = []
|
||||
return
|
||||
|
||||
# Sort and limit memories by count
|
||||
sorted_memories = sorted(memory_list, key=lambda x: getattr(x, 'created_at', ''), reverse=True)[
|
||||
:contra_repeat_max_count]
|
||||
sorted_memories = sorted(memory_list, key=lambda x: x.created_time, reverse=True)[:contra_repeat_max_count]
|
||||
|
||||
if len(sorted_memories) <= 1:
|
||||
logger.info("sorted_memories.size<=1, stop.")
|
||||
self.context.response.metadata["memory_list"] = sorted_memories
|
||||
self.context.response.metadata["deleted_memory_ids"] = []
|
||||
return
|
||||
|
||||
# Build prompt
|
||||
|
|
@ -77,22 +87,26 @@ class ContraRepeatOp(BaseLLMOp):
|
|||
# Return if empty
|
||||
if not response or not response.content:
|
||||
logger.warning("Empty response from LLM")
|
||||
self.context.response.metadata["memory_list"] = sorted_memories
|
||||
self.context.response.metadata["deleted_memory_ids"] = []
|
||||
return
|
||||
|
||||
response_text = response.content
|
||||
logger.info(f"contra_repeat_response={response_text}")
|
||||
|
||||
# Parse response and filter memories
|
||||
filtered_memories = self._parse_and_filter_memories(response_text, sorted_memories, user_name)
|
||||
filtered_memories, deleted_memory_ids = self._parse_and_filter_memories(response_text, sorted_memories)
|
||||
|
||||
# Update context with filtered memories
|
||||
# Update context with filtered memories and deleted memory IDs - standardized keys
|
||||
self.context.response.metadata["memory_list"] = filtered_memories
|
||||
self.context.response.metadata["deleted_memory_ids"] = deleted_memory_ids
|
||||
logger.info(f"Filtered {len(memory_list)} memories to {len(filtered_memories)} memories")
|
||||
logger.info(f"Deleted memory IDs: {json.dumps(deleted_memory_ids, indent=2)}")
|
||||
|
||||
def _parse_and_filter_memories(self, response_text: str, memories: List[BaseMemory], user_name: str) -> List[
|
||||
BaseMemory]:
|
||||
@staticmethod
|
||||
def _parse_and_filter_memories(response_text: str, memories: List[BaseMemory]) -> Tuple[
|
||||
List[BaseMemory], List[str]]:
|
||||
"""Parse LLM response and filter memories based on contradiction/containment analysis"""
|
||||
import re
|
||||
|
||||
# Parse the response to extract judgments
|
||||
pattern = r"<(\d+)>\s*<(矛盾|被包含|无|Contradiction|Contained|None)>"
|
||||
|
|
@ -100,10 +114,11 @@ class ContraRepeatOp(BaseLLMOp):
|
|||
|
||||
if not matches:
|
||||
logger.warning("No valid judgments found in response")
|
||||
return memories
|
||||
return memories, []
|
||||
|
||||
# Create a set of indices to remove (contradictory or contained memories)
|
||||
indices_to_remove = set()
|
||||
deleted_memory_ids = []
|
||||
|
||||
for idx_str, judgment in matches:
|
||||
try:
|
||||
|
|
@ -115,6 +130,7 @@ class ContraRepeatOp(BaseLLMOp):
|
|||
judgment_lower = judgment.lower()
|
||||
if judgment_lower in ['矛盾', 'contradiction', '被包含', 'contained']:
|
||||
indices_to_remove.add(idx)
|
||||
deleted_memory_ids.append(memories[idx].id)
|
||||
logger.info(f"Marking memory {idx + 1} for removal: {judgment} - {memories[idx].content[:100]}...")
|
||||
|
||||
except ValueError:
|
||||
|
|
@ -124,8 +140,4 @@ class ContraRepeatOp(BaseLLMOp):
|
|||
# Filter out the memories marked for removal
|
||||
filtered_memories = [memory for i, memory in enumerate(memories) if i not in indices_to_remove]
|
||||
|
||||
return filtered_memories
|
||||
|
||||
def get_language_value(self, value_dict: dict):
|
||||
"""Get language-specific value from dictionary"""
|
||||
return value_dict.get(self.language, value_dict.get("en"))
|
||||
return filtered_memories, deleted_memory_ids
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import re
|
||||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
|
|
@ -6,7 +7,6 @@ from loguru import logger
|
|||
|
||||
from reme_ai.schema.memory import BaseMemory, PersonalMemory
|
||||
from reme_ai.utils.datetime_handler import DatetimeHandler
|
||||
from reme_ai.utils.op_utils import parse_observation_response
|
||||
|
||||
|
||||
@C.register_op()
|
||||
|
|
@ -18,8 +18,8 @@ class GetObservationOp(BaseLLMOp):
|
|||
|
||||
def execute(self):
|
||||
"""Extract personal observations from chat messages"""
|
||||
# Get messages from context
|
||||
messages: List[Message] = self.context.get("messages", [])
|
||||
# Get messages from context - guaranteed to exist by flow input
|
||||
messages: List[Message] = self.context.messages
|
||||
if not messages:
|
||||
logger.warning("No messages found in context")
|
||||
return
|
||||
|
|
@ -28,6 +28,7 @@ class GetObservationOp(BaseLLMOp):
|
|||
filtered_messages = self._filter_messages(messages)
|
||||
if not filtered_messages:
|
||||
logger.warning("No messages left after filtering")
|
||||
self.context.observation_memories = []
|
||||
return
|
||||
|
||||
logger.info(f"Extracting observations from {len(filtered_messages)} filtered messages")
|
||||
|
|
@ -35,8 +36,8 @@ class GetObservationOp(BaseLLMOp):
|
|||
# Extract observations using LLM
|
||||
observation_memories = self._extract_observations_from_messages(filtered_messages)
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["observation_memories"] = observation_memories
|
||||
# Store results in context using standardized key
|
||||
self.context.observation_memories = observation_memories
|
||||
logger.info(f"Generated {len(observation_memories)} observation memories")
|
||||
|
||||
def _filter_messages(self, messages: List[Message]) -> List[Message]:
|
||||
|
|
@ -83,8 +84,8 @@ class GetObservationOp(BaseLLMOp):
|
|||
response_text = message.content
|
||||
logger.info(f"get_observation_response={response_text}")
|
||||
|
||||
# Parse observations using utility function
|
||||
parsed_observations = parse_observation_response(response_text)
|
||||
# Parse observations using class method
|
||||
parsed_observations = GetObservationOp.parse_observation_response(response_text)
|
||||
|
||||
observation_memories = []
|
||||
for obs in parsed_observations:
|
||||
|
|
@ -98,7 +99,7 @@ class GetObservationOp(BaseLLMOp):
|
|||
workspace_id=self.context.get("workspace_id", ""),
|
||||
content=obs["content"],
|
||||
target=user_name,
|
||||
author=getattr(self.llm, "model_name", "system"),
|
||||
author=self.llm.model_name,
|
||||
metadata={
|
||||
"keywords": obs["keywords"],
|
||||
"source_message": filtered_messages[idx].content,
|
||||
|
|
@ -113,6 +114,33 @@ class GetObservationOp(BaseLLMOp):
|
|||
# Use LLM chat with callback function
|
||||
return self.llm.chat(messages=[Message(content=full_prompt)], callback_fn=parse_observations)
|
||||
|
||||
def get_language_value(self, value_dict: dict):
|
||||
"""Get language-specific value from dictionary"""
|
||||
return value_dict.get(self.language, value_dict.get("en"))
|
||||
@staticmethod
|
||||
def parse_observation_response(response_text: str) -> List[dict]:
|
||||
"""Parse observation response to extract structured data"""
|
||||
# Pattern to match both Chinese and English observation formats
|
||||
pattern = r"信息：<(\d+)>\s*<>\s*<([^<>]+)>\s*<([^<>]*)>|Information:\s*<(\d+)>\s*<>\s*<([^<>]+)>\s*<([^<>]*)>"
|
||||
matches = re.findall(pattern, response_text, re.IGNORECASE | re.MULTILINE)
|
||||
|
||||
observations = []
|
||||
for match in matches:
|
||||
# Handle both Chinese and English patterns
|
||||
if match[0]: # Chinese pattern
|
||||
idx_str, content, keywords = match[0], match[1], match[2]
|
||||
else: # English pattern
|
||||
idx_str, content, keywords = match[3], match[4], match[5]
|
||||
|
||||
try:
|
||||
idx = int(idx_str)
|
||||
# Skip if content indicates no meaningful observation
|
||||
content_lower = content.lower().strip()
|
||||
if content_lower not in ['无', 'none', '', 'repeat']:
|
||||
observations.append({
|
||||
"index": idx,
|
||||
"content": content.strip(),
|
||||
"keywords": keywords.strip() if keywords else ""
|
||||
})
|
||||
except ValueError:
|
||||
logger.warning(f"Invalid index format: {idx_str}")
|
||||
continue
|
||||
|
||||
return observations
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import re
|
||||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
|
|
@ -6,7 +7,6 @@ from loguru import logger
|
|||
|
||||
from reme_ai.schema.memory import BaseMemory, PersonalMemory
|
||||
from reme_ai.utils.datetime_handler import DatetimeHandler
|
||||
from reme_ai.utils.op_utils import parse_observation_with_time_response
|
||||
|
||||
|
||||
@C.register_op()
|
||||
|
|
@ -18,8 +18,8 @@ class GetObservationWithTimeOp(BaseLLMOp):
|
|||
|
||||
def execute(self):
|
||||
"""Extract personal observations with time information from chat messages"""
|
||||
# Get messages from context
|
||||
messages: List[Message] = self.context.get("messages", [])
|
||||
# Get messages from context - guaranteed to exist by flow input
|
||||
messages: List[Message] = self.context.messages
|
||||
if not messages:
|
||||
logger.warning("No messages found in context")
|
||||
return
|
||||
|
|
@ -28,16 +28,17 @@ class GetObservationWithTimeOp(BaseLLMOp):
|
|||
filtered_messages = self._filter_messages(messages)
|
||||
if not filtered_messages:
|
||||
logger.warning("No messages with time keywords found")
|
||||
self.context.observation_memories_with_time = []
|
||||
return
|
||||
|
||||
logger.info(f"Extracting observations with time from {len(filtered_messages)} filtered messages")
|
||||
|
||||
# Extract observations using LLM
|
||||
observation_memories = self._extract_observations_with_time_from_messages(filtered_messages)
|
||||
observation_memories_with_time = self._extract_observations_with_time_from_messages(filtered_messages)
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["observation_with_time_memories"] = observation_memories
|
||||
logger.info(f"Generated {len(observation_memories)} observation memories with time")
|
||||
# Store results in context using standardized key
|
||||
self.context.observation_memories_with_time = observation_memories_with_time
|
||||
logger.info(f"Generated {len(observation_memories_with_time)} observation memories with time")
|
||||
|
||||
def _filter_messages(self, messages: List[Message]) -> List[Message]:
|
||||
"""
|
||||
|
|
@ -92,8 +93,8 @@ class GetObservationWithTimeOp(BaseLLMOp):
|
|||
response_text = message.content
|
||||
logger.info(f"get_observation_with_time_response={response_text}")
|
||||
|
||||
# Parse observations using utility function
|
||||
parsed_observations = parse_observation_with_time_response(response_text)
|
||||
# Parse observations using class method
|
||||
parsed_observations = GetObservationWithTimeOp.parse_observation_with_time_response(response_text)
|
||||
|
||||
observation_memories = []
|
||||
for obs in parsed_observations:
|
||||
|
|
@ -127,3 +128,37 @@ class GetObservationWithTimeOp(BaseLLMOp):
|
|||
"""Get language-specific colon word"""
|
||||
colon_dict = {"zh": "：", "cn": "：", "en": ": "}
|
||||
return colon_dict.get(self.language, ": ")
|
||||
|
||||
@staticmethod
|
||||
def parse_observation_with_time_response(response_text: str) -> List[dict]:
|
||||
"""Parse observation with time response to extract structured data"""
|
||||
# Pattern to match both Chinese and English observation formats with time information
|
||||
# Chinese: 信息：<1> <时间信息或不输出> <明确的重要信息或"无"> <关键词>
|
||||
# English: Information: <1> <Time information or do not output> <Clear important information or "None"> <Keywords>
|
||||
pattern = r"信息：<(\d+)>\s*<([^<>]*)>\s*<([^<>]+)>\s*<([^<>]*)>|Information:\s*<(\d+)>\s*<([^<>]*)>\s*<([^<>]+)>\s*<([^<>]*)>"
|
||||
matches = re.findall(pattern, response_text, re.IGNORECASE | re.MULTILINE)
|
||||
|
||||
observations = []
|
||||
for match in matches:
|
||||
# Handle both Chinese and English patterns
|
||||
if match[0]: # Chinese pattern
|
||||
idx_str, time_info, content, keywords = match[0], match[1], match[2], match[3]
|
||||
else: # English pattern
|
||||
idx_str, time_info, content, keywords = match[4], match[5], match[6], match[7]
|
||||
|
||||
try:
|
||||
idx = int(idx_str)
|
||||
# Skip if content indicates no meaningful observation
|
||||
content_lower = content.lower().strip()
|
||||
if content_lower not in ['无', 'none', '', 'repeat']:
|
||||
observations.append({
|
||||
"index": idx,
|
||||
"time_info": time_info.strip() if time_info else "",
|
||||
"content": content.strip(),
|
||||
"keywords": keywords.strip() if keywords else ""
|
||||
})
|
||||
except ValueError:
|
||||
logger.warning(f"Invalid index format: {idx_str}")
|
||||
continue
|
||||
|
||||
return observations
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ from flowllm.schema.message import Message
|
|||
from loguru import logger
|
||||
|
||||
from reme_ai.schema.memory import BaseMemory, PersonalMemory
|
||||
from reme_ai.utils.op_utils import parse_reflection_subjects_response
|
||||
|
||||
|
||||
@C.register_op()
|
||||
|
|
@ -42,76 +41,103 @@ class GetReflectionSubjectOp(BaseLLMOp):
|
|||
|
||||
def execute(self):
|
||||
"""
|
||||
Executes the main logic of reflecting on personal memories to derive new insights.
|
||||
Generate reflection subjects (topics) from personal memories for insight extraction.
|
||||
|
||||
Steps include:
|
||||
- Retrieving personal memories from context.
|
||||
- Checking if there are enough memories to process.
|
||||
- Compiling existing insight subjects.
|
||||
- Generating a reflection prompt with system message, few-shot examples, and user queries.
|
||||
- Calling the language model for new insights.
|
||||
- Parsing the model's responses for new insight subjects.
|
||||
- Creating new insight memories and storing them in context.
|
||||
Process:
|
||||
1. Retrieve personal memories and existing insights from context
|
||||
2. Check if sufficient memories exist for reflection
|
||||
3. Generate new reflection subjects using LLM
|
||||
4. Create insight memory objects for new subjects
|
||||
5. Store results in context for next operation
|
||||
"""
|
||||
# Get personal memories from context
|
||||
personal_memories: List[BaseMemory] = self.context.response.metadata.get("personal_memories", [])
|
||||
existing_insights: List[BaseMemory] = self.context.response.metadata.get("existing_insights", [])
|
||||
|
||||
# Get parameters from operation config
|
||||
reflect_obs_cnt_threshold: int = self.op_params.get("reflect_obs_cnt_threshold", 10)
|
||||
reflect_num_questions: int = self.op_params.get("reflect_num_questions", 1)
|
||||
# Get memories from previous operation
|
||||
personal_memories = self.context.response.metadata.get("personal_memories", [])
|
||||
existing_insights = self.context.response.metadata.get("existing_insights", [])
|
||||
|
||||
# Get operation parameters
|
||||
reflect_obs_cnt_threshold = self.op_params.get("reflect_obs_cnt_threshold", 10)
|
||||
reflect_num_questions = self.op_params.get("reflect_num_questions", 3)
|
||||
user_name = self.context.get("user_name", "user")
|
||||
|
||||
# Check if we have enough memories to reflect on
|
||||
# Validate sufficient memories for reflection
|
||||
if len(personal_memories) < reflect_obs_cnt_threshold:
|
||||
logger.info(
|
||||
f"personal_memories count({len(personal_memories)}) < threshold({reflect_obs_cnt_threshold}), skip reflection.")
|
||||
logger.info(f"Insufficient memories for reflection: {len(personal_memories)} < {reflect_obs_cnt_threshold}")
|
||||
self.context.response.metadata["insight_memories"] = []
|
||||
return
|
||||
|
||||
# Compile existing insight subjects
|
||||
exist_keys: List[str] = []
|
||||
# Extract existing insight subjects to avoid duplication
|
||||
existing_subjects = []
|
||||
if existing_insights:
|
||||
exist_keys = [memory.content for memory in existing_insights if hasattr(memory, 'content')]
|
||||
existing_subjects = [memory.content for memory in existing_insights if
|
||||
hasattr(memory, 'content') and memory.content]
|
||||
|
||||
logger.info(f"exist_keys={exist_keys}")
|
||||
logger.info(f"Found {len(existing_subjects)} existing insight subjects")
|
||||
|
||||
# Generate reflection prompt components
|
||||
user_query_list = []
|
||||
# Prepare memory content for LLM analysis
|
||||
memory_contents = []
|
||||
for memory in personal_memories:
|
||||
if hasattr(memory, 'content') and memory.content:
|
||||
user_query_list.append(memory.content)
|
||||
if hasattr(memory, 'content') and memory.content.strip():
|
||||
memory_contents.append(memory.content.strip())
|
||||
|
||||
# Determine number of questions to ask
|
||||
if reflect_num_questions > 0:
|
||||
num_questions = reflect_num_questions
|
||||
else:
|
||||
num_questions = len(user_query_list)
|
||||
if not memory_contents:
|
||||
logger.warning("No valid memory content found for reflection")
|
||||
self.context.response.metadata["insight_memories"] = []
|
||||
return
|
||||
|
||||
# Create prompt using the prompt format method
|
||||
system_prompt = self.prompt_format(prompt_name="get_reflection_subject_system",
|
||||
user_name=user_name,
|
||||
num_questions=num_questions)
|
||||
few_shot = self.prompt_format(prompt_name="get_reflection_subject_few_shot", user_name=user_name)
|
||||
user_query = self.prompt_format(prompt_name="get_reflection_subject_user_query",
|
||||
user_name=user_name,
|
||||
exist_keys=", ".join(exist_keys),
|
||||
user_query="\n".join(user_query_list))
|
||||
# Generate reflection subjects using LLM
|
||||
insight_memories = self._generate_reflection_subjects(
|
||||
memory_contents, existing_subjects, user_name, reflect_num_questions
|
||||
)
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["insight_memories"] = insight_memories
|
||||
logger.info(f"Generated {len(insight_memories)} new reflection subject memories")
|
||||
|
||||
def _generate_reflection_subjects(self, memory_contents: List[str], existing_subjects: List[str],
|
||||
user_name: str, num_questions: int) -> List[BaseMemory]:
|
||||
"""
|
||||
Generate new reflection subjects using LLM analysis of memory contents.
|
||||
|
||||
Args:
|
||||
memory_contents: List of memory content strings
|
||||
existing_subjects: List of already existing subject strings
|
||||
user_name: Target user name
|
||||
num_questions: Maximum number of new subjects to generate
|
||||
|
||||
Returns:
|
||||
List of PersonalMemory objects representing new reflection subjects
|
||||
"""
|
||||
# Build LLM prompt
|
||||
system_prompt = self.prompt_format(
|
||||
prompt_name="get_reflection_subject_system",
|
||||
user_name=user_name,
|
||||
num_questions=num_questions
|
||||
)
|
||||
few_shot = self.prompt_format(
|
||||
prompt_name="get_reflection_subject_few_shot",
|
||||
user_name=user_name
|
||||
)
|
||||
user_query = self.prompt_format(
|
||||
prompt_name="get_reflection_subject_user_query",
|
||||
user_name=user_name,
|
||||
exist_keys=", ".join(existing_subjects) if existing_subjects else "None",
|
||||
user_query="\n".join(memory_contents)
|
||||
)
|
||||
|
||||
full_prompt = f"{system_prompt}\n\n{few_shot}\n\n{user_query}"
|
||||
logger.info(f"reflection_subject_prompt={full_prompt}")
|
||||
logger.info(f"Reflection subject prompt length: {len(full_prompt)} chars")
|
||||
|
||||
def parse_reflection_subjects(message: Message) -> List[BaseMemory]:
|
||||
def parse_reflection_response(message: Message) -> List[BaseMemory]:
|
||||
"""Parse LLM response and create insight memories"""
|
||||
response_text = message.content
|
||||
logger.info(f"reflection_subject_response={response_text}")
|
||||
logger.info(f"Reflection subjects response: {response_text}")
|
||||
|
||||
# Parse new insight subjects using utility function
|
||||
new_subjects = parse_reflection_subjects_response(response_text, exist_keys)
|
||||
# Parse new subjects using class method
|
||||
new_subjects = GetReflectionSubjectOp.parse_reflection_subjects_response(response_text, existing_subjects)
|
||||
|
||||
# Create insight memory objects
|
||||
insight_memories = []
|
||||
for subject in new_subjects:
|
||||
# Create insight memory
|
||||
insight_memory = self.new_insight_memory(
|
||||
insight_content=subject,
|
||||
target=user_name
|
||||
|
|
@ -121,13 +147,33 @@ class GetReflectionSubjectOp(BaseLLMOp):
|
|||
|
||||
return insight_memories
|
||||
|
||||
# Use LLM chat with callback function
|
||||
insight_memories = self.llm.chat(messages=[Message(content=full_prompt)], callback_fn=parse_reflection_subjects)
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["insight_memories"] = insight_memories
|
||||
logger.info(f"Generated {len(insight_memories)} reflection subject memories")
|
||||
# Generate subjects using LLM
|
||||
return self.llm.chat(messages=[Message(content=full_prompt)], callback_fn=parse_reflection_response)
|
||||
|
||||
def get_language_value(self, value_dict: dict):
|
||||
"""Get language-specific value from dictionary"""
|
||||
return value_dict.get(self.language, value_dict.get("en"))
|
||||
|
||||
@staticmethod
|
||||
def parse_reflection_subjects_response(response_text: str, existing_subjects: List[str] = None) -> List[str]:
|
||||
"""Parse reflection subjects response to extract new subject attributes"""
|
||||
if existing_subjects is None:
|
||||
existing_subjects = []
|
||||
|
||||
# Split response into lines and clean up
|
||||
lines = response_text.strip().split('\n')
|
||||
subjects = []
|
||||
|
||||
for line in lines:
|
||||
line = line.strip()
|
||||
# Skip empty lines, "None" responses, and existing subjects
|
||||
if (line and
|
||||
line not in ['无', 'None', ''] and
|
||||
line not in existing_subjects and
|
||||
not line.startswith('新增') and # Skip Chinese header
|
||||
not line.startswith('New ') and # Skip English header
|
||||
len(line) > 1): # Skip single character responses
|
||||
subjects.append(line)
|
||||
|
||||
logger.info(f"Parsed {len(subjects)} new reflection subjects from response")
|
||||
return subjects
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import re
|
||||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
|
|
@ -5,7 +6,6 @@ from flowllm.schema.message import Message
|
|||
from loguru import logger
|
||||
|
||||
from reme_ai.schema.memory import PersonalMemory
|
||||
from reme_ai.utils.op_utils import parse_info_filter_response
|
||||
|
||||
|
||||
@C.register_op()
|
||||
|
|
@ -18,8 +18,8 @@ class InfoFilterOp(BaseLLMOp):
|
|||
|
||||
def execute(self):
|
||||
"""Filter messages based on information content scores"""
|
||||
# Get messages from context
|
||||
messages: List[Message] = self.context.get("messages", [])
|
||||
# Get messages from context - guaranteed to exist by flow input
|
||||
messages: List[Message] = self.context.messages
|
||||
if not messages:
|
||||
logger.warning("No messages found in context")
|
||||
return
|
||||
|
|
@ -33,6 +33,7 @@ class InfoFilterOp(BaseLLMOp):
|
|||
info_messages = self._filter_and_process_messages(messages, user_name, info_filter_msg_max_size)
|
||||
if not info_messages:
|
||||
logger.warning("No messages left after filtering")
|
||||
self.context.response.metadata["memory_list"] = []
|
||||
return
|
||||
|
||||
logger.info(f"Filtering {len(info_messages)} messages for information content")
|
||||
|
|
@ -40,23 +41,28 @@ class InfoFilterOp(BaseLLMOp):
|
|||
# Filter messages using LLM
|
||||
filtered_memories = self._filter_messages_with_llm(info_messages, user_name, preserved_scores)
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["filtered_memories"] = filtered_memories
|
||||
# Store results in context using standardized key
|
||||
self.context.response.metadata["memory_list"] = filtered_memories
|
||||
logger.info(f"Filtered to {len(filtered_memories)} high-information messages")
|
||||
|
||||
def _filter_and_process_messages(self, messages: List[Message], user_name: str, max_size: int) -> List[Message]:
|
||||
@staticmethod
|
||||
def _filter_and_process_messages(messages: List[Message], user_name: str, max_size: int) -> List[Message]:
|
||||
"""Filter and process messages for information filtering"""
|
||||
info_messages = []
|
||||
|
||||
for msg in messages:
|
||||
# Ensure metadata exists
|
||||
|
||||
# Skip memorized messages
|
||||
if hasattr(msg, 'memorized') and msg.memorized:
|
||||
if msg.metadata.get('memorized', False):
|
||||
continue
|
||||
|
||||
# Only process messages from the target user
|
||||
if hasattr(msg, 'role_name') and msg.role_name != user_name:
|
||||
continue
|
||||
elif hasattr(msg, 'role') and msg.role != 'user':
|
||||
# role_name = msg.metadata.get('role_name')
|
||||
# if role_name and role_name != user_name:
|
||||
# continue
|
||||
|
||||
elif msg.role.value != "user":
|
||||
continue
|
||||
|
||||
# Truncate long messages
|
||||
|
|
@ -95,8 +101,8 @@ class InfoFilterOp(BaseLLMOp):
|
|||
response_text = message.content
|
||||
logger.info(f"info_filter_response={response_text}")
|
||||
|
||||
# Parse scores using utility function
|
||||
info_scores = parse_info_filter_response(response_text)
|
||||
# Parse scores using class method
|
||||
info_scores = InfoFilterOp.parse_info_filter_response(response_text)
|
||||
|
||||
if len(info_scores) != len(info_messages):
|
||||
logger.warning(f"score_size != messages_size, {len(info_scores)} vs {len(info_messages)}")
|
||||
|
|
@ -113,7 +119,10 @@ class InfoFilterOp(BaseLLMOp):
|
|||
if score in preserved_scores:
|
||||
message_obj = info_messages[msg_idx]
|
||||
|
||||
# Create memory from filtered message
|
||||
# Get original message metadata or create empty dict
|
||||
original_metadata = getattr(message_obj, 'metadata', {}) or {}
|
||||
|
||||
# Create memory from filtered message with combined metadata
|
||||
memory = PersonalMemory(
|
||||
workspace_id=self.context.get("workspace_id", ""),
|
||||
content=message_obj.content,
|
||||
|
|
@ -122,7 +131,10 @@ class InfoFilterOp(BaseLLMOp):
|
|||
metadata={
|
||||
"info_score": score,
|
||||
"filter_type": "info_content",
|
||||
"original_message_time": getattr(message_obj, 'time_created', None)
|
||||
"original_message_time": getattr(message_obj, 'time_created', None),
|
||||
"role_name": original_metadata.get('role_name', user_name),
|
||||
"memorized": True,
|
||||
**original_metadata # Include all original metadata
|
||||
}
|
||||
)
|
||||
filtered_memories.append(memory)
|
||||
|
|
@ -137,3 +149,31 @@ class InfoFilterOp(BaseLLMOp):
|
|||
"""Get language-specific colon word"""
|
||||
colon_dict = {"zh": "：", "cn": "：", "en": ": "}
|
||||
return colon_dict.get(self.language, ": ")
|
||||
|
||||
@staticmethod
|
||||
def parse_info_filter_response(response_text: str) -> List[tuple]:
|
||||
"""Parse info filter response to extract message scores"""
|
||||
# Pattern to match both Chinese and English result formats
|
||||
# Chinese: 结果：<序号> <分数>
|
||||
# English: Result: <Index> <Score>
|
||||
pattern = r"结果：<(\d+)>\s*<([0-3])>|Result:\s*<(\d+)>\s*<([0-3])>"
|
||||
matches = re.findall(pattern, response_text, re.IGNORECASE | re.MULTILINE)
|
||||
|
||||
scores = []
|
||||
for match in matches:
|
||||
# Handle both Chinese and English patterns
|
||||
if match[0]: # Chinese pattern
|
||||
idx_str, score_str = match[0], match[1]
|
||||
else: # English pattern
|
||||
idx_str, score_str = match[2], match[3]
|
||||
|
||||
try:
|
||||
idx = int(idx_str)
|
||||
score = score_str
|
||||
scores.append((idx, score))
|
||||
except ValueError:
|
||||
logger.warning(f"Invalid index or score format: {idx_str}, {score_str}")
|
||||
continue
|
||||
|
||||
logger.info(f"Parsed {len(scores)} info filter scores from response")
|
||||
return scores
|
||||
|
|
|
|||
|
|
@ -1,165 +0,0 @@
|
|||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
from loguru import logger
|
||||
|
||||
from reme_ai.schema.memory import PersonalMemory
|
||||
from reme_ai.utils.datetime_handler import DatetimeHandler
|
||||
from reme_ai.utils.op_utils import load_memories_from_vector_store
|
||||
|
||||
|
||||
@C.register_op()
|
||||
class LoadMemoryOp(BaseLLMOp):
|
||||
"""
|
||||
A specialized operation class to load various types of personal memories using BaseLLMOp.
|
||||
This loads different categories of memories including observations, insights, and recent memories.
|
||||
"""
|
||||
file_path: str = __file__
|
||||
|
||||
def execute(self):
|
||||
"""
|
||||
Executes the main routine of the LoadMemoryOp. This involves loading various types
|
||||
of personal memories based on the configuration parameters.
|
||||
"""
|
||||
# Get operation parameters
|
||||
retrieve_not_reflected_top_k: int = self.op_params.get("retrieve_not_reflected_top_k", 0)
|
||||
retrieve_not_updated_top_k: int = self.op_params.get("retrieve_not_updated_top_k", 0)
|
||||
retrieve_insight_top_k: int = self.op_params.get("retrieve_insight_top_k", 0)
|
||||
retrieve_today_top_k: int = self.op_params.get("retrieve_today_top_k", 0)
|
||||
|
||||
# Get context parameters
|
||||
workspace_id = self.context.get("workspace_id", "")
|
||||
user_name = self.context.get("user_name", "user")
|
||||
|
||||
logger.info(f"Loading memories for user: {user_name} in workspace: {workspace_id}")
|
||||
|
||||
# Load different types of memories
|
||||
all_memories = []
|
||||
|
||||
# Load not reflected memories
|
||||
if retrieve_not_reflected_top_k > 0:
|
||||
not_reflected_memories = self._retrieve_not_reflected_memories(
|
||||
workspace_id, user_name, retrieve_not_reflected_top_k
|
||||
)
|
||||
all_memories.extend(not_reflected_memories)
|
||||
logger.info(f"Loaded {len(not_reflected_memories)} not reflected memories")
|
||||
|
||||
# Load not updated memories
|
||||
if retrieve_not_updated_top_k > 0:
|
||||
not_updated_memories = self._retrieve_not_updated_memories(
|
||||
workspace_id, user_name, retrieve_not_updated_top_k
|
||||
)
|
||||
all_memories.extend(not_updated_memories)
|
||||
logger.info(f"Loaded {len(not_updated_memories)} not updated memories")
|
||||
|
||||
# Load insight memories
|
||||
if retrieve_insight_top_k > 0:
|
||||
insight_memories = self._retrieve_insight_memories(
|
||||
workspace_id, user_name, retrieve_insight_top_k
|
||||
)
|
||||
all_memories.extend(insight_memories)
|
||||
logger.info(f"Loaded {len(insight_memories)} insight memories")
|
||||
|
||||
# Load today's memories
|
||||
if retrieve_today_top_k > 0:
|
||||
today_memories = self._retrieve_today_memories(
|
||||
workspace_id, user_name, retrieve_today_top_k
|
||||
)
|
||||
all_memories.extend(today_memories)
|
||||
logger.info(f"Loaded {len(today_memories)} today's memories")
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["loaded_memories"] = all_memories
|
||||
self.context.response.metadata["not_reflected_memories"] = [
|
||||
m for m in all_memories if m.metadata.get("memory_category") == "not_reflected"
|
||||
]
|
||||
self.context.response.metadata["not_updated_memories"] = [
|
||||
m for m in all_memories if m.metadata.get("memory_category") == "not_updated"
|
||||
]
|
||||
self.context.response.metadata["insight_memories"] = [
|
||||
m for m in all_memories if m.metadata.get("memory_category") == "insight"
|
||||
]
|
||||
self.context.response.metadata["today_memories"] = [
|
||||
m for m in all_memories if m.metadata.get("memory_category") == "today"
|
||||
]
|
||||
|
||||
logger.info(f"Total memories loaded: {len(all_memories)}")
|
||||
|
||||
def _retrieve_not_reflected_memories(self, workspace_id: str, user_name: str, top_k: int) -> List[PersonalMemory]:
|
||||
"""
|
||||
Retrieves top-K not reflected memories based on the query.
|
||||
"""
|
||||
filter_criteria = {
|
||||
"memory_type": "personal",
|
||||
"target": user_name,
|
||||
"reflected": False
|
||||
}
|
||||
|
||||
memories = load_memories_from_vector_store(
|
||||
workspace_id=workspace_id,
|
||||
filter_criteria=filter_criteria,
|
||||
top_k=top_k,
|
||||
memory_category="not_reflected"
|
||||
)
|
||||
|
||||
return memories
|
||||
|
||||
def _retrieve_not_updated_memories(self, workspace_id: str, user_name: str, top_k: int) -> List[PersonalMemory]:
|
||||
"""
|
||||
Retrieves top-K not updated memories based on the query.
|
||||
"""
|
||||
filter_criteria = {
|
||||
"memory_type": "personal",
|
||||
"target": user_name,
|
||||
"updated": False
|
||||
}
|
||||
|
||||
memories = load_memories_from_vector_store(
|
||||
workspace_id=workspace_id,
|
||||
filter_criteria=filter_criteria,
|
||||
top_k=top_k,
|
||||
memory_category="not_updated"
|
||||
)
|
||||
|
||||
return memories
|
||||
|
||||
def _retrieve_insight_memories(self, workspace_id: str, user_name: str, top_k: int) -> List[PersonalMemory]:
|
||||
"""
|
||||
Retrieves top-K insight memories based on the query.
|
||||
"""
|
||||
filter_criteria = {
|
||||
"memory_type": "personal_insight",
|
||||
"target": user_name
|
||||
}
|
||||
|
||||
memories = load_memories_from_vector_store(
|
||||
workspace_id=workspace_id,
|
||||
filter_criteria=filter_criteria,
|
||||
top_k=top_k,
|
||||
memory_category="insight"
|
||||
)
|
||||
|
||||
return memories
|
||||
|
||||
def _retrieve_today_memories(self, workspace_id: str, user_name: str, top_k: int) -> List[PersonalMemory]:
|
||||
"""
|
||||
Retrieves top-K memories from today based on the query.
|
||||
"""
|
||||
# Get today's date
|
||||
dt = DatetimeHandler().datetime_format()
|
||||
today_date = dt.split()[0] # Extract date part
|
||||
|
||||
filter_criteria = {
|
||||
"memory_type": "personal",
|
||||
"target": user_name,
|
||||
"created_date": today_date
|
||||
}
|
||||
|
||||
memories = load_memories_from_vector_store(
|
||||
workspace_id=workspace_id,
|
||||
filter_criteria=filter_criteria,
|
||||
top_k=top_k,
|
||||
memory_category="today"
|
||||
)
|
||||
|
||||
return memories
|
||||
117
reme_ai/summary/personal/load_today_memory_op.py
Normal file
117
reme_ai/summary/personal/load_today_memory_op.py
Normal file
|
|
@ -0,0 +1,117 @@
|
|||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
from flowllm.schema.vector_node import VectorNode
|
||||
from loguru import logger
|
||||
|
||||
from reme_ai.schema.memory import BaseMemory, vector_node_to_memory
|
||||
from reme_ai.utils.datetime_handler import DatetimeHandler
|
||||
|
||||
|
||||
@C.register_op()
|
||||
class LoadTodayMemoryOp(BaseLLMOp):
|
||||
"""
|
||||
Operation to load today's memories from vector store for deduplication.
|
||||
Focuses specifically on retrieving and deduplicating memories from the current date.
|
||||
"""
|
||||
file_path: str = __file__
|
||||
|
||||
def execute(self):
|
||||
"""
|
||||
Load today's memories from vector store and perform deduplication.
|
||||
|
||||
This operation:
|
||||
1. Retrieves memories from today using vector store search
|
||||
2. Converts vector nodes to memory objects
|
||||
3. Performs deduplication based on content similarity
|
||||
4. Stores deduplicated memories in context
|
||||
"""
|
||||
# Get operation parameters
|
||||
top_k = self.op_params.get("top_k", 50)
|
||||
|
||||
# Get required context values
|
||||
workspace_id = self.context.workspace_id
|
||||
user_name = self.context.get("user_name", "user")
|
||||
|
||||
logger.info(f"Loading today's memories for user: {user_name} (top_k: {top_k})")
|
||||
|
||||
# Get today's memories from vector store
|
||||
today_memories = self._retrieve_today_memories(workspace_id, user_name, top_k)
|
||||
|
||||
if not today_memories:
|
||||
logger.info("No memories found for today")
|
||||
self.context.today_memories = []
|
||||
return
|
||||
|
||||
logger.info(f"Retrieved {len(today_memories)} memories from today")
|
||||
self.context.today_memories = today_memories
|
||||
logger.info(f"Final today's memory list size: {len(today_memories)}")
|
||||
|
||||
def _retrieve_today_memories(self, workspace_id: str, user_name: str, top_k: int) -> List[BaseMemory]:
|
||||
"""
|
||||
Retrieve memories from today using vector store with date filtering.
|
||||
|
||||
Args:
|
||||
workspace_id: Workspace identifier
|
||||
user_name: Target username
|
||||
top_k: Maximum number of memories to retrieve
|
||||
|
||||
Returns:
|
||||
List of today's memories
|
||||
"""
|
||||
try:
|
||||
# Get today's date for filtering
|
||||
dt_handler = DatetimeHandler()
|
||||
today_date = dt_handler.datetime_format().split()[0] # Extract date part (YYYY-MM-DD)
|
||||
|
||||
logger.info(f"Searching for memories from date: {today_date}")
|
||||
|
||||
# Create filter criteria for today's memories
|
||||
filter_dict = {
|
||||
"memory_type": "personal",
|
||||
"target": user_name,
|
||||
"created_date": today_date
|
||||
}
|
||||
|
||||
# Search vector store with date filter
|
||||
nodes: List[VectorNode] = self.vector_store.search(
|
||||
query="", # Empty query to get all results for today
|
||||
workspace_id=workspace_id,
|
||||
top_k=top_k,
|
||||
filter_dict=filter_dict
|
||||
)
|
||||
|
||||
logger.info(f"Vector store returned {len(nodes)} nodes for today")
|
||||
|
||||
# Convert vector nodes to memory objects
|
||||
memories = self._convert_nodes_to_memories(nodes)
|
||||
logger.info(f"Successfully converted {len(memories)} nodes to memories")
|
||||
|
||||
return memories
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error retrieving today's memories: {e}")
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def _convert_nodes_to_memories(nodes: List[VectorNode]) -> List[BaseMemory]:
|
||||
"""
|
||||
Convert vector nodes to memory objects.
|
||||
|
||||
Args:
|
||||
nodes: List of vector nodes from vector store
|
||||
|
||||
Returns:
|
||||
List of converted memory objects
|
||||
"""
|
||||
memories = []
|
||||
|
||||
for i, node in enumerate(nodes):
|
||||
try:
|
||||
memory = vector_node_to_memory(node)
|
||||
memories.append(memory)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to convert node {i} to memory: {e}")
|
||||
continue
|
||||
|
||||
return memories
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
import re
|
||||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
|
|
@ -6,7 +7,6 @@ from flowllm.schema.message import Message
|
|||
from loguru import logger
|
||||
|
||||
from reme_ai.schema.memory import BaseMemory, PersonalMemory
|
||||
from reme_ai.utils.op_utils import parse_long_contra_repeat_response
|
||||
|
||||
|
||||
@C.register_op()
|
||||
|
|
@ -21,78 +21,104 @@ class LongContraRepeatOp(BaseLLMOp):
|
|||
|
||||
def execute(self):
|
||||
"""
|
||||
Executes the primary routine of the LongContraRepeatOp which involves:
|
||||
1. Gets memory list from context
|
||||
2. Retrieves similar memories for each memory
|
||||
3. Constructs a prompt with these memories for language model analysis
|
||||
4. Parses the model's response to detect contradictions or redundancies
|
||||
5. Filters and returns the processed memories
|
||||
Analyze memories for contradictions and redundancies, resolving conflicts.
|
||||
|
||||
Process:
|
||||
1. Get updated insight memories from previous operation
|
||||
2. Check for contradictions and redundancies among memories
|
||||
3. Resolve conflicts by keeping most recent/accurate information
|
||||
4. Filter out redundant memories
|
||||
5. Store cleaned memory list in context
|
||||
"""
|
||||
# Get memory list from context
|
||||
memory_list: List[BaseMemory] = self.context.response.metadata.get("memory_list", [])
|
||||
# Get memories from previous operation
|
||||
updated_insights = self.context.response.metadata.get("updated_insight_memories", [])
|
||||
|
||||
if not memory_list:
|
||||
logger.info("memory_list is empty!")
|
||||
if not updated_insights:
|
||||
logger.info("No updated insight memories to process for contradictions")
|
||||
self.context.response.metadata["memory_list"] = []
|
||||
return
|
||||
|
||||
# Get operation parameters
|
||||
long_contra_repeat_max_count: int = self.op_params.get("long_contra_repeat_max_count", 50)
|
||||
enable_long_contra_repeat: bool = self.op_params.get("enable_long_contra_repeat", True)
|
||||
max_memories_to_process = self.op_params.get("long_contra_repeat_max_count", 50)
|
||||
enable_processing = self.op_params.get("enable_long_contra_repeat", True)
|
||||
|
||||
if not enable_long_contra_repeat:
|
||||
logger.warning("long_contra_repeat is not enabled!")
|
||||
if not enable_processing:
|
||||
logger.info("Long contradiction/repeat processing is disabled")
|
||||
self.context.response.metadata["memory_list"] = updated_insights
|
||||
return
|
||||
|
||||
# Sort and limit memories by count
|
||||
sorted_memories = sorted(memory_list, key=lambda x: getattr(x, 'created_time', ''), reverse=True)[
|
||||
:long_contra_repeat_max_count]
|
||||
# Sort memories by creation time (most recent first) and limit count
|
||||
sorted_memories = sorted(
|
||||
updated_insights,
|
||||
key=lambda x: getattr(x, 'created_time', ''),
|
||||
reverse=True
|
||||
)[:max_memories_to_process]
|
||||
|
||||
if len(sorted_memories) <= 1:
|
||||
logger.info("sorted_memories.size<=1, stop.")
|
||||
logger.info("Only one memory to process, skipping contradiction analysis")
|
||||
self.context.response.metadata["memory_list"] = sorted_memories
|
||||
return
|
||||
|
||||
# Build prompt
|
||||
user_query_list = []
|
||||
for i, memory in enumerate(sorted_memories):
|
||||
user_query_list.append(f"{i + 1} {memory.content}")
|
||||
logger.info(f"Processing {len(sorted_memories)} memories for contradictions and redundancies")
|
||||
|
||||
# Analyze and resolve contradictions
|
||||
filtered_memories = self._analyze_and_resolve_conflicts(sorted_memories)
|
||||
|
||||
# Store results in context
|
||||
self.context.response.metadata["memory_list"] = filtered_memories
|
||||
logger.info(f"Conflict resolution: {len(sorted_memories)} -> {len(filtered_memories)} memories")
|
||||
|
||||
def _analyze_and_resolve_conflicts(self, memories: List[BaseMemory]) -> List[BaseMemory]:
|
||||
"""
|
||||
Analyze memories for contradictions and redundancies using LLM.
|
||||
|
||||
Args:
|
||||
memories: List of memories to analyze
|
||||
|
||||
Returns:
|
||||
List of filtered memories with conflicts resolved
|
||||
"""
|
||||
user_name = self.context.get("user_name", "user")
|
||||
|
||||
# Create prompt using the new pattern
|
||||
system_prompt = self.prompt_format(prompt_name="long_contra_repeat_system",
|
||||
num_obs=len(user_query_list),
|
||||
user_name=user_name)
|
||||
few_shot = self.prompt_format(prompt_name="long_contra_repeat_few_shot", user_name=user_name)
|
||||
user_query = self.prompt_format(prompt_name="long_contra_repeat_user_query",
|
||||
user_query="\n".join(user_query_list))
|
||||
# Prepare memory content for LLM analysis
|
||||
memory_texts = []
|
||||
for i, memory in enumerate(memories):
|
||||
memory_texts.append(f"{i + 1} {memory.content}")
|
||||
|
||||
# Build LLM prompt
|
||||
system_prompt = self.prompt_format(
|
||||
prompt_name="long_contra_repeat_system",
|
||||
num_obs=len(memory_texts),
|
||||
user_name=user_name
|
||||
)
|
||||
few_shot = self.prompt_format(
|
||||
prompt_name="long_contra_repeat_few_shot",
|
||||
user_name=user_name
|
||||
)
|
||||
user_query = self.prompt_format(
|
||||
prompt_name="long_contra_repeat_user_query",
|
||||
user_query="\n".join(memory_texts)
|
||||
)
|
||||
|
||||
full_prompt = f"{system_prompt}\n\n{few_shot}\n\n{user_query}"
|
||||
logger.info(f"long_contra_repeat_prompt={full_prompt}")
|
||||
logger.info(f"Contradiction analysis prompt length: {len(full_prompt)} chars")
|
||||
|
||||
# Call LLM
|
||||
# Get LLM analysis
|
||||
response = self.llm.chat([Message(role=Role.USER, content=full_prompt)])
|
||||
|
||||
# Return if empty
|
||||
if not response or not response.content:
|
||||
logger.warning("Empty response from LLM")
|
||||
return
|
||||
|
||||
response_text = response.content
|
||||
logger.info(f"long_contra_repeat_response={response_text}")
|
||||
logger.warning("Empty response from LLM, keeping all memories")
|
||||
return memories
|
||||
|
||||
# Parse response and filter memories
|
||||
filtered_memories = self._parse_and_filter_memories(response_text, sorted_memories, user_name)
|
||||
return self._parse_and_filter_memories(response.content, memories, user_name)
|
||||
|
||||
# Update context with filtered memories
|
||||
self.context.response.metadata["memory_list"] = filtered_memories
|
||||
logger.info(f"Filtered {len(memory_list)} memories to {len(filtered_memories)} memories")
|
||||
|
||||
def _parse_and_filter_memories(self, response_text: str, memories: List[BaseMemory], user_name: str) -> List[
|
||||
BaseMemory]:
|
||||
@staticmethod
|
||||
def _parse_and_filter_memories(response_text: str, memories: List[BaseMemory], user_name: str) -> List[BaseMemory]:
|
||||
"""Parse LLM response and filter memories based on contradiction/containment analysis"""
|
||||
|
||||
# Use utility function to parse the response
|
||||
judgments = parse_long_contra_repeat_response(response_text)
|
||||
# Use class method to parse the response
|
||||
judgments = LongContraRepeatOp.parse_long_contra_repeat_response(response_text)
|
||||
|
||||
if not judgments:
|
||||
logger.warning("No valid judgments found in response")
|
||||
|
|
@ -155,3 +181,30 @@ class LongContraRepeatOp(BaseLLMOp):
|
|||
def get_language_value(self, value_dict: dict):
|
||||
"""Get language-specific value from dictionary"""
|
||||
return value_dict.get(self.language, value_dict.get("en"))
|
||||
|
||||
@staticmethod
|
||||
def parse_long_contra_repeat_response(response_text: str) -> List[tuple]:
|
||||
"""Parse long contra repeat response to extract judgments"""
|
||||
# Pattern to match both Chinese and English judgment formats
|
||||
# Chinese: 判断：<序号> <矛盾|被包含|无> <修改后的内容>
|
||||
# English: Judgment: <Index> <Contradiction|Contained|None> <Modified content>
|
||||
pattern = r"判断：<(\d+)>\s*<(矛盾|被包含|无)>\s*<([^<>]*)>|Judgment:\s*<(\d+)>\s*<(Contradiction|Contained|None)>\s*<([^<>]*)>"
|
||||
matches = re.findall(pattern, response_text, re.IGNORECASE | re.MULTILINE)
|
||||
|
||||
judgments = []
|
||||
for match in matches:
|
||||
# Handle both Chinese and English patterns
|
||||
if match[0]: # Chinese pattern
|
||||
idx_str, judgment, modified_content = match[0], match[1], match[2]
|
||||
else: # English pattern
|
||||
idx_str, judgment, modified_content = match[3], match[4], match[5]
|
||||
|
||||
try:
|
||||
idx = int(idx_str)
|
||||
judgments.append((idx, judgment, modified_content))
|
||||
except ValueError:
|
||||
logger.warning(f"Invalid index format: {idx_str}")
|
||||
continue
|
||||
|
||||
logger.info(f"Parsed {len(judgments)} long contra repeat judgments from response")
|
||||
return judgments
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import re
|
||||
from typing import List
|
||||
|
||||
from flowllm import C, BaseLLMOp
|
||||
|
|
@ -5,7 +6,6 @@ from flowllm.schema.message import Message
|
|||
from loguru import logger
|
||||
|
||||
from reme_ai.schema.memory import PersonalMemory
|
||||
from reme_ai.utils.op_utils import parse_update_insight_response
|
||||
|
||||
|
||||
@C.register_op()
|
||||
|
|
@ -19,107 +19,132 @@ class UpdateInsightOp(BaseLLMOp):
|
|||
|
||||
def execute(self):
|
||||
"""
|
||||
Executes the main routine of the UpdateInsightOp. This involves filtering and updating insight nodes
|
||||
based on their association with observed nodes.
|
||||
Update insight values based on new observation memories.
|
||||
|
||||
Process:
|
||||
1. Get insight subjects and personal memories from context
|
||||
2. Find relevant observations for each insight subject
|
||||
3. Update insight values using LLM integration
|
||||
4. Store updated insights in context
|
||||
"""
|
||||
# Get insight memories from context
|
||||
insight_memories: List[PersonalMemory] = self.context.response.metadata.get("insight_memories", [])
|
||||
observation_memories: List[PersonalMemory] = self.context.response.metadata.get("observation_memories", [])
|
||||
# Get memories from previous operations
|
||||
insight_memories = self.context.response.metadata.get("insight_memories", [])
|
||||
personal_memories = self.context.response.metadata.get("personal_memories", [])
|
||||
|
||||
if not insight_memories:
|
||||
logger.warning("insight_memories is empty, stopping processing.")
|
||||
logger.info("No insight memories to update")
|
||||
self.context.response.metadata["updated_insight_memories"] = []
|
||||
return
|
||||
|
||||
if not observation_memories:
|
||||
logger.warning("observation_memories is empty, stopping processing.")
|
||||
if not personal_memories:
|
||||
logger.info("No observation memories available for insight updates")
|
||||
self.context.response.metadata["updated_insight_memories"] = []
|
||||
return
|
||||
|
||||
# Get operation parameters
|
||||
update_insight_threshold: float = self.op_params.get("update_insight_threshold", 0.1)
|
||||
update_insight_max_count: int = self.op_params.get("update_insight_max_count", 5)
|
||||
update_insight_threshold = self.op_params.get("update_insight_threshold", 0.3)
|
||||
update_insight_max_count = self.op_params.get("update_insight_max_count", 5)
|
||||
user_name = self.context.get("user_name", "user")
|
||||
|
||||
logger.info(
|
||||
f"Processing {len(insight_memories)} insight memories with {len(observation_memories)} observations")
|
||||
logger.info(f"Updating {len(insight_memories)} insights with {len(personal_memories)} observations")
|
||||
|
||||
# Filter and score insight memories based on relevance to observations
|
||||
scored_insights = self._filter_and_score_insights(insight_memories, observation_memories,
|
||||
update_insight_threshold, user_name)
|
||||
# Score and filter insights based on relevance to observations
|
||||
scored_insights = self._score_insights_by_relevance(
|
||||
insight_memories, personal_memories, update_insight_threshold
|
||||
)
|
||||
|
||||
if not scored_insights:
|
||||
logger.warning("No relevant insights found after filtering")
|
||||
logger.info("No insights meet relevance threshold for updating")
|
||||
self.context.response.metadata["updated_insight_memories"] = []
|
||||
return
|
||||
|
||||
# Select top insights to update
|
||||
# Select top insights for updating
|
||||
top_insights = sorted(scored_insights, key=lambda x: x[1], reverse=True)[:update_insight_max_count]
|
||||
logger.info(f"Selected {len(top_insights)} insights for updating")
|
||||
|
||||
# Update each selected insight
|
||||
updated_insights = []
|
||||
for insight_memory, score, relevant_observations in top_insights:
|
||||
updated_insight = self._update_single_insight(insight_memory, relevant_observations, user_name)
|
||||
for insight_memory, relevance_score, relevant_observations in top_insights:
|
||||
updated_insight = self._update_insight_with_observations(
|
||||
insight_memory, relevant_observations, user_name
|
||||
)
|
||||
if updated_insight:
|
||||
updated_insights.append(updated_insight)
|
||||
|
||||
# Store updated insights in context
|
||||
# Store results in context
|
||||
self.context.response.metadata["updated_insight_memories"] = updated_insights
|
||||
logger.info(f"Successfully updated {len(updated_insights)} insight memories")
|
||||
|
||||
def _filter_and_score_insights(self, insight_memories: List[PersonalMemory],
|
||||
def _score_insights_by_relevance(self, insight_memories: List[PersonalMemory],
|
||||
observation_memories: List[PersonalMemory],
|
||||
threshold: float, user_name: str) -> List[tuple]:
|
||||
threshold: float) -> List[tuple]:
|
||||
"""
|
||||
Filter and score insight memories based on their relevance to observation memories.
|
||||
Score insight memories based on relevance to observation memories.
|
||||
|
||||
Args:
|
||||
insight_memories: List of insight memories to score
|
||||
observation_memories: List of observation memories for comparison
|
||||
threshold: Minimum relevance score threshold
|
||||
|
||||
Returns:
|
||||
List[tuple]: List of (insight_memory, max_score, relevant_observations)
|
||||
List[tuple]: List of (insight_memory, relevance_score, relevant_observations)
|
||||
"""
|
||||
scored_insights = []
|
||||
|
||||
for insight_memory in insight_memories:
|
||||
# For each insight, find observations that are relevant to the same subject
|
||||
relevant_observations = []
|
||||
max_score = 0.0
|
||||
max_relevance = 0.0
|
||||
|
||||
insight_subject = insight_memory.reflection_subject or ""
|
||||
insight_subject = getattr(insight_memory, 'reflection_subject', '') or insight_memory.content
|
||||
insight_keywords = set(insight_memory.content.lower().split())
|
||||
|
||||
# Find observations relevant to this insight
|
||||
for obs_memory in observation_memories:
|
||||
score = 0.0
|
||||
relevance_score = self._calculate_relevance_score(
|
||||
insight_memory, obs_memory, insight_keywords
|
||||
)
|
||||
|
||||
# If both have the same reflection subject, they're highly relevant
|
||||
if (insight_subject and
|
||||
hasattr(obs_memory, 'reflection_subject') and
|
||||
obs_memory.reflection_subject == insight_subject):
|
||||
score = 0.8
|
||||
else:
|
||||
# Otherwise, use keyword-based similarity
|
||||
obs_keywords = set(obs_memory.content.lower().split())
|
||||
intersection = len(insight_keywords.intersection(obs_keywords))
|
||||
union = len(insight_keywords.union(obs_keywords))
|
||||
score = intersection / union if union > 0 else 0.0
|
||||
|
||||
if score >= threshold:
|
||||
if relevance_score >= threshold:
|
||||
relevant_observations.append(obs_memory)
|
||||
max_score = max(max_score, score)
|
||||
max_relevance = max(max_relevance, relevance_score)
|
||||
|
||||
# Include insight if it has relevant observations
|
||||
if relevant_observations:
|
||||
scored_insights.append((insight_memory, max_score, relevant_observations))
|
||||
scored_insights.append((insight_memory, max_relevance, relevant_observations))
|
||||
logger.info(
|
||||
f"Insight '{insight_memory.content[:50]}...' (subject: {insight_subject}) scored {max_score:.3f} with {len(relevant_observations)} relevant observations")
|
||||
f"Insight '{insight_subject[:40]}...' scored {max_relevance:.3f} with {len(relevant_observations)} observations"
|
||||
)
|
||||
|
||||
return scored_insights
|
||||
|
||||
def _update_single_insight(self, insight_memory: PersonalMemory,
|
||||
relevant_observations: List[PersonalMemory],
|
||||
user_name: str) -> PersonalMemory:
|
||||
@staticmethod
|
||||
def _calculate_relevance_score(insight_memory: PersonalMemory,
|
||||
obs_memory: PersonalMemory, insight_keywords: set) -> float:
|
||||
"""Calculate relevance score between insight and observation memory"""
|
||||
# High relevance for same reflection subject
|
||||
insight_subject = getattr(insight_memory, 'reflection_subject', '')
|
||||
obs_subject = getattr(obs_memory, 'reflection_subject', '')
|
||||
|
||||
if insight_subject and obs_subject and insight_subject == obs_subject:
|
||||
return 0.9
|
||||
|
||||
# Medium relevance for keyword overlap
|
||||
obs_keywords = set(obs_memory.content.lower().split())
|
||||
intersection = len(insight_keywords.intersection(obs_keywords))
|
||||
union = len(insight_keywords.union(obs_keywords))
|
||||
|
||||
return intersection / union if union > 0 else 0.0
|
||||
|
||||
def _update_insight_with_observations(self, insight_memory: PersonalMemory,
|
||||
relevant_observations: List[PersonalMemory],
|
||||
user_name: str) -> PersonalMemory:
|
||||
"""
|
||||
Update a single insight memory based on relevant observations using LLM.
|
||||
|
||||
Args:
|
||||
insight_memory: The insight memory to update
|
||||
relevant_observations: List of relevant observation memories
|
||||
user_name: The target user name
|
||||
user_name: The target username
|
||||
|
||||
Returns:
|
||||
PersonalMemory: Updated insight memory or None if update failed
|
||||
|
|
@ -150,7 +175,7 @@ class UpdateInsightOp(BaseLLMOp):
|
|||
logger.info(f"update_insight_response={response_text}")
|
||||
|
||||
# Parse the response to extract updated insight
|
||||
updated_content = parse_update_insight_response(response_text, self.language)
|
||||
updated_content = UpdateInsightOp.parse_update_insight_response(response_text, self.language)
|
||||
|
||||
if not updated_content or updated_content.lower() in ['无', 'none', '']:
|
||||
logger.info(f"No update needed for insight: {insight_memory.content[:50]}...")
|
||||
|
|
@ -187,3 +212,33 @@ class UpdateInsightOp(BaseLLMOp):
|
|||
except Exception as e:
|
||||
logger.error(f"Error updating insight: {e}")
|
||||
return insight_memory
|
||||
|
||||
@staticmethod
|
||||
def parse_update_insight_response(response_text: str, language: str = "en") -> str:
|
||||
"""Parse update insight response to extract updated insight content"""
|
||||
# Pattern to match both Chinese and English insight formats
|
||||
# Chinese: {user_name}的资料: <信息>
|
||||
# English: {user_name}'s profile: <Information>
|
||||
if language in ["zh", "cn"]:
|
||||
pattern = r"的资料[：:]\s*<([^<>]+)>"
|
||||
else:
|
||||
pattern = r"profile[：:]\s*<([^<>]+)>"
|
||||
|
||||
matches = re.findall(pattern, response_text, re.IGNORECASE | re.MULTILINE)
|
||||
|
||||
if matches:
|
||||
insight_content = matches[0].strip()
|
||||
logger.info(f"Parsed insight content: {insight_content}")
|
||||
return insight_content
|
||||
|
||||
# Fallback: try to find content between angle brackets
|
||||
fallback_pattern = r"<([^<>]+)>"
|
||||
fallback_matches = re.findall(fallback_pattern, response_text)
|
||||
if fallback_matches:
|
||||
# Get the last match as it's likely the final answer
|
||||
insight_content = fallback_matches[-1].strip()
|
||||
logger.info(f"Parsed insight content (fallback): {insight_content}")
|
||||
return insight_content
|
||||
|
||||
logger.warning("No insight content found in response")
|
||||
return ""
|
||||
|
|
|
|||
|
|
@ -1,119 +0,0 @@
|
|||
from typing import List, Dict
|
||||
|
||||
from memoryscope.constants.common_constants import RESULT
|
||||
from memoryscope.core.utils.datetime_handler import DatetimeHandler
|
||||
from memoryscope.core.worker.memory_base_worker import MemoryBaseWorker
|
||||
from memoryscope.enumeration.action_status_enum import ActionStatusEnum
|
||||
from memoryscope.enumeration.memory_type_enum import MemoryTypeEnum
|
||||
from memoryscope.scheme.memory_node import MemoryNode
|
||||
|
||||
|
||||
class UpdateMemoryWorker(MemoryBaseWorker):
|
||||
|
||||
def _parse_params(self, **kwargs):
|
||||
self.method: str = kwargs.get("method", "")
|
||||
self.memory_key: str = kwargs.get("memory_key", "")
|
||||
|
||||
def from_query(self):
|
||||
"""
|
||||
Creates a MemoryNode from the provided query if present in chat_kwargs.
|
||||
|
||||
Returns:
|
||||
List[MemoryNode]: A list containing a single MemoryNode created from the query.
|
||||
"""
|
||||
if "query" not in self.chat_kwargs:
|
||||
return
|
||||
|
||||
query = self.chat_kwargs["query"].strip()
|
||||
if not query:
|
||||
return
|
||||
|
||||
dt_handler = DatetimeHandler()
|
||||
node = MemoryNode(user_name=self.user_name,
|
||||
target_name=self.target_name,
|
||||
content=query,
|
||||
memory_type=MemoryTypeEnum.OBS_CUSTOMIZED.value,
|
||||
action_status=ActionStatusEnum.NEW.value,
|
||||
timestamp=dt_handler.timestamp)
|
||||
return [node]
|
||||
|
||||
def from_memory_key(self):
|
||||
"""
|
||||
Retrieves memories based on the memory key if it exists.
|
||||
|
||||
Returns:
|
||||
List[MemoryNode]: A list of MemoryNode objects retrieved using the memory key.
|
||||
"""
|
||||
if not self.memory_key:
|
||||
return
|
||||
|
||||
return self.memory_manager.get_memories(keys=self.memory_key)
|
||||
|
||||
def delete_all(self):
|
||||
"""
|
||||
Marks all memories for deletion by setting their action_status to 'DELETE'.
|
||||
|
||||
Returns:
|
||||
List[MemoryNode]: A list of all MemoryNode objects marked for deletion.
|
||||
"""
|
||||
nodes: List[MemoryNode] = self.memory_manager.get_memories(keys="all")
|
||||
for node in nodes:
|
||||
node.action_status = ActionStatusEnum.DELETE.value
|
||||
self.logger.info(f"delete_all.size={len(nodes)}")
|
||||
return nodes
|
||||
|
||||
def delete_memory(self):
|
||||
"""
|
||||
Marks specific memories for deletion based on query or memory_id present in chat_kwargs.
|
||||
|
||||
Returns:
|
||||
List[MemoryNode]: A list of MemoryNode objects marked for deletion based on the query or memory_id.
|
||||
"""
|
||||
if "query" in self.chat_kwargs:
|
||||
query = self.chat_kwargs["query"].strip()
|
||||
if not query:
|
||||
return
|
||||
|
||||
i = 0
|
||||
nodes: List[MemoryNode] = self.memory_manager.get_memories(keys="all")
|
||||
for node in nodes:
|
||||
if node.content == query:
|
||||
i += 1
|
||||
node.action_status = ActionStatusEnum.DELETE.value
|
||||
self.logger.info(f"delete_memory.query.size={len(nodes)}")
|
||||
return nodes
|
||||
|
||||
elif "memory_id" in self.chat_kwargs:
|
||||
memory_id = self.chat_kwargs["memory_id"].strip()
|
||||
if not memory_id:
|
||||
return
|
||||
|
||||
i = 0
|
||||
nodes: List[MemoryNode] = self.memory_manager.get_memories(keys="all")
|
||||
for node in nodes:
|
||||
if node.memory_id == memory_id:
|
||||
i += 1
|
||||
node.action_status = ActionStatusEnum.DELETE.value
|
||||
self.logger.info(f"delete_memory.memory_id.size={len(nodes)}")
|
||||
return nodes
|
||||
|
||||
return []
|
||||
|
||||
def _run(self):
|
||||
"""
|
||||
Executes a memory update method provided via the 'method' attribute.
|
||||
|
||||
The method specified by the 'method' attribute is invoked,
|
||||
which updates memories accordingly.
|
||||
"""
|
||||
method = self.method.strip()
|
||||
if not hasattr(self, method):
|
||||
self.logger.info(f"method={method} is missing!")
|
||||
return
|
||||
|
||||
updated_nodes: Dict[str, List[MemoryNode]] = self.memory_manager.update_memories(nodes=getattr(self, method)())
|
||||
line = ["[MEMORY ACTIONS]:"]
|
||||
for action, nodes in updated_nodes.items():
|
||||
for node in nodes:
|
||||
line.append(f"{action} {node.memory_type}: {node.content} ({node.store_status})")
|
||||
self.set_workflow_context(RESULT, "\n".join(line))
|
||||
|
|
@ -439,7 +439,8 @@ class MinerUPDFProcessor:
|
|||
self.logger.error(f"Error occurred while saving files: {e}")
|
||||
raise
|
||||
|
||||
def get_content_statistics(self, content_list: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
@staticmethod
|
||||
def get_content_statistics(content_list: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
"""
|
||||
Generate detailed statistics about the processed content.
|
||||
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ class UpdateVectorStoreOp(BaseLLMOp):
|
|||
logger.info(f"insert insert_node.size={len(insert_nodes)}")
|
||||
|
||||
# Store results in context
|
||||
self.context.update_result = {
|
||||
self.context.response.metadata["update_result"] = {
|
||||
"deleted_count": len(deleted_memory_ids) if deleted_memory_ids else 0,
|
||||
"inserted_count": len(insert_memory_list) if insert_memory_list else 0
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,26 +42,11 @@ class VectorStoreActionOp(BaseLLMOp):
|
|||
path=path,
|
||||
callback_fn=memory_dict_to_node)
|
||||
|
||||
elif action == "update_freq":
|
||||
memory_ids: list = self.context.memory_ids
|
||||
result = self.vector_store.update_freq(workspace_id=workspace_id, node_ids=memory_ids)
|
||||
|
||||
elif action == "update_utility":
|
||||
memory_ids: list = self.context.memory_ids
|
||||
result = self.vector_store.update_utility(workspace_id=workspace_id, node_ids=memory_ids)
|
||||
|
||||
elif action == "utility_based_delete":
|
||||
freq_threshold: int = self.context.freq_threshold
|
||||
utility_threshold: float = self.context.utility_threshold
|
||||
result = self.vector_store.utility_based_delete(workspace_id=workspace_id,
|
||||
freq_threshold=freq_threshold,
|
||||
utility_threshold=utility_threshold)
|
||||
|
||||
else:
|
||||
raise ValueError(f"invalid action={action}")
|
||||
|
||||
# Store results in context
|
||||
if isinstance(result, dict):
|
||||
self.context.action_result = result
|
||||
self.context.response.metadata["action_result"] = result
|
||||
else:
|
||||
self.context.action_result = {"result": str(result)}
|
||||
self.context.response.metadata["action_result"] = {"result": str(result)}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue