diff --git a/reme_ai/config/default.yaml b/reme_ai/config/default.yaml index 4462cc7b..a63dcba5 100644 --- a/reme_ai/config/default.yaml +++ b/reme_ai/config/default.yaml @@ -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 diff --git a/reme_ai/config/default_config.yaml b/reme_ai/config/default_config.yaml deleted file mode 100644 index 02ed9dbd..00000000 --- a/reme_ai/config/default_config.yaml +++ /dev/null @@ -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 diff --git a/reme_ai/config/demo_config.yaml b/reme_ai/config/demo_config.yaml deleted file mode 100644 index 2e54eac8..00000000 --- a/reme_ai/config/demo_config.yaml +++ /dev/null @@ -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 \ No newline at end of file diff --git a/reme_ai/retrieve/personal/extract_time_op.py b/reme_ai/retrieve/personal/extract_time_op.py index 3957f47e..ba58bfbb 100644 --- a/reme_ai/retrieve/personal/extract_time_op.py +++ b/reme_ai/retrieve/personal/extract_time_op.py @@ -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 diff --git a/reme_ai/retrieve/personal/fuse_rerank_op.py b/reme_ai/retrieve/personal/fuse_rerank_op.py index fc7353d0..4301a8d1 100644 --- a/reme_ai/retrieve/personal/fuse_rerank_op.py +++ b/reme_ai/retrieve/personal/fuse_rerank_op.py @@ -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 diff --git a/reme_ai/retrieve/personal/print_memory_op.py b/reme_ai/retrieve/personal/print_memory_op.py index 7a83cefc..b7a3f98e 100644 --- a/reme_ai/retrieve/personal/print_memory_op.py +++ b/reme_ai/retrieve/personal/print_memory_op.py @@ -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) diff --git a/reme_ai/retrieve/personal/semantic_rank_op.py b/reme_ai/retrieve/personal/semantic_rank_op.py index 969e7195..837357e8 100644 --- a/reme_ai/retrieve/personal/semantic_rank_op.py +++ b/reme_ai/retrieve/personal/semantic_rank_op.py @@ -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): diff --git a/reme_ai/retrieve/personal/set_query_op.py b/reme_ai/retrieve/personal/set_query_op.py index 937589d4..fb6cde30 100644 --- a/reme_ai/retrieve/personal/set_query_op.py +++ b/reme_ai/retrieve/personal/set_query_op.py @@ -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}") diff --git a/reme_ai/retrieve/task/build_query_op.py b/reme_ai/retrieve/task/build_query_op.py index e043a2eb..0cf80a88 100644 --- a/reme_ai/retrieve/task/build_query_op.py +++ b/reme_ai/retrieve/task/build_query_op.py @@ -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 = [] diff --git a/reme_ai/retrieve/task/rewrite_memory_op.py b/reme_ai/retrieve/task/rewrite_memory_op.py index f50d7ec1..0ca77251 100644 --- a/reme_ai/retrieve/task/rewrite_memory_op.py +++ b/reme_ai/retrieve/task/rewrite_memory_op.py @@ -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 diff --git a/reme_ai/schema/memory.py b/reme_ai/schema/memory.py index 93b12142..baa46b93 100644 --- a/reme_ai/schema/memory.py +++ b/reme_ai/schema/memory.py @@ -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!") diff --git a/reme_ai/summary/personal/__init__.py b/reme_ai/summary/personal/__init__.py index 7d024ec7..235ee091 100644 --- a/reme_ai/summary/personal/__init__.py +++ b/reme_ai/summary/personal/__init__.py @@ -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 \ No newline at end of file diff --git a/reme_ai/summary/personal/contra_repeat_op.py b/reme_ai/summary/personal/contra_repeat_op.py index 087a9fda..f4dc768b 100644 --- a/reme_ai/summary/personal/contra_repeat_op.py +++ b/reme_ai/summary/personal/contra_repeat_op.py @@ -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 diff --git a/reme_ai/summary/personal/get_observation_op.py b/reme_ai/summary/personal/get_observation_op.py index b67f40cb..662011a8 100644 --- a/reme_ai/summary/personal/get_observation_op.py +++ b/reme_ai/summary/personal/get_observation_op.py @@ -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 diff --git a/reme_ai/summary/personal/get_observation_with_time_op.py b/reme_ai/summary/personal/get_observation_with_time_op.py index 1afeaf5d..27c7f615 100644 --- a/reme_ai/summary/personal/get_observation_with_time_op.py +++ b/reme_ai/summary/personal/get_observation_with_time_op.py @@ -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>