第六章：记忆与上下文子系统¶

import time

class MemoryConsolidationTrigger:
    """记忆整合触发器"""

    def __init__(self, config):
        self.time_threshold = config.get("time_threshold", 86400)  # 24小时
        self.session_threshold = config.get("session_threshold", 5)  # 5个会话
        self.last_consolidation_time = time.time()
        self.session_count_since_consolidation = 0

    def should_consolidate(self) -> bool:
        """检查是否应该触发整合"""
        # 触发条件1:时间门槛
        if time.time() - self.last_consolidation_time > self.time_threshold:
            return True

        # 触发条件2:会话计数门槛
        if self.session_count_since_consolidation >= self.session_threshold:
            return True

        # 触发条件3:显式锁定(用户或系统主动触发)
        # 这通过外部API调用实现:POST /consolidate

        return False

    def on_session_end(self):
        """会话结束时调用"""
        self.session_count_since_consolidation += 1

import time
from typing import List, Dict, Any

class MemoryConsolidationPipeline:
    """记忆整合管道"""

    async def consolidate(self, workflow_id: str, session_history: List[Dict[str, Any]]):
        """执行完整的整合流程"""

        # 阶段1:Orient - 分析会话目标和上下文
        orientation = await self.orient_phase(workflow_id, session_history)
        # 结果:{goals, context, decisions, new_patterns}

        # 阶段2:Gather - 从对话历史提取关键信息
        gathered = await self.gather_phase(session_history, orientation)
        # 结果:{facts, preferences, outcomes, patterns}

        # 阶段3:Consolidate - 合并到长期记忆,避免冗余
        consolidated = await self.consolidate_phase(workflow_id, gathered)
        # 结果:更新了MEMORY.md, embedding_index, session_logs

        # 阶段4:Prune - 删除过期或低价值的信息
        pruned = await self.prune_phase(workflow_id, consolidated)
        # 结果:清理过时的会话摘要

        return {
            "orientation": orientation,
            "gathered": gathered,
            "consolidated": consolidated,
            "pruned": pruned
        }

    async def orient_phase(self, workflow_id: str, session_history: List[Message]):
        """阶段1:Orient - 理解会话的核心目标和决策点"""
        prompt = f"""
分析这个工作流的会话记录,提供:
1. 会话的核心目标(1-3个)
2. 关键的决策点和选择(2-3个)
3. 发现的系统级模式或约束(2-3个)
4. 用户展现的新的偏好或工作风格(如有)

会话记录:
{self._format_session_history(session_history)}
"""
        # 调用LLM进行分析
        return await self.llm.analyze(prompt)

    async def gather_phase(self, session_history: List[Message], orientation: dict):
        """阶段2:Gather - 提取结构化事实"""
        facts = {
            "user_preferences": [],
            "task_outcomes": [],
            "technical_discoveries": [],
            "system_constraints": []
        }

        for message in session_history:
            if message.role == "user":
                # 从用户消息提取偏好
                pref = self._extract_preference(message.content)
                if pref:
                    facts["user_preferences"].append(pref)

            elif message.role == "assistant":
                # 从助手消息提取成果
                outcome = self._extract_outcome(message.content)
                if outcome:
                    facts["task_outcomes"].append(outcome)

        return facts

    async def consolidate_phase(self, workflow_id: str, gathered: dict):
        """阶段3:Consolidate - 合并到长期记忆,避免冗余"""

        # 读取现有MEMORY.md
        memory = await self._load_memory_file(workflow_id)

        # 去重:与现有内容比较,只添加新信息
        new_facts = await self._deduplicate(gathered["user_preferences"], memory.get("user_preferences", []))

        # 更新MEMORY.md
        memory["user_preferences"].extend(new_facts)
        memory["last_updated"] = time.time()

        # 更新向量索引(用于语义搜索)
        embeddings = await self._embed_facts(gathered)
        await self._upsert_to_vector_index(workflow_id, embeddings)

        # 追加到日志
        await self._append_to_session_log(workflow_id, gathered)

        return {"updated_memory": memory, "updated_embeddings": embeddings}

    async def prune_phase(self, workflow_id: str, consolidated: dict):
        """阶段4:Prune - 删除过期或低价值的信息"""

        memory = consolidated["updated_memory"]

        # 策略1:删除超过6个月的会话摘要
        cutoff_time = time.time() - 180 * 24 * 3600
        memory["session_summaries"] = [
            s for s in memory.get("session_summaries", [])
            if s.get("timestamp", 0) > cutoff_time
        ]

        # 策略2:删除重复或相似的用户偏好
        memory["user_preferences"] = await self._deduplicate_with_similarity(
            memory["user_preferences"],
            similarity_threshold=0.95
        )

        # 策略3:删除与当前工作无关的信息
        relevant_prefs = await self._filter_by_relevance(
            memory["user_preferences"],
            workflow_id,
            relevance_threshold=0.6
        )
        memory["user_preferences"] = relevant_prefs

        # 保存回磁盘
        await self._save_memory_file(workflow_id, memory)

        return {"pruned_memory": memory}

class SessionMemoryCache:
    """会话记忆缓存(JSON格式)"""

    def __init__(self, max_sessions: int = 10):
        self.max_sessions = max_sessions
        # open() 不会展开 ~，需要 expanduser
        self.cache_file = os.path.expanduser("~/.harness/session_cache.jsonl")

    async def record_session_summary(self, workflow_id: str, session_data: dict):
        """记录一个会话的摘要"""
        summary = {
            "workflow_id": workflow_id,
            "timestamp": time.time(),
            "user_input": session_data.get("user_input", ""),
            "key_outputs": session_data.get("key_outputs", []),
            "decisions_made": session_data.get("decisions", []),
            "errors_encountered": session_data.get("errors", []),
        }

        # 追加到缓存
        with open(self.cache_file, "a") as f:
            f.write(json.dumps(summary) + "\n")

        # 如果超过max_sessions,删除最旧的
        await self._prune_old_sessions()

    async def _prune_old_sessions(self):
        """只保留最近 max_sessions 条会话摘要"""
        with open(self.cache_file, "r") as f:
            lines = f.readlines()
        if len(lines) > self.max_sessions:
            with open(self.cache_file, "w") as f:
                f.writelines(lines[-self.max_sessions:])

    async def get_recent_sessions(self, workflow_id: str, count: int = 5) -> List[dict]:
        """快速获取最近的几个会话"""
        sessions = []
        with open(self.cache_file, "r") as f:
            for line in f:
                session = json.loads(line)
                if session["workflow_id"] == workflow_id:
                    sessions.append(session)

        return sessions[-count:]  # 返回最近的count个

# EMBEDDING_DIM 需与所用 embedding 模型一致：
# text-embedding-3-small=1536, text-embedding-3-large=3072, voyage-3-large=1024

class MemoryVectorIndex:
    """向量索引用于语义搜索"""

    # 关键：维度必须与 embedding 模型匹配
    # 默认 1536 对应 text-embedding-3-small；如使用其他模型需修改
    EMBEDDING_DIM = 1536  # 默认使用 text-embedding-3-small 维度

    def __init__(self, index_path: str, embedding_dim: int = None):
        # 使用轻量级的向量库(如Hnswlib或Faiss)
        # dim 参数必须与所用 embedding 模型的输出维度一致
        dim = embedding_dim or self.EMBEDDING_DIM
        self.index = hnswlib.Index(space='cosine', dim=dim)
        self.index.load_index(index_path)
        self.metadata = {}  # 存储向量到原始数据的映射

    async def search_memory(self, query: str, top_k: int = 5) -> List[dict]:
        """语义搜索记忆"""
        # 将查询嵌入
        query_embedding = await self._embed(query)

        # 搜索最相似的向量
        labels, distances = self.index.knn_query(query_embedding, k=top_k)

        # 返回元数据
        results = [
            {
                "similarity_score": 1 - distances[0][i],  # 归一化
                "memory_content": self.metadata[label],
            }
            for i, label in enumerate(labels[0])
        ]

        return results

async def decide_write_memory(session_context, content):
    """判断是否应该写入记忆"""

    signals = []

    # 检查显式信号
    if "remember" in content.lower() or "记住" in content:
        signals.append(("explicit_user", 0.95))

    # 检查隐式信号
    if is_new_user_preference(content, session_context):
        signals.append(("new_preference", 0.7))

    if is_error_or_lesson_learned(content):
        signals.append(("learning", 0.8))

    if is_milestone_completion(content):
        signals.append(("milestone", 0.85))

    # 综合评分
    max_score = max([score for _, score in signals]) if signals else 0

    if max_score > 0.6:  # 阈值
        return True, dict(signals)
    return False, {}

def keyword_search(query: str, memory_base: str) -> List[str]:
    """在 MEMORY.md 和日志中执行关键词匹配"""
    results = []
    keywords = query.lower().split()

    for memory_file in get_memory_files():
        content = read_file(memory_file)
        matches = [line for line in content.split('\n')
                   if any(kw in line.lower() for kw in keywords)]
        results.extend(matches)

    return results

def semantic_search(query: str, embedding_index) -> List[str]:
    """使用向量相似度查找相关内容"""
    query_embedding = embed(query)

    scores = []
    for memory_id, stored_embedding in embedding_index.items():
        similarity = cosine_similarity(query_embedding, stored_embedding)
        scores.append((memory_id, similarity))

    # 返回相似度最高的 Top-K（按相似度排序，而非默认的按 memory_id 排序）
    return [mid for mid, _ in sorted(scores, key=lambda x: x[1], reverse=True)[:5]]

def hybrid_search(query: str, memory_base: str, embedding_index) -> List[str]:
    """组合关键词和向量搜索结果"""
    kw_results = keyword_search(query, memory_base)
    sem_results = semantic_search(query, embedding_index)

    # 融合:先返回同时匹配两者的结果,再返回单一匹配
    combined = set(kw_results) & set(sem_results)
    rest = (set(kw_results) | set(sem_results)) - combined

    return list(combined) + list(rest)

def write_memory_atomic(memory_id: str, new_content: str,
                        expected_version: int) -> bool:
    """原子式写入,基于版本号"""
    current = read_memory(memory_id)

    if current['version'] != expected_version:
        # 版本冲突,进行 3-way merge（基础版本从版本历史中取写入方所基于的那一版）
        base = read_memory_version(memory_id, expected_version)
        return merge_versions(base['content'], current['content'], new_content)

    new_version = {
        'version': expected_version + 1,
        'last_modified': now(),
        'modified_by': current_session_id(),
        'content': new_content
    }

    return atomic_write(memory_id, new_version)

def merge_versions(base, current, new):
    """3-way merge 用于记忆文件"""
    # 简化示例:按行级别的 diff
    base_lines = base.split('\n')
    curr_lines = current.split('\n')
    new_lines = new.split('\n')

    merged = []
    for i, (b, c, n) in enumerate(zip(base_lines, curr_lines, new_lines)):
        if c == n:  # 无冲突
            merged.append(c)
        elif b == c:  # 只有新版本改动
            merged.append(n)
        elif b == n:  # 只有当前版本改动
            merged.append(c)
        else:  # 双向改动,标记为冲突
            merged.append(f"<<< CONFLICT >>>\n{c}\n---\n{n}\n>>>")

    return '\n'.join(merged)

class MemoryWriter:
    """Agent 可调用的记忆写入工具"""

    async def save_user_profile(self, key: str, value: any) -> bool:
        """保存用户属性"""
        pass

    async def record_feedback(self, action: str, approved: bool,
                              reason: str) -> bool:
        """记录用户反馈"""
        pass

    async def update_project_context(self, updates: dict) -> bool:
        """更新项目信息"""
        pass

    async def add_reference(self, category: str, content: str,
                            tags: List[str]) -> bool:
        """添加参考资料"""
        pass

    async def record_lesson(self, lesson: str, confidence: float) -> bool:
        """记录学到的教训"""
        pass

class ContextRequirement:
    """上下文需求规范"""
    needs_user_profile: bool = False  # 需要用户档案
    needs_project_context: bool = False  # 需要项目信息
    needs_recent_history: bool = False  # 需要最近历史
    needs_references: bool = False  # 需要参考资料
    needs_feedback: bool = False  # 需要反馈记录

def analyze_query(user_message: str) -> ContextRequirement:
    """分析查询内容,确定上下文需求"""
    query_lower = user_message.lower()

    req = ContextRequirement()

    # 启发式规则
    if any(word in query_lower for word in
           ["prefer", "style", "like", "habit"]):
        req.needs_user_profile = True

    if any(word in query_lower for word in
           ["project", "task", "status", "progress"]):
        req.needs_project_context = True

    if any(word in query_lower for word in
           ["previous", "before", "last time", "remember"]):
        req.needs_recent_history = True

    if any(word in query_lower for word in
           ["example", "sample", "pattern", "how to"]):
        req.needs_references = True

    return req

async def search_memory_sources(requirement: ContextRequirement,
                                query: str,
                                memory_manager) -> MemorySearchResult:
    """并行搜索多个记忆源"""
    tasks = []

    if requirement.needs_user_profile:
        tasks.append(memory_manager.search_user_profile(query))

    if requirement.needs_project_context:
        tasks.append(memory_manager.search_project_context(query))

    if requirement.needs_recent_history:
        tasks.append(memory_manager.search_recent_history(query))

    if requirement.needs_references:
        tasks.append(memory_manager.search_references(query))

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return MemorySearchResult(results)

def assemble_context(user_message: str,
                     search_result: MemorySearchResult,
                     system_prompt: str,
                     token_budget: int = 50000) -> str:
    """组装最终的上下文"""

    # 优先级:系统提示 > 用户档案 > 项目信息 > 历史 > 参考
    prioritized_items = [
        ("system", system_prompt, 100),
        ("user_profile", search_result.user_profile, 90),
        ("project", search_result.project_context, 80),
        ("history", search_result.recent_history, 70),
        ("references", search_result.references, 60),
    ]

    # 按优先级排序
    prioritized_items.sort(key=lambda x: x[2], reverse=True)

    assembled = []
    current_tokens = 0

    for section_name, content, _ in prioritized_items:
        if not content:
            continue

        content_tokens = estimate_tokens(content)

        if current_tokens + content_tokens <= token_budget:
            assembled.append(f"## {section_name}\n{content}")
            current_tokens += content_tokens
        else:
            # 容量不足,尝试截断
            remaining = token_budget - current_tokens
            if remaining > 100:  # 最少保留 100 tokens
                truncated = truncate_to_tokens(content, remaining)
                assembled.append(f"## {section_name} (truncated)\n{truncated}")
            break

    return '\n\n'.join(assembled)

class ContextPlugin:
    """上下文插件的基类"""

    async def generate(self, query: str, token_budget: int) -> str:
        """生成该插件贡献的上下文部分"""
        raise NotImplementedError

    @property
    def priority(self) -> int:
        """该插件的优先级(0-100)"""
        raise NotImplementedError

    @property
    def name(self) -> str:
        """插件名称"""
        raise NotImplementedError

class UserProfilePlugin(ContextPlugin):
    """用户档案插件"""

    @property
    def name(self) -> str:
        return "user_profile"

    @property
    def priority(self) -> int:
        return 90

    async def generate(self, query: str, token_budget: int) -> str:
        profile = await self.memory_manager.get_user_profile()
        return self._format_profile(profile, token_budget)

class ProjectContextPlugin(ContextPlugin):
    """项目上下文插件"""

    @property
    def name(self) -> str:
        return "project"

    @property
    def priority(self) -> int:
        return 80

    async def generate(self, query: str, token_budget: int) -> str:
        # 搜索相关项目信息
        relevant_info = await self.search_project_db(query)
        return self._format_project_info(relevant_info, token_budget)

class ContextEngine:
    """上下文生成引擎"""

    def __init__(self):
        self.plugins: Dict[str, ContextPlugin] = {}

    def register_plugin(self, plugin: ContextPlugin):
        """注册上下文插件"""
        self.plugins[plugin.name] = plugin

    async def assemble(self, query: str,
                       token_budget: int = 100000) -> str:
        """根据查询组装上下文"""
        # 按优先级排序插件
        sorted_plugins = sorted(
            self.plugins.values(),
            key=lambda p: p.priority,
            reverse=True
        )

        assembled = []
        remaining_tokens = token_budget

        for plugin in sorted_plugins:
            if remaining_tokens < 500:  # 最小阈值
                break

            try:
                content = await plugin.generate(query, remaining_tokens)
                if content:
                    assembled.append(content)
                    remaining_tokens -= estimate_tokens(content)
            except Exception as e:
                logger.warning(f"Plugin {plugin.name} failed: {e}")

        return '\n\n'.join(assembled)

class ContextCache:
    """上下文缓存,支持失效管理"""

    def __init__(self, ttl_seconds: int = 300):
        self.cache: Dict[str, CacheEntry] = {}
        self.ttl = ttl_seconds

    def get(self, key: str) -> Optional[str]:
        """获取缓存"""
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry.timestamp < self.ttl:
                return entry.value
            else:
                del self.cache[key]
        return None

    def set(self, key: str, value: str, tags: List[str] = None):
        """设置缓存,关联标签用于批量失效"""
        self.cache[key] = CacheEntry(value, time.time(), tags or [])

    def invalidate_by_tag(self, tag: str):
        """按标签批量失效缓存"""
        to_delete = [k for k, v in self.cache.items()
                     if tag in (v.tags or [])]
        for k in to_delete:
            del self.cache[k]

from datetime import datetime

def check_time_gate(last_consolidation: datetime) -> bool:
    """检查距离上次整合是否超过 24 小时"""
    elapsed = datetime.now() - last_consolidation
    return elapsed.total_seconds() > 86400  # 24 hours

def check_session_gate(session_count_since_consolidation: int) -> bool:
    """检查是否已完成 5 个新会话"""
    return session_count_since_consolidation >= 5

def check_explicit_lock(user_signal: str) -> bool:
    """用户明确触发整合(如"保存进度")"""
    return "consolidate" in user_signal.lower() or \
           "save progress" in user_signal.lower()

def should_consolidate(state: ConsolidationState) -> bool:
    """判断是否应该触发 autoDream"""
    time_gate = check_time_gate(state.last_consolidation)
    session_gate = check_session_gate(state.sessions_since_consolidation)
    explicit_lock = check_explicit_lock(state.latest_user_message)

    return time_gate or session_gate or explicit_lock

async def orient_phase(recent_conversations: List[Message],
                       current_project: str) -> OrientResult:
    """确定整合的方向和范围"""

    # 提取最近对话的摘要
    summary = await extract_summary(recent_conversations)

    # 识别关键主题
    topics = await identify_topics(summary)

    # 确定与项目的相关性
    project_relevance = await assess_relevance(summary, current_project)

    return OrientResult(
        summary=summary,
        topics=topics,
        project_relevance=project_relevance,
        scope="deep" if project_relevance > 0.7 else "shallow"
    )

async def gather_phase(conversations: List[Message],
                       orient_result: OrientResult) -> GatherResult:
    """收集需要记忆的关键信息"""

    gathered = {
        'user_preferences': [],  # 用户偏好
        'project_updates': [],   # 项目进度
        'learned_lessons': [],   # 学到的教训
        'decisions': [],         # 做出的决策
        'errors': []            # 发现的错误
    }

    for msg in conversations:
        if orient_result.scope == "deep":
            # 深度收集:细粒度提取
            gathered['user_preferences'].extend(
                extract_user_preferences(msg)
            )
            gathered['project_updates'].extend(
                extract_project_updates(msg)
            )
        else:
            # 浅层收集:仅提取顶级要点
            gathered['decisions'].extend(
                extract_top_level_decisions(msg)
            )

    return GatherResult(gathered=gathered)

async def consolidate_phase(gather_result: GatherResult,
                            claude_md: dict) -> ConsolidateResult:
    """融合新信息到长期记忆"""

    updated_claude_md = copy.deepcopy(claude_md)

    # 更新用户档案
    for pref in gather_result.gathered['user_preferences']:
        existing = find_existing(pref, updated_claude_md['user_profile'])
        if existing:
            # 冲突解决:新信息覆盖(可配置)
            updated_claude_md['user_profile'][existing['key']] = pref
        else:
            updated_claude_md['user_profile'].append(pref)

    # 更新项目进度
    for update in gather_result.gathered['project_updates']:
        category = update['category']
        updated_claude_md['project_context'][category].extend(
            update['items']
        )

    # 添加教训和决策到参考资料
    for lesson in gather_result.gathered['learned_lessons']:
        updated_claude_md['learning'].append({
            'date': datetime.now(),
            'content': lesson,
            'confidence': 0.8
        })

    return ConsolidateResult(updated=updated_claude_md)

async def prune_phase(consolidated: dict,
                      config: PruneConfig) -> dict:
    """清理过期或冗余信息"""

    pruned = copy.deepcopy(consolidated)

    # 删除过期的项目信息（project_context 是“类别 -> 条目列表”的字典，
    # 与 consolidate_phase 的写入结构保持一致）
    cutoff_date = datetime.now() - timedelta(days=config.expiry_days)
    pruned['project_context'] = {
        category: [
            item for item in items
            if item.get('date', datetime.now()) > cutoff_date
        ]
        for category, items in pruned['project_context'].items()
    }

    # 删除置信度过低的学习项
    pruned['learning'] = [
        item for item in pruned['learning']
        if item.get('confidence', 1.0) > config.confidence_threshold
    ]

    # 合并相似的信息
    pruned['user_profile'] = deduplicate_profiles(
        pruned['user_profile']
    )

    # 压缩冗长的文本
    pruned['references'] = compress_references(
        pruned['references'],
        max_lines=config.max_reference_lines
    )

    return pruned

async def autoDream_consolidation(state: ConsolidationState) -> bool:
    """autoDream 整合的完整流程(WAL-风格回滚)"""

    if not should_consolidate(state):
        return False

    # [WAL] 保存原始状态用于故障回滚
    original_claude_md = copy.deepcopy(state.claude_md)
    original_state = {
        "last_consolidation": state.last_consolidation,
        "sessions_since_consolidation": state.sessions_since_consolidation
    }

    try:
        # Phase 1: Orient
        orient_result = await orient_phase(
            state.recent_conversations,
            state.current_project
        )

        # Phase 2: Gather
        gather_result = await gather_phase(
            state.conversations,
            orient_result
        )

        # Phase 3: Consolidate
        consolidate_result = await consolidate_phase(
            gather_result,
            state.claude_md
        )

        # Phase 4: Prune
        pruned_result = await prune_phase(
            consolidate_result.updated,
            config=state.prune_config
        )

        # [WAL] 原子化保存：先写临时文件，再原子重命名
        await save_claude_md_atomic(pruned_result)

        # 状态更新(仅在持久化成功后)
        state.last_consolidation = datetime.now()
        state.sessions_since_consolidation = 0
        state.claude_md = pruned_result
        state.consolidation_failed_count = 0  # 重置失败计数

        logger.info("autoDream consolidation completed successfully")
        return True

    except Exception as e:
        logger.error(f"autoDream consolidation failed: {e}")

        # [WAL] 失败回滚：恢复到上次已知的好状态
        try:
            # 1. 放弃本次持久化,不修改磁盘上的 CLAUDE.md
            # 2. 恢复内存中的状态
            state.claude_md = original_claude_md
            state.last_consolidation = original_state["last_consolidation"]
            state.sessions_since_consolidation = original_state["sessions_since_consolidation"]

            # 3. 记录故障，允许下次重试
            state.consolidation_failed_count += 1
            if state.consolidation_failed_count > 3:
                logger.warning(
                    f"Multiple consolidation failures ({state.consolidation_failed_count}) detected. "
                    "Consider manual review of CLAUDE.md integrity."
                )
        except Exception as rollback_error:
            logger.critical(f"Rollback failed: {rollback_error}. Manual intervention required.")

        return False

def check_memory_flush_trigger(context_utilization: float) -> bool:
    """当上下文占用达到 70% 时触发刷写（本书示例默认值，非 OpenClaw 官方阈值）"""
    return context_utilization >= 0.7

async def flush_memory_to_permanent(daily_log: str,
                                    memory_md: str) -> str:
    """自动刷写:将日志转换为长期记忆"""

    # 步骤 1:提取日志摘要
    summary = await summarize_log(daily_log, max_lines=10)

    # 步骤 2:从摘要中提取学习
    learnings = await extract_learnings(summary)

    # 步骤 3:合并到 MEMORY.md
    updated_memory = memory_md + "\n\n## Recent Learnings\n"
    for learning in learnings:
        updated_memory += f"- {learning}\n"

    # 步骤 4:版本控制
    updated_memory = add_metadata(
        updated_memory,
        last_updated=datetime.now(),
        version=increment_version(memory_md)
    )

    return updated_memory

class EmbeddingIndex:
    """记忆的向量索引"""

    async def add_entry(self, memory_id: str, content: str):
        """添加新的记忆条目到索引"""
        embedding = await embed_text(content)
        self.index[memory_id] = {
            'embedding': embedding,
            'content': content,
            'timestamp': datetime.now()
        }

    async def search(self, query: str, top_k: int = 5) -> List[str]:
        """向量相似度搜索"""
        query_embedding = await embed_text(query)
        scores = []

        for mid, entry in self.index.items():
            similarity = cosine_similarity(
                query_embedding,
                entry['embedding']
            )
            scores.append((mid, similarity, entry['content']))

        # 按相似度排序,返回 Top-K
        return [content for _, _, content in
                sorted(scores, key=lambda x: x[1], reverse=True)[:top_k]]

    async def rebuild_incremental(self, new_entries: List[dict]):
        """增量更新索引"""
        for entry in new_entries:
            await self.add_entry(entry['id'], entry['content'])

class KeywordIndex:
    """基于关键词的反向索引"""

    def __init__(self):
        self.index: Dict[str, List[str]] = {}  # keyword -> memory_ids

    def add_entry(self, memory_id: str, content: str):
        """提取关键词并建立索引"""
        keywords = extract_keywords(content)  # 使用 TF-IDF 或更简单的方法

        for keyword in keywords:
            if keyword not in self.index:
                self.index[keyword] = []
            self.index[keyword].append(memory_id)

    def search(self, query: str) -> Set[str]:
        """关键词搜索返回所有匹配的记忆ID"""
        query_keywords = extract_keywords(query)
        matching_ids = set()

        for keyword in query_keywords:
            matching_ids.update(self.index.get(keyword, []))

        return matching_ids

async def cleanup_old_memories(memory_store, embedding_index, keyword_index,
                              retention_days: int = 90):
    """定期清理过期记忆"""
    cutoff = datetime.now() - timedelta(days=retention_days)

    for memory_id in memory_store.list_all():
        metadata = memory_store.get_metadata(memory_id)

        if metadata['last_accessed'] < cutoff:
            logger.info(f"Removing unused memory: {memory_id}")
            memory_store.delete(memory_id)

            # 同步更新索引（仅对被删除的记忆执行）
            await embedding_index.remove(memory_id)
            keyword_index.remove(memory_id)

class ConsolidationMetrics:
    """记忆整合的性能指标"""

    def __init__(self):
        self.consolidation_latency = []  # 整合耗时
        self.context_compression_ratio = []  # 压缩率
        self.memory_size_growth = []  # 记忆库增长
        self.search_latency = []  # 搜索响应时间
        self.false_negatives = []  # 搜索未命中率

    async def record_consolidation(self,
                                   before_size: int,
                                   after_size: int,
                                   duration_ms: float):
        """记录整合操作"""
        ratio = after_size / before_size if before_size > 0 else 1.0
        self.consolidation_latency.append(duration_ms)
        self.context_compression_ratio.append(ratio)

    async def analyze(self) -> ConsolidationReport:
        """生成整合性能报告"""
        return ConsolidationReport(
            avg_latency=statistics.mean(self.consolidation_latency),
            avg_compression=statistics.mean(self.context_compression_ratio),
            p95_latency=statistics.quantiles(
                self.consolidation_latency,
                n=20
            )[18],  # 95th percentile
        )

from datetime import datetime
from typing import List, Optional

class MemoryEntry:
    """包含 5 个维度的记忆条目"""
    def __init__(self, memory_id: str, content: str,
                 memory_type: str = "episodic",     # 类型划分
                 tags: List[str] = None,
                 confidence: float = 1.0,           # 置信度
                 expiry: Optional[datetime] = None):  # 过期时间
        self.id = memory_id
        self.content = content
        self.type = memory_type
        self.tags = tags or []
        self.confidence = confidence
        self.expiry = expiry
        # 自动跟踪时间戳和版本
        self.created_at = datetime.now()
        self.last_modified = datetime.now()
        self.version = 1

async def analyze_query(self, user_message: str) -> MemoryRequirement:
    """启发式规则:输入 -> 需求"""
    req = MemoryRequirement()
    msg_lower = user_message.lower()

    # 关键词触发
    if any(w in msg_lower for w in ['prefer', 'style', 'like']):
        req.needs_user_profile = True
    if any(w in msg_lower for w in ['project', 'task', 'status']):
        req.needs_project_context = True
    # ... 其他规则

    # 默认回退:当无法识别时加载用户档案
    if not any([...]):
        req.needs_user_profile = True
    return req

async def assemble(self, user_message: str) -> str:
    requirement = await self.analyze_query(user_message)

    # 根据需求并行加载多个记忆源
    tasks = []
    if requirement.needs_user_profile:
        tasks.append(('user_profile', self._gather_user_profile()))
    if requirement.needs_project_context:
        tasks.append(('project', self._gather_project_context()))
    # ... 其他需求

    # 并行执行,节省 I/O 等待
    results = await asyncio.gather(*[task[1] for task in tasks])

    # 按优先级排序,限制总大小
    # 保证组装结果不超过 token_budget (如 50k)

def should_consolidate(self) -> bool:
    # 时间门:距上次整合已超 24 小时
    time_gate = (datetime.now() - self.state.last_consolidation
                 > timedelta(hours=24))
    # 会话门:至少完成 5 次对话
    session_gate = self.state.sessions_since_consolidation >= 5
    # 显式门:由外部调用者手动触发(如用户主动要求)
    return time_gate or session_gate

async def orient_phase(self, recent_messages: List[str]) -> Dict:
    """识别关键主题"""
    topics = set()
    for msg in recent_messages[-5:]:
        if any(w in msg.lower() for w in ['bug', 'error']):
            topics.add('issues')
        if any(w in msg.lower() for w in ['feature', 'new']):
            topics.add('features')
    return {'topics': list(topics)}

async def gather_phase(self, messages: List[str], orient_result: Dict) -> Dict[str, List[str]]:
    """从消息中抽取不同维度的信息"""
    gathered = {
        'user_preferences': [],      # 用户习惯
        'project_updates': [],       # 进度变化
        'learned_lessons': [],       # 知识积累
        'decisions': []              # 重要决策
    }
    # 按关键词提取相关句子
    return gathered

async def consolidate_phase(self, gathered: Dict) -> bool:
    """写入对应类型的存储"""
    if gathered['user_preferences']:
        entry = MemoryEntry(..., memory_type='user')
        await self.memory_store.save(entry)
    # ...
    return True

async def prune_phase(self) -> int:
    """清理两类项:过期项 + 低置信度项"""
    removed = await self.memory_store.cleanup_expired()
    # 清理 confidence < 0.3 且超过 30 天未更新的项
    low_confidence_count = await self.memory_store.cleanup_low_confidence(
        threshold=0.3, stale_days=30
    )
    return removed + low_confidence_count

# 初始化三个核心组件
memory_store = MemoryStore("/path/to/memory")
assembler = ContextAssembler(memory_store, token_budget=50000)
consolidation = ConsolidationEngine(memory_store)

# 在每次对话前:组装上下文
context = await assembler.assemble(user_message)

# 在后台:定期整合
if consolidation.should_consolidate():
    success = await consolidation.consolidate(recent_messages)

class MemoryMetrics:
    memory_size_bytes: int        # 记忆库总大小
    consolidation_latency_ms: float  # 整合耗时
    context_assembly_latency_ms: float  # 上下文组装耗时
    search_hit_rate: float        # 搜索命中率
    cache_hit_rate: float         # 缓存命中率
    average_confidence: float     # 记忆平均置信度
    expired_entries_per_day: int  # 每日过期项数