ultron-memory

ultron-memory 是一个轻量、可本地部署的 Python 组件，用于构建反馈驱动的行为记忆与 Skill 自进化能力。

它把用户反馈沉淀为可复用的规则、护栏、事实和偏好，使用词法/语义混合检索注入后续 Agent 上下文，并为每次变更保留不可变版本与来源记录。

注意：本项目改变的是 Agent 的外部行为上下文，不会训练模型参数，也不会让模型发生参数级“自学习”。宿主仍负责主 Agent Loop、工具、权限和模型调用。

特性

• 反馈进化：从下一轮用户反馈中抽取至多一个候选，并由 Maintainer 决定 add、merge 或 discard。
• 四类记忆：rule、guardrail、fact、preference 分开存储、检索和注入。
• 混合检索：SQLite FTS5/BM25-lite 词法检索 + 可选 embedding 语义检索，通过 RRF（Reciprocal Rank Fusion）融合排序。
• 版本与审计：每个语义变更生成 immutable version，记录 parent、provenance、决策和使用统计。
• 可回滚：明确负反馈或多个独立硬失败可触发回滚；冲突来源保留，不物理删除历史。
• 可评测：从 provenance 生成 replay 样本，支持程序规则和可选 LLM Judge，并对比演化前后版本。
• 隐私优先：默认只持久化有界、脱敏的 pending evidence；凭据不会主动写入存储。
• 无运行时依赖：使用 Python 标准库和 SQLite，支持 Python 3.10 及以上版本。

安装与快速验证

项目要求 Python 3.10 及以上版本（Python 3.9 不支持本项目使用的 dataclass slots）。推荐在项目目录创建独立虚拟环境；如果本机命令名不是 python3.11，请替换为任意可用的 Python 3.10+ 解释器：

python3.11 -m venv .venv
.venv/bin/python -m pip install -e .
.venv/bin/python -m unittest discover -s tests -v
.venv/bin/python examples/evolution_demo.py

如果本地 Python/SQLite 提供 FTS5，组件会使用 SQLite FTS5；否则自动退回内置的 BM25-lite 风格词法评分器。单实例约 1 万条记录以内不需要向量数据库。

最小集成

宿主只需要提供模型回调；embedding 回调是可选的。embedding 不可用时，已有记录仍可通过 BM25/FTS 检索，新候选会进入 awaiting_embedding，待服务恢复后再重试。

from ultron_memory import EvolutionMemory

memory = EvolutionMemory(
    storage_dir="./.ultron-memory",
    namespace="my-project",
    side_query=side_query,  # async (system_prompt, JSON_payload) -> JSON 字符串
    embed=embed,            # async (text) -> list[float]，可选
)

memory.record_turn({
    "conversation_id": "c1",
    "user": "修改 API",
    "assistant": "API 已修改。",
})
await memory.observe_feedback("c1", "每次修改 API 都要更新并运行测试")
retrieval = await memory.retrieve(
    "修改订单 API",
    top_k=3,
    conversation_id="c1",
)
memory.record_outcome("c1", "tests_passed", metadata={"command": "pytest"})
report = await memory.evaluate()
paths = memory.export_skills("./exported-skills")
memory.close()

在真实 Harness 中，应将 retrieval.prompt_context 注入主模型的 system context，并把实际采用的版本传给 record_outcome()。完整生命周期示例见 examples/minimal_harness.py 和 UltronMemoryAdapter。

默认注入分区如下：

类型	注入区块	用途
rule	<rules>	正向行为规则
guardrail	<warnings>	错误经验与禁止事项，以警告形式呈现
fact	<facts>	项目、环境或业务事实
preference	<preferences>	用户稳定偏好

只有 active 的 rule 会导出为 SKILL.md；其他类型保留在结构化存储和公共 API 中。

自进化流程

record_turn()
    ↓
保存 pending window
    ↓
observe_feedback()
    ↓
Extractor 抽取候选
    ↓
RRF 混合检索相似记录
    ↓
Maintainer 决定 add / merge / discard
    ↓
程序安全门禁与事务写入
    ↓
后续 retrieve() 注入新版本
    ↓
record_outcome() 记录真实结果
    ↓
replay、负反馈和硬失败更新健康状态

反馈与候选

side_query(system_prompt, payload) 由宿主接入任意模型，Extractor 要求返回严格 JSON，例如：

{
  "persist": true,
  "kind": "rule",
  "title": "接口修改同步测试",
  "trigger": "修改 API 或接口行为时",
  "content": "修改接口后同步更新并运行自动化测试",
  "confidence": 0.91,
  "evidence": "用户明确提出长期要求"
}

Maintainer 会结合 exact-match 和相似检索结果返回：

{
  "action": "merge",
  "target_id": "memory_017",
  "reason": "与已有接口测试规则相似，并增加了运行测试要求",
  "confidence": 0.88,
  "merged_content": "..."
}

模型负责语义判断；程序负责 JSON/字段校验、敏感信息脱敏、embedding 维度、provenance、事务、幂等和版本完整性。低置信度或明显冲突的候选不会直接污染 active 数据。

Pending window 与隐私

1. record_turn() 默认只在当前进程保留完整 turn；SQLite 只保存有界、脱敏的 preview。
2. 设置 save_raw_evidence=True 才会保留完整（仍经脱敏处理）的证据。
3. embedding 服务暂时不可用时，候选会持久化为 pending，可通过 await memory.retry_pending() 重试。
4. 所有语义写入都必须携带可持久化来源：conversation id、保留的 feedback 或 evidence。仅有 decision_reason 不算来源。

混合检索

• 词法侧使用 SQLite FTS5/BM25-lite，默认候选 Top-20。

• 语义侧通过宿主注入的 embed(text) -> list[float] 回调召回 Top-20。

• 两侧使用 Reciprocal Rank Fusion 融合，而不是直接相加不同尺度的分数：

  RRF(d) = Σ 1 / (k + rank(d))
  

  默认 k=60，并按确认时间做时间衰减。

• embedding 不可用时继续 BM25-only 检索；新候选暂停激活，等待向量生成成功。

• 同一 conversation 的每次 retrieve() 都有独立 retrieval batch；record_outcome() 默认只归因最近一次召回，跨进程重启仍然有效。显式传入 metadata["retrieved_version_ids"] 时优先使用显式版本。

版本、冲突与回滚

每次 add/merge 都生成不可变版本，并保存以下信息：

memory_id
version
parent_version
action
source_conversation_id
source_feedback
retrieved_version_ids
decision_reason
created_at

冲突记录不会被删除：模型可以选择 preferred 记录，被替代的一方会标记为 superseded、conflict 或 shadow。自动回滚条件为：

• 用户明确负反馈；或
• 两个不同 conversation/replay 样本出现硬失败。

一次偶然失败只会进入 watch，不会立即回滚。reject 与 restore 在同一事务中完成，避免留下“当前版本已拒绝但没有健康版本”的半完成状态。

Replay 评测

evaluate() 是 Memory/Skill 的回复级、规则级 replay 评测，不是完整 Coding Agent benchmark。它可以检查：

• 上下文是否非空：nonempty；
• 是否为合法 JSON：json / json_parseable；
• 是否包含或禁止指定文本：contains / not_contains；
• 是否包含来源引用：cite_sources；
• 最大长度：max_chars；
• 记录类型：kind。

样本可以通过 add_replay_sample() 手动写入，也可以直接传给 MemoryEvaluator.evaluate(samples=...)。没有显式样本时，评测器会从 provenance 生成保守的 replay 样本；有 side_query 时才启用可选语义/LLM Judge，模型失败会退回程序规则并记录原因。

报告中的主要指标：

retrieval_hit_at_k       # 是否召回了来源 Memory
version_match_rate       # 指定版本时是否精确命中 immutable version
rule_pass_rate           # 规则通过率
rule_pass_rate_delta     # 相对演化前 baseline 的变化
feedback_correction_rate # 反馈后得到有效修正的比例
regression_rate          # 演化后出现回归的比例
rollback_rate            # 回滚比例

Replay 检索不会增加线上 retrieved 计数。插件为 merge 样本自动保存 before_version；未显式提供 baseline_samples 时，evaluate() 会自动重放旧版本快照并生成前后对比。

评测报告会以脱敏审计产物保存在本地，可通过 memory.list_evaluations() 读取。它不是生产环境的 Champion Registry，也不等价于真实工具执行成功率、代码正确率或端到端 Agent 任务准确率；这些结果应由 record_outcome() 和宿主侧测试提供。

导出 Skills

from ultron_memory.exporter import export_skills

files = export_skills(memory, "./exported-skills")

每个 active rule 会写入确定性的 <slug>/SKILL.md，包含标题、触发条件、指令、版本和来源 Memory ID。已有的无关文件会保留。事实、偏好、guardrail、shadow 记录和 rejected 版本继续保存在 SQLite 与公共 API 中。

Benchmark：命中、提升、回归、成本与延迟

仓库顶层的 benchmarks/ 是独立评测工具，不会改变运行时包的行为。默认命令使用固定的离线数据集和概念 embedding stub，不访问外部网站，也不需要 API Key：

PYTHONPATH=src .venv/bin/python -m benchmarks.runner \
  --data benchmarks/data/cases.jsonl \
  --output-dir benchmarks/reports \
  --top-k 3 --repetitions 5 --warmups 2

命令会生成 benchmarks/reports/latest.json 和 benchmarks/reports/latest.md。报告中的主要字段如下：

指标	定义	解释边界
retrieval_hit_at_k	正向 probe 的 gold Memory 是否出现在 Top-k	只证明召回，不证明回答正确
retrieval_version_hit_at_k	是否命中指定 immutable version	用来区分演化前后的版本传播
rule_pass_rate	正向 probe 的 required/forbidden 规则通过率	离线版检查记忆上下文，不是主模型生成质量
feedback_correction_rate	before 失败、after 通过的比例	只在配对样本上计算
positive_rule_regression_rate	before 通过、after 失败的正向规则比例	与负样本特异性回归分开报告
negative_target_hit_at_k	负向 probe 命中其自身 target 的比例	越高表示拒绝/相关性阈值越不足，不等同于所有错误召回率
latency_ms	检索、契约检查及组合路径的 mean/p50/p95	离线组合路径不包含模型生成或工具执行

rrf_evolved 会通过公开的 record_turn() → observe_feedback() 闭环把 fixture 反馈合并为 v2；它使用 oracle sidecar，只验证版本写入和后续召回传播，不能据此宣称真实 LLM 抽取能力。当前受控集的负样本可能出现过度召回，因此必须同时查看 negative_target_hit_at_k 和回归率，不能只看正向通过率提升。

在线回答级评测

如果需要测量真实模型生成质量，可使用在线 runner。它要求模型返回严格 JSON，再由程序检查 probe 的 required/forbidden 条件；不会让模型自己充当 judge：

export BASE_URL="https://your-openai-compatible-gateway/v1"
export API_KEY="<your-key>"
export MODEL="your-model"
# 按供应商价格填写；不确定时保持注释，报告会显示 unknown
# export INPUT_USD_PER_MTOK="2.00"
# export OUTPUT_USD_PER_MTOK="8.00"

PYTHONPATH=src .venv/bin/python -m benchmarks.online_runner \
  --data benchmarks/data/cases.jsonl \
  --output-dir benchmarks/reports \
  --top-k 3 --repetitions 1 --warmups 0 \
  --limit-cases 2

在线报告写入 online-latest.json/online-latest.md，额外展示 generation 和真正包含“检索 + 生成”的 end_to_end 延迟。默认 rrf_evolved 仍使用离线 oracle；加 --evolve-with-model 才会用配置的模型执行 Extractor/Maintainer，并按 evolution_extractor、evolution_maintainer、workload_generation、warmup_generation 分阶段计量。

成本只接受 provider 返回的 usage 和显式价格：

• usage 缺失显示 unknown，不从字符数或本地 tokenizer 猜 token；
• 价格缺失时 token 仍可展示，但成本为 unknown；
• 离线模式没有网络请求，成本显示 N/A (offline)，不是伪造的 $0；
• 报告和错误信息会脱敏 API Key，但仍应避免把密钥写入命令历史或数据集。

在线评测同样不是完整 Coding Agent benchmark：它不执行 Shell、文件编辑、MCP 或真实代码测试。若要证明端到端收益，应把宿主实际工具结果通过 record_outcome() 关联到被召回的版本，并单独报告工具成功率、成本和延迟。

与 Ultron / DeepSeek Harness 集成

本项目只负责“行为记忆插件”边界，不搬入完整 Agent Loop、MCP、Shell/文件工具、权限模式或 Session Compaction。宿主 Harness 可以在每轮请求前调用 retrieve()，将返回的 prompt context 注入模型；请求完成后调用 record_turn()，收到用户反馈时调用 observe_feedback()，任务结束时调用 record_outcome()。

OpenAI-compatible 回调示例见 examples/deepseek_adapter.py。该示例支持通过环境变量配置聊天模型；未配置 embedding 模型时自动使用 BM25-only 模式。

项目结构

ultron-memory/
├── pyproject.toml
├── README.md
├── LICENSE
├── src/ultron_memory/
│   ├── models.py       # 数据模型与公共数据契约
│   ├── plugin.py       # EvolutionMemory 主编排 API
│   ├── store.py        # SQLite、版本、provenance 与事务
│   ├── retriever.py    # BM25/FTS + embedding + RRF
│   ├── extractor.py    # 反馈候选抽取
│   ├── maintainer.py   # add/merge/discard 决策
│   ├── evaluator.py    # replay 评测
│   ├── exporter.py     # SKILL.md 导出
│   └── adapters/ultron.py
├── benchmarks/         # 离线/在线指标 runner 与报告
├── examples/
└── tests/

隐私与安全

证据在进入模型或 embedding 回调前会进行脱敏。默认情况下，完整 turn 只存在于当前进程；pending SQLite 行和审计字段使用有界、脱敏值。只有在宿主具备明确数据留存策略时才建议设置 save_raw_evidence=True。API Key、Bearer Token、密码等凭据不会被设计为持久化内容。

许可证

MIT，详见 LICENSE.