构建永不遗忘的AI智能体（第二部分）：传送功能
永不遗忘的AI智能体第二部分：跨会话的记忆传输、技能管理、记忆衰减机制，以及适用于Claude Code及类似智能体的分支级记忆功能。
以下笔记为“构建永不遗忘的AI智能体（第二部分）：瞬移、技能、衰减与分支级记忆”提供了一条实用实现路径。重点在于契约、校验机制以及可直接插入的代码占位符，而非激励性描述。 在完成概览阶段时，首先明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及代币或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
上一次的进展
“我们上次停下的地方”这一概念在被视为可度量的界面时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个结构就能进行审计。 保持数据结构的层次简单且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在出现中断后导致无法继续处理。
第5步：内存传输
将第5步的“内存传送”阶段视为可测量的界面来处理效果最佳。在扩大范围之前，先记录一份成功的案例、一个失败案例以及回滚说明。同时记录正常流程与恢复流程的细节。重试机制、人工干预环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。要保持图表状态简洁且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致流程无法继续。
工作流程
将工作流阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的流程示例、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某个步骤出现故障时，故障应指向单一的责任主体，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。
Agent A: snapshot("for_review.vdb")
| send file to Agent B
v
Agent B: open("for_review.vdb") -> work -> add new memories
| send file back
v
Agent A: merge("for_review.vdb")
-> new memories imported
-> conflicts detected and resolved
将工作流阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的流程示例、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询的成本。尽早了解成本情况，可避免在流程从演示环境过渡到共享环境时出现意外费用。
快照
在快照阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，需经过人工审批。编译时的连接方式并不等同于业务流程的完整性。
class MemoryTeleport:
def __init__(self, db: vectlite.Database):
self.db = db
def snapshot(self, path: str) -> str:
"""Create a portable copy of the entire database."""
self.db.snapshot(path)
return path
差异对比——无需修改即可预览
若要在不修改阶段的情况下进行差异预览，应在更改代码之前定义输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
def diff(self, remote_path: str,
namespaces: list[str] | None = None) -> MergeResult:
"""Preview what a merge would do, without writing anything."""
return self._merge_internal(remote_path, namespaces, dry_run=True)
合并——最有趣的部分
在“合并有趣部分”阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。 在“合并有趣部分”阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外账单。
冲突解决策略。
def merge(self, remote_path: str,
namespaces: list[str] | None = None,
strategy: ConflictStrategy = ConflictStrategy.NEWER_WINS
) -> MergeResult:
"""Merge memories from a remote .vdb into the local database."""
return self._merge_internal(
remote_path, namespaces,
dry_run=False, strategy=strategy,
)
冲突解决策略
在制定冲突解决策略时，首先列出相关要求：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持透明。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，以便操作员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
class ConflictStrategy(str, Enum):
LOCAL_WINS = "local_wins" # keep local version
REMOTE_WINS = "remote_wins" # keep remote version
NEWER_WINS = "newer_wins" # keep whichever has a more recent updated_at
KEEP_BOTH = "keep_both" # import remote as a new record
合并引擎
在处理合并引擎阶段时，首先写下相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会偏离原有设计。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。 在成本较高的操作之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的大型语言模型。
def _merge_internal(self, remote_path, namespaces=None,
dry_run=False, strategy=ConflictStrategy.NEWER_WINS):
result = MergeResult()
remote_db = vectlite.open(remote_path, dimension=DIMENSION)
try:
ns_list = namespaces or self._discover_namespaces(remote_db)
local_index = self._build_index(self.db, ns_list) for ns in ns_list:
remote_records = remote_db.list(namespace=ns, limit=10_000) for record in remote_records:
rec_id = record["id"]
remote_meta = record.get("metadata", {})
remote_hash = remote_meta.get("content_hash", "") if rec_id in local_index:
local_info = local_index[rec_id] if local_info["content_hash"] == remote_hash:
result.skipped_duplicates += 1 # same content
continue # Conflict -- same ID, different content
conflict = MergeConflict(
record_id=rec_id,
namespace=ns,
local_content=local_info.get("content", ""),
remote_content=remote_meta.get("content", ""),
local_updated=local_info.get("updated_at", ""),
remote_updated=remote_meta.get("updated_at", ""),
)
if not dry_run:
self._resolve_conflict(conflict, record, strategy)
result.conflict_resolutions += 1
result.conflicts.append(conflict)
else:
# New record -- import
if not dry_run:
self._import_record(record, ns)
result.imported += 1
finally:
remote_db.close() return result
实际运行效果
在“验证其功能”阶段，首先需明确相关约定：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元，而非庞大的脚本。当某一步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
from agentmem.store.memory_store import MemoryStore
from agentmem.services.teleport import MemoryTeleport
# Agent A: has been working for weeks
store_a = MemoryStore("agent_a.vdb")
tp = MemoryTeleport(store_a.db)# Create a snapshot for Agent B
tp.snapshot("for_agent_b.vdb")# ... send for_agent_b.vdb to Agent B's machine ...# Agent B works with the snapshot, adds new memories
store_b = MemoryStore("for_agent_b.vdb")
store_b.store(Memory(
id="mem_new_finding",
content="The payment API requires idempotency keys for all POST requests",
memory_type=MemoryType.REFERENCE,
importance=0.8,
))
store_b.save()
store_b.close()# ... send for_agent_b.vdb back to Agent A ...# Agent A merges the new knowledge
result = tp.merge("for_agent_b.vdb")
print(result.summary())
# -> Imported: 1 new memories
# Skipped: 47 duplicates
# Conflicts: 0
在“验证其功能”阶段，首先需明确相关约定：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 除了功能结果外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
为何以 .vdb 作为传输格式可行
将 .vdb 视为可测量的数据结构时，作为传输格式最为有效。在扩大范围之前，先记录一份理想的转录数据、一个故障案例以及回滚说明。
将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个数据结构即可进行审计。
保持数据结构的层次简单且类型明确。嵌套的二进制数据会掩盖哪个节点编写了哪个字段的信息，还会导致在中断后无法继续处理。
第6步：技能引擎
将第6阶段的技能引擎视为可度量的对象时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在出现中断后导致流程无法继续。
---
description: Docker best practices
glob_patterns: ["Dockerfile", "docker-compose*.yml"]
trigger_keywords: ["docker", "container", "compose"]
---
## Docker Guidelines
- Always use multi-stage builds
- Pin base image versions
- Never run as root in production
三种激活信号
将三个激活信号阶段视为可测量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想状态下的输出、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤出现故障时，故障应能指向单一责任主体，而非复杂的流程链。 保持图表状态简洁且具有类型定义。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。
User message: "deploy the containers to staging"
|
v
+-- SkillEngine.activate() ----------------------------+
| |
| 1. Hybrid search (semantic + keyword via VectLite) |
| -> matches "docker" skill (score 0.91) |
| -> matches "devops" skill (score 0.78) |
| |
| 2. Glob match (if file path detected in message) |
| -> "Dockerfile" mentioned? -> boost "docker" |
| |
| 3. Keyword boost |
| -> "containers" matches docker triggers |
| |
| 4. Weighted combination -> top-k activated skills |
+------------------------------------------------------+
|
v
Injected into system prompt as ## Active skills
将三个激活信号阶段视为可测量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想状态下的输出、一个故障案例以及回滚说明。 在功能结果之外，还需记录执行时间以及令牌或查询的成本。尽早了解成本情况，可避免在流程从演示环境过渡到共享环境时出现意外费用。
技能数据模型
在“技能数据模型”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。 当下一步操作为代码编写或工具调用时，应优先使用具有架构验证的结构化输出，而非自由形式的文本描述。
@dataclass
class Skill:
name: str # "docker", "python", "security"
content: str # Full Markdown instructions
glob_patterns: list[str] # ["Dockerfile", "*.py"]
trigger_keywords: list[str] # ["container", "docker"]
tools_allowed: list[str] # ["docker_build", "docker_push"]
importance: float = 0.7 # Base weight
tags: list[str] = field(default_factory=list)
description: str = ""
注册——在 VectLite 中对技能进行索引
在开发注册索引功能阶段，应在修改代码之前明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品功能的一部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。
NAMESPACE = "__skills__"
class SkillEngine:
def __init__(self, db: vectlite.Database,
max_active: int = 3, min_score: float = 0.30):
self.db = db
self.max_active = max_active
self.min_score = min_score def register(self, skill: Skill) -> str:
skill_id = f"skill_{skill.name}_{uuid.uuid4().hex[:6]}" # Rich embedding text: name + description + keywords + content
embed_text = (
f"{skill.name}: {skill.description or ''} "
f"{' '.join(skill.trigger_keywords)} "
f"{skill.content[:500]}"
)
vector = embed(embed_text)
sparse = vectlite.sparse_terms(embed_text) self.db.upsert(
skill_id, vector, skill.to_metadata(),
sparse=sparse, namespace=NAMESPACE,
)
return skill_id
激活机制——核心匹配逻辑
在激活核心匹配阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一责任主体，而非复杂的流程链。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。 在激活核心匹配阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果之外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示模式转为共享模式时出现意外账单。
环境。def activate(self, message: str, file_path: str | None = None,
k: int | None = None) -> list[ActivatedSkill]:
k = k or self.max_active
# 1. Hybrid search via VectLite
vector = embed(message)
sparse = vectlite.sparse_terms(message)
results = self.db.search(
vector,
k=min(k * 3, self.count()), # over-fetch for re-ranking
sparse=sparse, fusion="rrf",
namespace=NAMESPACE,
) # Build candidates with normalized semantic scores
candidates = {}
max_score = max((r.get("score", 0) for r in results), default=1) or 1
for r in results:
meta = r.get("metadata", {})
name = meta.get("name", "")
if name:
candidates[name] = _Candidate(
skill=self._meta_to_skill(meta),
semantic_score=r.get("score", 0) / max_score,
) # 2. Keyword boost
msg_lower = message.lower()
for name, cand in candidates.items():
hits = sum(1 for kw in cand.skill.trigger_keywords
if kw.lower() in msg_lower)
if hits > 0:
cand.keyword_score = min(0.5, hits * 0.2) # 3. Glob match
detected_paths = self._detect_paths(message)
if file_path:
detected_paths.append(file_path)
for name, cand in candidates.items():
for pattern in cand.skill.glob_patterns:
for path in detected_paths:
if fnmatch.fnmatch(os.path.basename(path), pattern):
cand.glob_score = 0.35
break # 4. Weighted combination
activated = []
for name, cand in candidates.items():
final_score = (
cand.semantic_score * 0.55
+ cand.keyword_score * 0.25
+ cand.glob_score * 0.15
+ cand.skill.importance * 0.05
)
if final_score >= self.min_score:
activated.append(ActivatedSkill(
skill=cand.skill,
score=final_score,
activated_by=..., # "semantic", "keyword", "glob", "combined"
)) activated.sort(key=lambda a: a.score, reverse=True) # Gap filter: drop skills scoring < 75% of top skill
if activated:
top = activated[0].score
activated = [a for a in activated if a.score >= top * 0.75] return activated[:k]
从 Markdown 文件加载技能
在处理“从 Markdown 文件加载技能”这一阶段时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型接口。
def load_from_directory(self, dir_path: str) -> list[str]:
"""Load skills from .md files with optional YAML-like frontmatter."""
ids = []
for md_file in sorted(Path(dir_path).glob("*.md")):
skill = self._parse_skill_file(md_file)
if skill:
ids.append(self.register(skill))
return ids
---
description: Python best practices
glob_patterns: ["*.py", "*.pyx", "pyproject.toml"]
trigger_keywords: ["python", "pip", "venv", "pytest"]
tools_allowed: ["python_run", "pytest"]
importance: 0.8
tags: ["language", "backend"]
---
## Python Guidelines
- Use type hints for all function signatures
- Prefer pathlib over os.path
- Use pytest with fixtures, not unittest
- Format with black, lint with ruff
提示词注入
在处理提示注入阶段时，首先列出相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会偏离原定方向。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。 缓存稳定的系统指令和工具结构。重复发送相同的开头信息是导致资源浪费的常见原因。
def format_for_prompt(self, activated: list[ActivatedSkill]) -> str:
if not activated:
return ""
lines = ["\n\n## Active skills\n"]
for act in activated:
lines.append(f"### {act.skill.name}")
if act.skill.tools_allowed:
lines.append(f"**Allowed tools:** {', '.join(act.skill.tools_allowed)}")
lines.append(act.skill.content)
lines.append("") return "\n".join(lines)
验证其功能
在“验证其功能”阶段，首先需明确相关约定：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元，而非庞大的脚本。当某一步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
engine = SkillEngine(db)
engine.register(Skill(
name="docker",
content="## Docker\n- Always use multi-stage builds\n- Pin versions",
glob_patterns=["Dockerfile", "docker-compose*.yml"],
trigger_keywords=["docker", "container", "image", "compose"],
))engine.register(Skill(
name="python",
content="## Python\n- Use type hints\n- Prefer pathlib\n- Use pytest",
glob_patterns=["*.py", "pyproject.toml"],
trigger_keywords=["python", "pip", "venv"],
))# Semantic + keyword match
activated = engine.activate("deploy the containers to staging")
# -> [ActivatedSkill(skill=<docker>, score=0.87, activated_by="combined")]# Glob match
activated = engine.activate("review the Dockerfile")
# -> [ActivatedSkill(skill=<docker>, score=0.91, activated_by="combined")]# Semantic match only (no exact keywords)
activated = engine.activate("optimize the CI/CD pipeline for faster builds")
# -> [ActivatedSkill(skill=<docker>, score=0.52, activated_by="semantic")]
在“验证其功能”阶段，首先需明确相关约定：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
第7步：内存衰减与冲突解决
将第7步的内存衰减阶段视为可测量的指标来处理效果最佳。在扩大范围之前，先记录一份理想的运行日志、一个故障案例以及回滚说明。 应将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 需保持系统状态的扁平化与类型化。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会导致在中断后无法继续运行。
衰减公式
将衰减公式阶段视为可测量的界面来处理时，其效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致流程无法继续。
new_importance = current_importance * 2^(-elapsed_hours / half_life)
The MemoryDecayEngine
将MemoryDecayEngine阶段视为可测量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的运行结果、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应能指向单一的责任主体，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致无法继续执行。
class MemoryDecayEngine:
def __init__(self, store: MemoryStore,
half_life_hours: float = 168.0, # 1 week
tombstone_threshold: float = 0.05,
conflict_threshold: float = 0.80):
self.store = store
self.half_life_hours = half_life_hours
self.tombstone_threshold = tombstone_threshold
self.conflict_threshold = conflict_threshold
应用衰减机制
将“应用衰减阶段”视为可测量的表面来处理效果最佳。在扩大范围之前，先记录一份成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续处理。
def apply_decay(self, now: datetime | None = None) -> DecayReport:
"""Run exponential decay over all memories."""
now = now or datetime.now(timezone.utc)
report = DecayReport()
for mtype in MemoryType:
memories = self.store.recall_by_type(mtype, k=500)
for mem in memories:
if mem.superseded_by:
continue # already replaced report.total_scanned += 1
last_accessed = self._parse_time(
mem.last_accessed or mem.created_at
)
elapsed_hours = max(
(now - last_accessed).total_seconds() / 3600.0, 0.0
) # The core formula
decay_factor = math.pow(
2.0, -elapsed_hours / self.half_life_hours
)
new_importance = mem.importance * decay_factor if new_importance < self.tombstone_threshold:
# Below threshold -> auto-delete
self.store.update_importance(mem.id, 0.0)
self.store.forget(mem.id)
self.store.compact()
report.pruned += 1
elif abs(new_importance - mem.importance) > 0.001:
# Importance decayed -> update
self.store.update_importance(mem.id, new_importance)
report.decayed += 1
else:
report.untouched += 1 return report
冲突检测
将冲突检测阶段视为可度量的对象时，其效果最佳。在扩大范围之前，先记录一份理想的测试用例、一个故障案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解这些成本，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图结构的层次简单且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在出现中断后导致流程无法继续。
def check_conflicts(self, new_memory: Memory) -> list[ConflictResult]:
"""Find existing memories that contradict the new one."""
# Get candidates via VectLite search
similar = self.store.find_similar(
new_memory.content,
memory_type=new_memory.memory_type,
k=5, min_score=0.0,
) new_hash = self.store._hash(new_memory.content)
new_vector = embed(new_memory.content) conflicts = []
for existing, _ in similar:
if existing.id == new_memory.id:
continue
if existing.superseded_by:
continue # Skip exact duplicates (same hash = not a conflict)
if self.store._hash(existing.content) == new_hash:
continue # Compute direct cosine similarity
existing_vector = embed(existing.content)
cosine_sim = self._cosine_similarity(new_vector, existing_vector) if cosine_sim >= self.conflict_threshold:
conflicts.append(ConflictResult(
existing=existing,
new_content=new_memory.content,
similarity=cosine_sim,
resolution=self.default_resolution,
)) return conflicts
具备冲突解决功能的存储系统
将具有冲突解决阶段的存储系统视为可度量的对象时，其效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个失败案例以及回滚说明。 将配置信息与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一处，这样操作人员无需查看整个系统结构即可进行审计。 保持系统状态的扁平化与类型化。嵌套的数据块会掩盖是哪个节点修改了哪个字段，且在中断后会导致恢复失败。
def store_with_conflict_check(self, memory: Memory,
resolution=None) -> tuple[str, list]:
resolution = resolution or self.default_resolution
conflicts = self.check_conflicts(memory)
if not conflicts:
return self.store.store(memory), [] for conflict in conflicts:
conflict.resolution = resolution if resolution == ConflictResolution.SUPERSEDE:
mem_id = self.store.store(memory)
self.store.mark_superseded(conflict.existing.id, mem_id) elif resolution == ConflictResolution.KEEP_BOTH:
mem_id = self.store.store(memory) elif resolution == ConflictResolution.KEEP_EXISTING:
mem_id = "" # don't store the new one return mem_id, conflicts
实际应用效果
将“验证其功能”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一个成功的用例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。保持图结构的状态简洁且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致流程无法继续。
store = MemoryStore("decay_demo.vdb")
decay = MemoryDecayEngine(store)
# Store two memories
store.store(Memory(id="m1", content="The database is PostgreSQL 14",
memory_type=MemoryType.PROJECT, importance=0.7))# Later, the project migrates
new_mem = Memory(id="m2", content="The database is PostgreSQL 16",
memory_type=MemoryType.PROJECT, importance=0.8)mem_id, conflicts = decay.store_with_conflict_check(new_mem)
print(conflicts[0])
# -> Conflict: [project] sim=0.953 -> supersede
# m1 is now marked as superseded by m2# Weeks later, run decay
report = decay.apply_decay()
print(report)
# -> Decay: 15 scanned, 8 decayed, 2 pruned, 5 untouched
第8步：考虑工作树的内存管理
将第8步的“工作树感知内存”阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某一步骤出现故障时，故障原因应能明确指向某个具体的责任模块，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在进程中断后导致无法继续执行。 将第8步的“工作树感知内存”阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的执行日志、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询的成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外的费用支出。
命名空间策略
在命名空间策略阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
# Global -- shared across all branches
GLOBAL_TYPES = {MemoryType.USER, MemoryType.FEEDBACK}
# Scoped -- isolated per branch
SCOPED_TYPES = {MemoryType.PROJECT, MemoryType.REFERENCE}
Memory Type Branch VectLite Namespace
-----------------------------------------------------
user (any) "user"
feedback (any) "feedback"
project main "project:main"
project feature/auth "project:feature/auth"
reference main "reference:main"
reference feature/auth "reference:feature/auth"
WorktreeManager
对于WorktreeManager阶段，在修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
BRANCH_SEP = ":"
class WorktreeManager:
def __init__(self, db: vectlite.Database,
branch: str | None = None):
self.db = db
self._branch = branch or self._detect_git_branch() def resolve_namespace(self, memory_type: MemoryType) -> str:
"""Map a MemoryType to its VectLite namespace."""
if memory_type in GLOBAL_TYPES:
return memory_type.value # "user", "feedback"
return f"{memory_type.value}{BRANCH_SEP}{self._branch}"
# -> "project:main", "reference:feature/auth" def recall_namespaces(self) -> list[str]:
"""All namespaces to search for the current branch."""
namespaces = []
for mt in MemoryType:
if mt in GLOBAL_TYPES:
namespaces.append(mt.value)
else:
namespaces.append(f"{mt.value}{BRANCH_SEP}{self._branch}")
return namespaces
# -> ["user", "feedback", "project:main", "reference:main"]
自动检测git分支
在自动检测 Git 分支阶段时，应在修改代码之前明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一的责任主体，而非混乱的整个流程。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
@staticmethod
def _detect_git_branch() -> str:
"""Auto-detect the current git branch."""
try:
result = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True, text=True, timeout=5,
)
if result.returncode == 0:
branch = result.stdout.strip()
if branch and branch != "HEAD":
return branch
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
return "main" # fallback
为自动检测 git 分支阶段，应在修改代码之前定义输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境切换到共享环境时出现意外费用。
切换分支
在处理“分支切换”阶段时，首先写下相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
def switch_branch(self, branch: str) -> str:
"""Switch to a different branch context. Returns previous branch."""
prev = self._branch
self._branch = branch
return prev
合并分支（类似 git 的合并功能）
在处理类似 git stage 的分支合并时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改符合要求。 同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。 在耗时较高的操作之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的大型语言模型。
def merge_branch(self, source: str,
into: str | None = None) -> BranchMergeResult:
"""Merge branch-scoped memories from source into target."""
target = into or self._branch
result = BranchMergeResult(source=source, target=target)
for mt in SCOPED_TYPES: # only project + reference
src_ns = f"{mt.value}{BRANCH_SEP}{source}"
tgt_ns = f"{mt.value}{BRANCH_SEP}{target}" src_records = self.db.list(namespace=src_ns, limit=500) # Build dedup set from target
tgt_records = self.db.list(namespace=tgt_ns, limit=500)
tgt_hashes = {
r.get("metadata", {}).get("content_hash", "")
for r in tgt_records
} for record in src_records:
content_hash = record.get("metadata", {}).get("content_hash", "")
if content_hash in tgt_hashes:
result.skipped_duplicates += 1
continue # Import with new ID into target namespace
new_id = f"mem_{uuid.uuid4().hex[:12]}"
new_meta = dict(record.get("metadata", {}))
new_meta["merged_from"] = source self.db.upsert(
new_id, record["vector"], new_meta,
sparse=record.get("sparse", {}),
namespace=tgt_ns,
)
result.imported += 1 return result
差异对比与删除
在处理差异对比与删除阶段时，首先需明确合同规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某一步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在处理差异对比与删除阶段时，首先需明确合同规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及token或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外收费。
def diff_branch(self, source: str,
target: str | None = None) -> BranchDiffResult:
"""Like git diff -- preview without modifying anything."""
# Same logic as merge_branch but without writing
...
def delete_branch(self, branch: str) -> int:
"""Delete all branch-scoped memories. Global types untouched."""
total_deleted = 0
for mt in SCOPED_TYPES:
ns = f"{mt.value}{BRANCH_SEP}{branch}"
records = self.db.list(namespace=ns, limit=500)
for record in records:
self.db.delete(record["id"])
total_deleted += 1
if total_deleted > 0:
self.db.compact()
return total_deleted
查看完整生命周期
将“查看完整生命周期”这一阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先收集一份典型的成功案例、一个故障实例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个架构就能进行审计。 需保持架构状态的简洁性与类型化。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，且在中断后会导致恢复失败。
wt = WorktreeManager(db, branch="main")
# Store memories on main
store.store(Memory(content="FastAPI backend", memory_type=MemoryType.PROJECT))
# -> stored in namespace "project:main"store.store(Memory(content="I prefer type hints", memory_type=MemoryType.USER))
# -> stored in namespace "user" (global)# Switch to feature branch
wt.switch_branch("feature/auth")store.store(Memory(content="OAuth2 with PKCE for mobile clients",
memory_type=MemoryType.PROJECT))
# -> stored in namespace "project:feature/auth"# Recall on feature/auth -- sees global + branch-scoped
namespaces = wt.recall_namespaces()
# -> ["user", "feedback", "project:feature/auth", "reference:feature/auth"]
# "I prefer type hints" is visible (global)
# "FastAPI backend" is NOT visible (it's in project:main)
# "OAuth2 with PKCE" is visible (feature branch)# Merge feature into main
result = wt.merge_branch("feature/auth", into="main")
# -> Imported: 1, Skipped: 0# Now on main, "OAuth2 with PKCE" is also visible
wt.switch_branch("main")
# -> recall sees: "FastAPI backend" + "OAuth2 with PKCE" + all globals# Clean up
wt.delete_branch("feature/auth")
# -> project:feature/auth and reference:feature/auth emptied
最终项目结构
将最终项目结构阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。保持图结构的层次简单且类型明确，嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致流程无法继续。
agentmem/
core/
memory_types.py # MemoryType + Memory dataclass
embedder.py # embed() / embed_batch()
store/
memory_store.py # MemoryStore -- VectLite persistence
services/
extractor.py # Background memory extraction
session.py # Session summary
compaction.py # 4-level cascading compaction
swarm.py # Multi-agent swarm
teleport.py # Snapshot / merge / diff
skills.py # Skill engine
decay.py # Exponential decay + conflict resolution
worktree.py # Branch-scoped namespaces
agent.py # Agent -- the full loop
mcp_server.py # MCP server (bonus)
全貌
“完整视图”阶段若被视为可度量的对象，效果最佳。在扩大范围之前，需记录一份理想的执行日志、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先使用小型且可测试的单元。当某一步骤出错时，故障应指向单一责任主体，而非复杂的流程链。 保持图表状态简洁且具有类型定义。嵌套的数据结构会掩盖哪个节点修改了哪个字段，还会在流程中断后导致无法继续执行。 “完整视图”阶段若被视为可度量的对象，效果最佳。在扩大范围之前，需记录一份理想的执行日志、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。
User message
|
v
+-- Agent.chat() ----------------------------------------+
| |
| RECALL: hybrid search (HNSW + BM25 + RRF) [P1] |
| SKILLS: semantic + keyword + glob activation [P2] |
| BUILD PROMPT: memories + skills + session summary |
| COMPACT: snip -> micro -> collapse -> auto [P1] |
| LLM RESPONSE |
| EXTRACT: background sub-agent learns facts [P1] |
| SESSION: update summary if threshold crossed [P1] |
| |
| TELEPORT: snapshot / merge / diff [P2] |
| DECAY: exponential forgetting + conflict detect [P2] |
| WORKTREE: branch-scoped namespaces [P2] |
| SWARM: multi-agent via shared namespaces [P1] |
| |
| Storage: single .vdb file (VectLite) |
+--------------------------------------------------------+
VectLite 的优势
对于 What VectLite 所支持的阶段，在修改代码之前需明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务流程的完整性。
与 Claude Code 的对比
在将代码进行修改之前，需先明确与阶段对比的标准、该步骤的负责人以及结束条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
试一试
在“尝试使用”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相比冗长的脚本，更应采用小型、可测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体职责，而非整个复杂的流程。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
pip install vectlite sentence-transformers
git clone https://github.com/mcsedition-hub/agentmem-vectlite.git
cd agentmem-vectlite
python demo.py
在“尝试使用”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 除了功能结果外，还应记录执行时间以及令牌或查询的成本。尽早了解成本情况，可避免在流程从演示环境过渡到共享环境时出现意外账单。
操作检查清单
在制定操作检查清单时，需明确输入内容、各步骤的负责人以及完成标准，然后再进行代码修改。操作人员应能够从已知的检查点重新执行相应步骤，而无需猜测隐藏状态。
将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。
对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务上的完整性。
编写简短的操作手册：包括如何轮换密钥、如何清空队列、以及如何回滚上一次的数据导入操作。