Agent 记忆与规划交叉：记忆驱动决策、规划反写经验的双向数据流

作者：yumking | 2026 年 10 月 4 日 | 技术标签：记忆×规划 / 经验驱动规划 / 执行反写记忆 / 交叉架构

摘要

第 7/39 篇把规划策略讲透了——ReAct / Plan-and-Execute / Tree-of-Thought / Reflexion 四种机制 internals + 混合策略。第 26 篇做了记忆调度——怎么把记忆塞进有限窗口。第 32 篇打开了记忆存储黑盒——五层记忆模型 + 混合存储 + 遗忘衰减。但四篇有一个共同的盲区：记忆和规划是两个独立系统，它们之间的双向数据流没人设计。规划前该检索哪些记忆？规划执行后该往记忆写什么？记忆过时了怎么不让它污染规划？规划失败了经验怎么存才能让下次规划更好？这些交叉问题不解决，记忆和规划各自优化到极致也只是一堆"孤岛精品"。本文拆解记忆与规划的双向数据流：记忆→规划（经验记忆指导规划方向、情景记忆提供成功先例、语义记忆注入领域约束、程序记忆复用规划模板）、规划→记忆（成功路径存为情景记忆、失败信号存为反思经验、规划模式存为程序记忆、执行轨迹存为语义事实）、交叉闭环架构（检索-规划-执行-写入四阶段 + 一致性保证）、记忆类型×规划阶段的映射矩阵。实测将长程任务首次成功率从 91%（第 39 篇混合策略）提至 95%、规划首步方向错误率从 28% 降至 6%、重复犯错率从 3% 降至 0.4%、冷启动任务成功率从 54% 提至 79%（经验记忆跨任务迁移）。

一、第 7/26/32/39 篇之后：各自讲透了，交叉呢

1.1 四篇的接力与共同盲区

第 7 篇（规划概述）→ ReAct 近视 + Plan-and-Execute + 规划深度自适应，41%→83%
第 39 篇（规划深潜）→ 四种机制 internals + 混合策略，83%→91%
  → 两篇把"怎么规划"讲透了
  → 但规划时用的"经验"从哪来？规划后"经验"往哪去？黑盒

第 26 篇（上下文工程）→ 记忆调度：短期工作记忆 × 长期经验记忆，塞进有限窗口
  → 讲了"记忆怎么进窗口"
  → 但记忆进窗口后怎么影响规划决策？没展开

第 32 篇（记忆架构）→ 五层记忆模型 + 混合存储 + 遗忘衰减 + 一致性
  → 讲了"记忆怎么存、怎么查、怎么衰减"
  → 但记忆的"写入时机"和"读取时机"由谁决定？规划器？执行器？没讲

四篇形成一个十字盲区：

            记忆侧（第26/32篇）
                │
    调度/存储讲透了 │ 但"何时读/何时写"没讲
                │
  ──────────────┼──────────────
                │
  但"记忆怎么   │ 规划侧（第7/39篇）
  驱动规划"没讲  │
                │ 策略/机制讲透了

1.2 一个翻车案例：记忆和规划各跑各的

场景：Agent 处理"部署一个 Python 微服务到 K8s"

团队 A 的实现（记忆和规划各跑各的）：
  ① 规划器（Plan-and-Execute）生成规划：
     [建 Dockerfile → 建 K8s manifest → kubectl apply → 验证]
     → 规划器完全没查记忆，从零规划

  ② 执行第 1 步"建 Dockerfile"：
     → Agent 写了一个 Dockerfile，用了 python:3.9 基础镜像
     → 但记忆库里有一条情景记忆：
       "上次用 python:3.9 部署到 K8s，glibc 版本不兼容导致 segfault，
        改用 python:3.11-slim 解决"
     → 这条记忆没被检索（因为规划器没查记忆）
     → Agent 重复踩坑，segfault，调试 2 小时

  ③ 执行完，结果存哪？
     → 规划器不知道要往记忆写什么
     → 执行器也不知道
     → 这次的经验（python:3.11-slim 才行）没存进记忆
     → 下次同类任务，Agent 又从零开始，又踩 segfault

根因：记忆系统和规划系统各跑各的，没有双向数据流。
  - 记忆→规划断了：规划不从记忆学，每次从零开始
  - 规划→记忆断了：执行经验不回写，记忆不增长

1.3 交叉设计要回答的四个问题

问题一：规划前该检索哪些记忆？
  → 不是全查（太贵），不是不查（重复犯错），查什么、查多少？

问题二：记忆怎么影响规划决策？
  → 检索到的记忆怎么注入规划器？改规划方向？改工具选择？改参数？

问题三：规划执行后该往记忆写什么？
  → 不是全写（噪音爆炸），不是不写（记忆不增长），写什么、什么时候写？

问题四：记忆过时了怎么不让它污染规划？
  → 上次的"正确做法"可能已经过时（API 变了/环境变了）
  → 怎么检测过时？怎么让规划器不信过时记忆？

二、记忆与规划的双向数据流

2.1 总览：一个闭环

    ┌──────────────────────────────────────────┐
    │              记忆系统（第32篇）            │
    │  情景记忆  语义记忆  程序记忆  经验记忆     │
    └──────┬───────────────────────────┬───────┘
           │ ① 检索                    │ ④ 写入
           ▼                           ▲
    ┌──────────────┐           ┌──────────────┐
    │  规划器       │ ② 注入    │  执行器       │
    │  (第39篇)    │──────────▶│              │
    │              │           │  ③ 执行+收集  │
    └──────────────┘           └──────────────┘
         ▲                           │
         │ ⑤ 重规划触发              │
         └───────────────────────────┘

五个箭头是交叉设计的五个接口：

2.2 与第 26/32/39 篇的接口

第 32 篇（记忆架构）
  → 提供 retrieve(layer, query) 和 write(layer, item) 两个原语
  → 本文决定"什么时候调 retrieve、什么时候调 write、查哪层、写哪层"

第 26 篇（上下文工程）
  → 提供"记忆怎么塞进窗口"的预算分配
  → 本文决定"塞进窗口的记忆怎么影响规划"（第 26 篇管装填，本文管使用）

第 39 篇（规划策略）
  → 提供 ReAct/Plan-Execute/ToT/Reflexion 四种规划机制
  → 本文给这些机制加上"记忆输入"和"记忆输出"两个接口

三、记忆→规划：记忆如何驱动规划决策

3.1 四种记忆各怎么驱动规划

接第 32 篇五层记忆模型，其中四层对规划有直接驱动作用：

from dataclasses import dataclass
from typing import Optional

@dataclass
class PlanningContext:
    episodic_hints: list[dict]    # 情景记忆：同类任务的成功路径
    semantic_facts: list[dict]    # 语义记忆：领域知识和约束
    procedural_templates: list    # 程序记忆：可复用的规划模板
    experiential_lessons: list    # 经验记忆：踩过的坑和教训
    task_profile: dict            # 当前任务特征


async def retrieve_for_planning(
    task: str,
    memory_store,
    top_k: int = 5,
) -> PlanningContext:
    """规划前检索相关记忆——不是全查，按任务特征路由到不同层"""

    # 情景记忆：查同类任务的历史执行记录
    episodic = await memory_store.retrieve(
        layer="episodic",
        query=task,
        filter={"task_type": classify_task(task)},
        top_k=top_k,
    )

    # 语义记忆：查任务涉及的领域知识
    entities = extract_entities(task)
    semantic = await memory_store.retrieve(
        layer="semantic",
        query=" ".join(entities),
        top_k=top_k,
    )

    # 程序记忆：查是否有可复用的规划模板
    procedural = await memory_store.retrieve(
        layer="procedural",
        query=task,
        filter={"type": "plan_template"},
        top_k=2,
    )

    # 经验记忆：查同类任务踩过的坑
    experiential = await memory_store.retrieve(
        layer="experiential",
        query=task,
        filter={"type": "lesson", "recency": "recent"},
        top_k=top_k,
    )

    return PlanningContext(
        episodic_hints=episodic,
        semantic_facts=semantic,
        procedural_templates=procedural,
        experiential_lessons=experiential,
        task_profile=analyze_task(task),
    )

3.2 情景记忆：提供成功先例

情景记忆存的是"过去完整执行过的任务轨迹"。规划器拿到同类任务的历史轨迹，可以复用或微调而非从零规划。

async def plan_with_episodic(
    task: str,
    episodic_hints: list[dict],
    planner,
) -> list:
    """用情景记忆驱动规划"""

    if not episodic_hints:
        return await planner.plan_from_scratch(task)

    # 找最相似的历史轨迹
    best_match = max(episodic_hints, key=lambda h: h["similarity"])

    if best_match["similarity"] > 0.85:
        # 高相似度：直接复用历史规划，微调差异部分
        historical_plan = best_match["plan"]
        diff = diff_tasks(best_match["task"], task)
        return adapt_plan(historical_plan, diff)
    elif best_match["similarity"] > 0.6:
        # 中相似度：用历史规划做参考，重新规划
        return await planner.plan_with_reference(task, reference=best_match)
    else:
        # 低相似度：从零规划，但历史轨迹作为背景知识
        return await planner.plan_from_scratch(task, background=episodic_hints)

效果：高相似度任务规划时间从 3.2s 降至 0.4s（复用而非重算），规划质量与历史最佳持平。

3.3 语义记忆：注入领域约束

语义记忆存的是"事实知识"——API 用法、系统约束、领域规则。规划器不知道这些就会生成语法正确但语义错误的规划。

async def inject_semantic_constraints(
    plan: list,
    semantic_facts: list[dict],
) -> list:
    """用语义记忆修正规划中的约束违反"""

    for step in plan:
        for fact in semantic_facts:
            if violates_constraint(step, fact):
                step = apply_constraint(step, fact)
                step["constraint_source"] = fact["id"]

    return plan

例子：

规划器生成：  step = "kubectl apply -f manifest.yaml（用 v1beta1 API）"
语义记忆有：  "K8s 1.22+ 已废弃 v1beta1，必须用 v1"
修正后：      step = "kubectl apply -f manifest.yaml（用 v1 API）"

没有语义记忆注入，规划器会生成"看起来对但实际跑不通"的步骤。

3.4 程序记忆：复用规划模板

程序记忆存的是"规划模式"——某类任务的标准规划骨架。规划器拿到模板只需填空而非从头设计流程。

async def plan_with_template(
    task: str,
    procedural_templates: list,
    planner,
) -> list:
    """用程序记忆中的规划模板加速规划"""

    for template in procedural_templates:
        if template_matches(task, template):
            # 模板匹配：填充任务特定参数到模板骨架
            return fill_template(template, task)

    # 无匹配模板：从零规划，但规划结果可存为新模板
    plan = await planner.plan_from_scratch(task)
    return plan

例子：

程序记忆中有模板："部署微服务到 K8s"的标准 5 步骨架
  [建 Dockerfile → 建 manifest → 建 service → 建 ingress → 验证]

新任务："部署一个 FastAPI 微服务到 K8s"
  → 模板匹配 → 填空：Dockerfile 用 fastapi 基础、manifest 加 liveness probe
  → 规划从 0.4s 完成（vs 从零规划 3.2s）

3.5 经验记忆：避免重复踩坑

经验记忆存的是"教训"——第 9 篇经验库的沉淀。规划器拿到教训可以在规划阶段就绕开已知的坑，而非执行阶段才发现。

async def apply_experiential_lessons(
    plan: list,
    experiential_lessons: list[dict],
) -> list:
    """用经验记忆在规划阶段预避坑"""

    for step in plan:
        relevant_lessons = [
            l for l in experiential_lessons
            if lesson_applies(l, step)
        ]
        for lesson in relevant_lessons:
            if lesson["severity"] == "critical":
                # 严重教训：直接修改步骤
                step = apply_lesson(step, lesson)
            elif lesson["severity"] == "warning":
                # 警告教训：标注风险，执行时注意
                step["warnings"].append(lesson["description"])

    return plan

例子：

经验记忆有教训："python:3.9 基础镜像在 K8s 上 glibc 不兼容，用 3.11-slim"
规划器生成：  step = "FROM python:3.9 ..."
应用教训后：  step = "FROM python:3.11-slim ..."
              → 规划阶段就绕开了 segfault 坑

这是记忆→规划最核心的价值：第 39 篇的 Reflexion 是"执行后反思纠错"，而经验记忆驱动是"规划前预避"——纠错不如避错。

3.6 四种记忆的规划驱动对比

四、规划→记忆：规划执行如何反写记忆

4.1 什么时候写入

不是每步都写（噪音爆炸），不是只写最终结果（丢中间过程）。按事件写入：

@dataclass
class MemoryWriteTrigger:
    trigger_type: str    # success / failure / deviation / pattern
    write_layer: str     # episodic / semantic / procedural / experiential
    write_content: dict


async def evaluate_write_triggers(
    task: str,
    plan: list,
    execution_trace: list,
) -> list[MemoryWriteTrigger]:
    """评估执行轨迹，决定往记忆写什么"""

    triggers = []

    # 触发 1：任务成功完成 → 写情景记忆（完整成功路径）
    if all(s["status"] == "success" for s in execution_trace):
        triggers.append(MemoryWriteTrigger(
            trigger_type="success",
            write_layer="episodic",
            write_content={
                "task": task,
                "plan": plan,
                "trace": execution_trace,
                "outcome": "success",
            },
        ))

    # 触发 2：某步失败后纠错成功 → 写经验记忆（教训）
    for step in execution_trace:
        if step["status"] == "failed" and step.get("recovered"):
            triggers.append(MemoryWriteTrigger(
                trigger_type="failure_recovery",
                write_layer="experiential",
                write_content={
                    "task_type": classify_task(task),
                    "failure_point": step["description"],
                    "root_cause": step["error"],
                    "fix": step["recovery_action"],
                    "lesson": f"{step['error']} → {step['recovery_action']}",
                },
            ))

    # 触发 3：执行偏离规划 → 写经验记忆（偏差教训）
    deviations = detect_deviations(plan, execution_trace)
    for dev in deviations:
        triggers.append(MemoryWriteTrigger(
            trigger_type="deviation",
            write_layer="experiential",
            write_content={
                "task_type": classify_task(task),
                "planned": dev["planned_step"],
                "actual": dev["actual_step"],
                "reason": dev["reason"],
            },
        ))

    # 触发 4：规划模式重复出现 → 写程序记忆（规划模板）
    plan_pattern = extract_plan_pattern(plan)
    if await is_novel_pattern(plan_pattern):
        triggers.append(MemoryWriteTrigger(
            trigger_type="pattern",
            write_layer="procedural",
            write_content={
                "task_type": classify_task(task),
                "template": plan_pattern,
                "sample_count": 1,
            },
        ))

    # 触发 5：发现新事实 → 写语义记忆
    new_facts = extract_new_facts(execution_trace)
    for fact in new_facts:
        triggers.append(MemoryWriteTrigger(
            trigger_type="fact_discovery",
            write_layer="semantic",
            write_content=fact,
        ))

    return triggers

4.2 写入什么：四种记忆各写什么

4.3 写入的质量控制

不是所有触发都写——写入前过质量关，防止噪音污染记忆：

async def controlled_write(
    trigger: MemoryWriteTrigger,
    memory_store,
) -> bool:
    """写入前质量控制"""

    content = trigger.write_content

    # 质量关 1：去重——已有高度相似记忆则不写
    existing = await memory_store.retrieve(
        layer=trigger.write_layer,
        query=serialize(content),
        top_k=1,
    )
    if existing and existing[0]["similarity"] > 0.92:
        return False  # 已有几乎相同的记忆，不写

    # 质量关 2：置信度——失败教训需确认根因
    if trigger.trigger_type == "failure_recovery":
        if content.get("root_cause_confidence", 0) < 0.7:
            return False  # 根因不确定，不写（避免错误教训）

    # 质量关 3：时效——事实知识需验证不过时
    if trigger.trigger_type == "fact_discovery":
        if not await verify_fact_freshness(content):
            return False  # 事实可能已过时，不写

    # 质量关 4：频次——规划模板需多次出现才沉淀
    if trigger.trigger_type == "pattern":
        count = await count_similar_patterns(content["template"])
        if count < 3:
            return False  # 只出现 1-2 次可能是偶然，不沉淀为模板

    await memory_store.write(trigger.write_layer, content)
    return True

效果：写入通过率约 34%（100 次执行触发中 34 次通过质量关写入），记忆库噪音率从 41% 降至 6%。

五、交叉闭环架构

5.1 完整的检索-规划-执行-写入闭环

async def memory_driven_agent_loop(
    task: str,
    memory_store,
    planner,
    executor,
    max_iterations: int = 10,
) -> dict:
    """记忆驱动的 Agent 主循环——记忆与规划的双向闭环"""

    # ① 检索：规划前查相关记忆
    planning_ctx = await retrieve_for_planning(task, memory_store)

    # ② 规划：用记忆驱动规划
    if planning_ctx.procedural_templates:
        plan = await plan_with_template(task, planning_ctx.procedural_templates, planner)
    elif planning_ctx.episodic_hints:
        plan = await plan_with_episodic(task, planning_ctx.episodic_hints, planner)
    else:
        plan = await planner.plan_from_scratch(task)

    # 用语义记忆修正约束
    plan = await inject_semantic_constraints(plan, planning_ctx.semantic_facts)
    # 用经验记忆预避坑
    plan = await apply_experiential_lessons(plan, planning_ctx.experiential_lessons)

    # ③ 执行 + ④ 写入
    execution_trace = []
    for i, step in enumerate(plan):
        result = await executor.execute_step(step)
        execution_trace.append(result)

        if result["status"] == "failed":
            # ⑤ 重规划：失败时用反馈重规划
            deviation = analyze_failure(step, result)
            plan = await replan_with_feedback(
                task, plan[:i], deviation, planning_ctx, planner
            )
            break

    # ④ 写入：执行完评估写入触发
    triggers = await evaluate_write_triggers(task, plan, execution_trace)
    for trigger in triggers:
        await controlled_write(trigger, memory_store)

    return {
        "plan": plan,
        "trace": execution_trace,
        "writes": len(triggers),
    }

5.2 检索时机：不是只查一次

记忆检索不是"规划前查一次就完了"——执行中也可能需要补充检索：

async def execute_step_with_memory(
    step: dict,
    memory_store,
    executor,
) -> dict:
    """执行单步，按需补充检索记忆"""

    # 执行前：查这一步相关的经验教训
    step_lessons = await memory_store.retrieve(
        layer="experiential",
        query=step["description"],
        filter={"applies_to": step["tool"]},
        top_k=3,
    )
    if step_lessons:
        step = apply_lessons_to_step(step, step_lessons)

    # 执行
    result = await executor.execute_step(step)

    # 执行后：如果结果意外，查记忆找解释
    if result["status"] == "unexpected":
        explanations = await memory_store.retrieve(
            layer="semantic",
            query=f"why {result['output']}",
            top_k=3,
        )
        result["explanations"] = explanations

    return result

5.3 检索的预算控制

接第 26 篇上下文预算——记忆检索结果也要占窗口，不能无限查：

async def budgeted_retrieve(
    task: str,
    memory_store,
    token_budget: int = 2000,
) -> PlanningContext:
    """在 token 预算内检索记忆"""

    # 各层检索的 token 成本估算
    layer_budgets = {
        "episodic": int(token_budget * 0.35),    # 情景最贵（完整轨迹）
        "semantic": int(token_budget * 0.25),    # 语义中等（事实条目）
        "procedural": int(token_budget * 0.15),  # 程序最省（模板骨架）
        "experiential": int(token_budget * 0.25),# 经验中等（教训条目）
    }

    results = {}
    for layer, budget in layer_budgets.items():
        # 按 token 预算决定 top_k
        avg_token_per_item = estimate_avg_tokens(layer)
        top_k = max(1, budget // avg_token_per_item)
        results[layer] = await memory_store.retrieve(
            layer=layer, query=task, top_k=top_k,
        )

    return PlanningContext(**results)

效果：记忆检索占用窗口从 3500 token（无预算）降至 2000 token（有预算），规划质量不受影响（2000 token 足以覆盖高价值记忆）。

六、记忆类型×规划阶段的映射矩阵

6.1 哪种记忆在哪个规划阶段起作用

6.2 映射矩阵的工程含义

情景记忆 → 主要在"规划生成"和"执行后写入"
  → 读少写多（一次检索复用，一次执行写入）
  → 存储侧重完整轨迹，检索侧重高相似度

语义记忆 → 主要在"规划校验"和"执行中查解释"
  → 读多写少（每次校验都查，很少发现新事实）
  → 存储侧重事实准确性，检索侧重精确匹配

程序记忆 → 主要在"规划生成"和"执行后沉淀"
  → 读少写更少（模板匹配才读，多次出现才写）
  → 存储侧重模板泛化性，检索侧重任务类型匹配

经验记忆 → 全程参与（规划生成预避/校验标注/执行中查/写入/重规划）
  → 读写均频繁
  → 存储侧重教训质量，检索侧重任务类型+工具匹配
  → 接第 32 篇遗忘衰减：经验记忆衰减最快（环境变化导致教训过时）

6.3 衰减差异化

接第 32 篇 Ebbinghaus 遗忘曲线，不同记忆类型的衰减策略不同：

from math import exp

def memory_retention(layer: str, age_days: float, access_count: int) -> float:
    """不同记忆类型的衰减曲线"""

    configs = {
        # 情景记忆：中速衰减 + 频次加权（常复用的不易忘）
        "episodic": {"stability": 30, "decay": 0.5, "freq_boost": 0.1},
        # 语义记忆：极慢衰减（事实知识稳定）
        "semantic": {"stability": 365, "decay": 0.1, "freq_boost": 0.02},
        # 程序记忆：慢衰减（规划模板一旦沉淀很少变）
        "procedural": {"stability": 180, "decay": 0.15, "freq_boost": 0.05},
        # 经验记忆：快衰减（环境变化导致教训过时）
        "experiential": {"stability": 14, "decay": 0.8, "freq_boost": 0.2},
    }

    cfg = configs[layer]
    base_retention = exp(-age_days / cfg["stability"])
    freq_factor = 1 + cfg["freq_boost"] * min(access_count, 20)
    return min(1.0, base_retention * freq_factor)

关键：经验记忆衰减最快（stability=14 天）——因为环境变化（API 更新/系统升级）会让"上次踩的坑"不再适用。过快的衰减可能导致"又踩同一个坑"，所以用 freq_boost 补偿：反复触发的教训衰减更慢（说明这个坑一直存在）。

七、一致性挑战与解决

7.1 挑战一：过时记忆污染规划

情景记忆有："部署用 Docker compose"（6 个月前存）
当前环境：已迁移到 K8s，Docker compose 不再适用
→ 规划器检索到这条记忆 → 生成 Docker compose 规划 → 执行失败

解决：时效校验 + 置信度衰减

async def retrieve_with_freshness(
    layer: str,
    query: str,
    memory_store,
    freshness_threshold: float = 0.3,
) -> list:
    """检索时过滤过时记忆"""

    results = await memory_store.retrieve(layer=layer, query=query, top_k=10)

    fresh_results = []
    for r in results:
        retention = memory_retention(
            layer, age_days=r["age_days"], access_count=r["access_count"]
        )
        if retention < freshness_threshold:
            # 衰减严重：标注为"可能过时"，不直接用
            r["stale_warning"] = True
            r["retention"] = retention
            # 降权但不删除——可能还有参考价值
            r["similarity"] *= retention
        fresh_results.append(r)

    # 按衰减后相似度重排
    fresh_results.sort(key=lambda x: x["similarity"], reverse=True)
    return fresh_results

7.2 挑战二：记忆冲突

情景记忆 A："用方法 X 部署成功"（3 个月前）
情景记忆 B："用方法 X 部署失败，改用方法 Y"（1 周前）
→ 规划器检索到两条矛盾记忆 → 不知该用 X 还是 Y

解决：时间优先 + 冲突检测

async def resolve_memory_conflicts(memories: list) -> list:
    """检测并解决记忆冲突——时间新的优先"""

    conflicts = detect_conflicts(memories)
    for conflict in conflicts:
        # 同一任务类型，不同结论 → 取时间新的
        newest = max(conflict["items"], key=lambda x: x["timestamp"])
        oldest = min(conflict["items"], key=lambda x: x["timestamp"])

        # 旧的降级为"历史参考"，新的为"当前最佳"
        newest["confidence"] = 1.0
        oldest["confidence"] = 0.3
        oldest["superseded_by"] = newest["id"]

    return memories

7.3 挑战三：写入与检索的一致性

Agent A 写入经验："python:3.9 有 segfault，用 3.11-slim"
Agent B 同时在规划同类任务，检索时还没查到这条经验
→ Agent B 重复踩坑

解决：接第 32 篇乐观锁 + 最终一致性

async def consistent_write_and_notify(
    trigger: MemoryWriteTrigger,
    memory_store,
    event_bus,
) -> bool:
    """写入记忆并通知其他 Agent"""

    # 乐观锁写入
    success = await controlled_write(trigger, memory_store)
    if success:
        # 通知其他 Agent 记忆已更新（最终一致性）
        await event_bus.publish(
            topic="memory_updated",
            message={
                "layer": trigger.write_layer,
                "content": trigger.write_content,
                "timestamp": now(),
            },
        )
    return success

7.4 一致性保证总结

八、实施难度评估与分步落地

8.1 实施难度

8.2 实测数据

在 200 个长程任务（10-20 步）上对比：

关键发现：

• 经验记忆预避 是单点收益最大的模块（重复犯错率 3%→1.1%，首步错误率 28%→12%）
• 情景记忆复用 是规划时间收益最大的模块（3.2s→1.8s）
• 冷启动成功率 从 54% 提至 79%——跨任务经验迁移是冷启动的杀手锏
• 写入闭环 让记忆持续增长，任务越多效果越好（飞轮效应）

8.3 分步落地建议

第一步：经验记忆预避（1-2 周，ROI 最高）
  → 在规划前查经验记忆，用教训修正规划
  → 只需 retrieve + apply_lessons 两个函数
  → 依赖第 9 篇经验库已有数据
  → 预期：重复犯错率 -60%，首步错误率 -40%
  → 这一步就能消除大部分"重复踩坑"问题

第二步：情景记忆复用（2-3 周）
  → 任务成功后写入情景记忆（完整轨迹）
  → 规划前查同类任务历史轨迹，高相似度复用
  → 依赖第 32 篇情景记忆层
  → 预期：规划时间 -40%，同类任务成功率 +2%

第三步：写入闭环（2-3 周）
  → 实现完整的触发评估 + 质量控制
  → 四种写入触发（成功/失败/偏差/模式）
  → 4 道质量关（去重/置信度/时效/频次）
  → 预期：记忆库噪音率 41%→6%，记忆持续增长

第四步：一致性保证（3-4 周）
  → 时效校验 + 冲突检测 + 事件通知
  → 接第 32 篇遗忘曲线和乐观锁
  → 预期：过时记忆污染率 -80%
  → 多 Agent 场景才需要，单 Agent 可暂缓

8.4 常见坑

坑 1：检索全查（不分层不预算）
  → 每次规划前查所有记忆层 → token 爆炸 → 窗口被记忆占满
  → 解法：按任务特征路由 + token 预算（第 26 篇）

坑 2：写入不过质量关
  → 每次执行都写 → 记忆库噪音爆炸 → 检索质量下降
  → 解法：4 道质量关，写入通过率 ~34%

坑 3：经验记忆不衰减
  → 教训永远不过期 → 环境变了还在避旧坑 → 避了旧坑踩新坑
  → 解法：经验记忆 stability=14 天 + 频次加权补偿

坑 4：记忆冲突不解决
  → 两条矛盾记忆都注入规划 → 规划器困惑 → 生成矛盾规划
  → 解法：时间优先 + 旧记忆降级

坑 5：只读不写（只检索不回写）
  → 记忆永远不增长 → Agent 不学习 → 每次都从零开始
  → 解法：写入闭环是飞轮，不写就没意义

九、总结与下一篇

9.1 本文核心

记忆与规划是双向数据流，不是两个孤岛：
  记忆→规划：四种记忆各驱动规划不同阶段
    情景复用历史 / 语义修正约束 / 程序填充模板 / 经验预避坑
  规划→记忆：执行后按事件写入四种记忆
    成功存情景 / 失败存经验 / 模式存程序 / 事实存语义

交叉闭环：检索→规划→执行→写入→（下次检索更好）
  → 飞轮效应：任务越多记忆越丰富 → 规划越准 → 成功率越高

关键洞察：
  纠错不如避错——第39篇Reflexion是执行后纠错，经验记忆是规划前预避
  冷启动靠迁移——情景记忆跨任务复用让冷启动成功率 54%→79%

9.2 与系列的关系

第 7 篇（规划概述）     → 规划深度自适应
第 39 篇（规划深潜）    → 四种机制 internals
第 26 篇（上下文工程）  → 记忆怎么进窗口
第 32 篇（记忆架构）    → 记忆怎么存怎么查
本文（记忆×规划交叉）   → 记忆和规划怎么双向联动

五篇递进：怎么规划 → 规划机制 → 记忆进窗口 → 记忆存储 → 记忆×规划联动

9.3 下一篇预告

记忆与规划的交叉让 Agent “越做越好”。但还有一个维度没碰：Agent 的自我认知与元认知——Agent 知道自己"擅长什么、不擅长什么"吗？能评估自己"这次有多大把握"吗？能在"没把握"时主动求助而非硬编吗？这是从"能做事"到"知道自己能做什么"的跨越，涉及能力边界建模、置信度校准、主动求助策略、元认知监控。下一篇将深潜 Agent 元认知：能力边界自建模、置信度校准（接第 31 篇决策审计的 confidence）、主动求助机制（接第 8 篇确认门的智能化升级）、元认知监控循环。

本文是 AI Agent 技术博客系列第 40 篇。系列目录见 blog.html。