diff --git a/agent/memory/manager.py b/agent/memory/manager.py index 00e4c96..bb007fc 100644 --- a/agent/memory/manager.py +++ b/agent/memory/manager.py @@ -408,179 +408,6 @@ class MemoryManager: self._dirty = True return success - def build_memory_guidance(self, lang: str = "zh", include_context: bool = True) -> str: - """ - Build natural memory guidance for agent system prompt - - Following clawdbot's approach: - 1. Load MEMORY.md as bootstrap context (blends into background) - 2. Load daily files on-demand via memory_search tool - 3. Agent should NOT proactively mention memories unless user asks - - Args: - lang: Language for guidance ("en" or "zh") - include_context: Whether to include bootstrap memory context (default: True) - MEMORY.md is loaded as background context (like clawdbot) - Daily files are accessed via memory_search tool - - Returns: - Memory guidance text (and optionally context) for system prompt - """ - today_file = self.flush_manager.get_today_memory_file().name - - if lang == "zh": - guidance = f"""## 记忆系统 - -**背景知识**: 下方包含核心长期记忆,可直接使用。需要查找历史时,用 memory_search 搜索(搜索一次即可,不要重复)。 - -**主动存储**: 遇到以下情况时,主动用 edit/write 工具存储(无需告知用户): -- 用户要求记住的信息、个人偏好、重要决策 -- 对话中产生的重要结论、方案、约定 -- 完成复杂任务后的关键步骤和结果 -- 长期信息 → MEMORY.md(保持精简) -- 当天笔记 → memory/{today_file} - -**使用原则**: 自然使用记忆,就像你本来就知道。不需要生硬地提起或列举记忆,除非用户提到。""" - else: - guidance = f"""## Memory System - -**Background Knowledge**: Core long-term memories below - use directly. For history, use memory_search once (don't repeat). - -**Proactive Storage**: Store memories silently when: -- User asks to remember something, shares preferences or decisions -- Important conclusions, plans, or agreements emerge in conversation -- Complex tasks are completed (record key steps and results) -- Durable info → MEMORY.md (keep concise) -- Daily notes → memory/{today_file} - -**Usage**: Use memories naturally as if you always knew. Don't mention or list unless user explicitly asks.""" - - if include_context: - # Load bootstrap context (MEMORY.md only, like clawdbot) - bootstrap_context = self.load_bootstrap_memories() - if bootstrap_context: - guidance += f"\n\n## Background Context\n\n{bootstrap_context}" - - return guidance - - def load_bootstrap_memories(self, user_id: Optional[str] = None) -> str: - """ - Load bootstrap memory files for session start - - Loads: - 1. MEMORY.md from workspace root (long-term curated memory) - 2. User-specific MEMORY.md if user_id provided - 3. Recent daily memory files (today + yesterday) for continuity - - Returns memory content WITHOUT obvious headers so it blends naturally - into the context as background knowledge. - - Args: - user_id: Optional user ID for user-specific memories - - Returns: - Memory content to inject into system prompt - """ - workspace_dir = self.config.get_workspace() - memory_dir = self.config.get_memory_dir() - - sections = [] - - # 1. Load MEMORY.md from workspace root (long-term curated memory) - memory_file = Path(workspace_dir) / "MEMORY.md" - if memory_file.exists(): - try: - content = memory_file.read_text(encoding='utf-8').strip() - if content: - sections.append(content) - except Exception as e: - from common.log import logger - logger.warning(f"[MemoryManager] Failed to read MEMORY.md: {e}") - - # 2. Load user-specific MEMORY.md if user_id provided - if user_id: - user_memory_dir = memory_dir / "users" / user_id - user_memory_file = user_memory_dir / "MEMORY.md" - if user_memory_file.exists(): - try: - content = user_memory_file.read_text(encoding='utf-8').strip() - if content: - sections.append(content) - except Exception as e: - from common.log import logger - logger.warning(f"[MemoryManager] Failed to read user memory: {e}") - - # 3. Load recent daily memory files (today + yesterday) for context continuity - recent_daily = self._load_recent_daily_memories( - memory_dir, user_id, days=2, max_tokens=2000 - ) - if recent_daily: - sections.append(recent_daily) - - if not sections: - return "" - - return "\n\n".join(sections) - - def _load_recent_daily_memories( - self, - memory_dir: Path, - user_id: Optional[str], - days: int = 2, - max_tokens: int = 2000 - ) -> str: - """ - Load recent daily memory files for bootstrap context. - Loads the most recent N days that have non-empty content. - - Args: - memory_dir: Memory directory path - user_id: Optional user ID - days: Number of recent days to load - max_tokens: Approximate max tokens to include (rough char estimate) - """ - from common.log import logger - - daily_sections = [] - total_chars = 0 - max_chars = max_tokens * 4 # rough token-to-char ratio - - for i in range(days): - date = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d") - - # Check user-specific daily file first, then shared - candidates = [] - if user_id: - candidates.append(memory_dir / "users" / user_id / f"{date}.md") - candidates.append(memory_dir / f"{date}.md") - - for daily_file in candidates: - if not daily_file.exists(): - continue - try: - content = daily_file.read_text(encoding='utf-8').strip() - if not content or len(content) < 30: - continue - - # Truncate if adding this would exceed limit - remaining = max_chars - total_chars - if remaining <= 0: - break - if len(content) > remaining: - content = content[:remaining] + "\n...(truncated)" - - label = "Today" if i == 0 else "Yesterday" if i == 1 else date - daily_sections.append(f"### {label} ({date})\n{content}") - total_chars += len(content) - break # only load one file per date (user-specific takes priority) - except Exception as e: - logger.warning(f"[MemoryManager] Failed to read daily memory {daily_file}: {e}") - - if not daily_sections: - return "" - - return "### Recent Activity\n\n" + "\n\n".join(daily_sections) - def get_status(self) -> Dict[str, Any]: """Get memory status""" stats = self.storage.get_stats()