自动记忆:官方 memories 对齐的提取+注入闭环(确定性 v1)

开关默认关(generate/use),设置接口 GET/PUT /users/me/auto-memory-settings,任务级覆盖 POST /tasks/{id}/auto-memory-flags,手动触发 POST /users/me/auto-memory/run,日志 GET /users/me/auto-memory-log。

提取:worker run_loop 轮次边界扫 completed 任务,闲置/时长/长度/外部上下文/配额剩余百分比五道门,脱敏后写 project 级 identity_memories,同一任务去重、同一目标 supersede 合并;只写 project 级,不跨项目泄漏。

注入:worker run_one 首轮把最新各 5 条用户/项目记忆拼进内存 prompt 副本(goal 任务同步改内存 goal),库原 prompt 不动;续跑/断点/协调员不注入,注入失败 fail-open。

v1 诚实边界:extract/consolidation 模型只存设置不调模型;MCP/web 线程内调用暂无 tasks 表标记,外部上下文门只认任务间消息与协调员子任务;无控制台 UI。
This commit is contained in:
host committed 2026-10-28 12:12:10 +08:00
1 parent b93666fbe8
commit cbc90f9d30
8 files changed
+1370 -6

No files matched your search

+81
View File
@@ -145,6 +145,9 @@ from cloud.cost import CostStore, estimate_tokens, price_usd_micros, \
from cloud.db import connect as _db_connect
from cloud.logging_config import get_logger, log_event, setup_logging
from cloud.models import ModelRouter, is_fallbackable, record_fallback
from cloud.auto_memory import AutoMemoryStore
from cloud.auto_memory import build_injection_prefix as _auto_prefix
from cloud.auto_memory import process_due_tasks as _auto_process_due
from cloud.peer_messages import deliver_peer_message as _deliver_peer_message
from cloud.peer_messages import peer_message_prefix as _peer_prefix
from cloud.peer_messages import row_get as _row_get
@@ -977,6 +980,78 @@ class Worker:
status=event_type,
message="notify exception: %s" % exc)
# -- 自动记忆 ------------------------------------------------------
# 注入失败绝不挡任务(fail-open:无记忆跑也比不跑强);提取
# 失败只记日志(run_loop 轮次边界已包一层,见 process 方法)。
def _apply_auto_memory_injection(self, task):
"""首轮注入记忆:只改内存副本,原 prompt/goal 不落库。
跳过:协调员任务(不跑模型回合)、续跑与断点续跑(已有
会话历史)、无属主、use_memories 关、无记忆。goal 任务
同时改内存里的 goal(执行器取 goal 优先于 prompt),
前缀里写明“仅供参考,不得改写本任务目标”。
"""
if task.get("coordinator_parts"):
return task
if task.get("_resume") or task.get("_goal_resume_breakpoint"):
return task
owner = task.get("created_by")
if not owner:
return task
try:
store = AutoMemoryStore(self.db_path)
try:
flags = store.get_task_flags(task["id"])
finally:
store.close()
prefix = _auto_prefix(
self.db_path, owner, task["project_id"], flags=flags)
except Exception as exc:
try:
self.logger.warning("auto memory injection failed: %s",
exc)
except Exception:
pass
return task
if not prefix:
return task
new_task = dict(task)
prompt = task.get("prompt") or ""
new_task["prompt"] = prefix + "\n\n" + prompt
if task.get("goal_mode") or task.get("verify_command"):
new_task["goal"] = prefix + "\n\n" + (
task.get("goal") or prompt)
new_task["_memory_injected"] = True
self._audit(task["project_id"], "worker_memory_injected",
task["id"])
return new_task
def process_auto_memories(self, now=None, force=False,
owner_id=None):
"""扫 completed 任务并提取记忆(run_loop 轮次边界调用)。
返回 process_due_tasks 的结果列表;提取本身失败时返回
空列表并只记日志,不炸循环。提取成功的任务各补一条
worker 侧审计(服务端手动触发路径另有自己的审计)。
"""
try:
results = _auto_process_due(
self.db_path, now=now, force=force, owner_id=owner_id)
except Exception as exc:
try:
self.logger.warning("auto memory sweep failed: %s", exc)
except Exception:
pass
return []
for item in results:
if item.get("status") == "extracted":
# project_id 不在结果里:审计 detail 只带任务 id,
# project 字段用空串(与系统审计同款占位)。
self._audit("", "worker_memory_extracted",
item.get("task_id") or "")
return results
# -- 认领 ----------------------------------------------------------
def _budget_exceeded(self):
return budget_exceeded(self.usage_turns, self.max_turns,
@@ -3158,6 +3233,9 @@ class Worker:
else:
task["prompt"] = resume_message
task["_resume"] = True
# 自动记忆注入(官方 use_memories):首轮才注入,见
# _apply_auto_memory_injection 的跳过规则;只改内存副本。
task = self._apply_auto_memory_injection(task)
self._audit(task["project_id"], "worker_claim",
"%s executor=%s" % (task_id, self._executor_name()))
# log_event 只接受 task_id/project_id/status/message 四个
@@ -4122,6 +4200,9 @@ class Worker:
# G11/G12 后台投递:摘要(每小时)与 outbox 重试,
# 轮次边界做,不中断在跑任务,失败只记日志。
self._process_push_due()
# 自动记忆提取:同款轮次边界后台扫,失败只记日志
# (process_auto_memories 内部已兜底,不炸循环)。
self.process_auto_memories()
# 自动陈旧回收:死 worker 的 running 任务在这里被
# 标 interrupted;活任务心跳新鲜,不会被扫到。
if sweep_threshold is not None: