218 lines
9.4 KiB
Python
218 lines
9.4 KiB
Python
"""双层记忆: `workspace/users/<user_id>/.memory/` (§3.7 / §7.4)。
|
|
|
|
core.md —— 注 system prompt,每次都看到。装稳定事实
|
|
(用户偏好 / 常用命令 / 项目约定 / 模型 quirk 备忘等)
|
|
extended/<x>.md —— 索引(frontmatter `description` + 路径)注 prompt,内容 agent
|
|
用 `read` 按需拉。装少数任务才用的专题资料(某 API 速查 / 某历史
|
|
事件等)。description 是召回依据:写得准,模型才知道何时该拉。
|
|
|
|
为什么这样切:
|
|
core 一直挂在上下文里,token 成本固定 ⇒ 只放跨任务高频用的精炼内容
|
|
extended 索引只占几行,内容按需付费 ⇒ 适合大量低频专题
|
|
|
|
memory 是 per-user(同一 workspace 内按 user_id 隔离),同 user 的所有 task 共享。
|
|
**dotfile `.memory/` 命名**:跟用户起的项目目录(同样落 `<uid>/` 下)区分,避免
|
|
项目名取 `memory` 时撞名;`.` 起头也被 `validate_task_name` 拒,双向防呆。
|
|
user_id 由 web auth 入口(JWT `sub`)透传到 build_agent。SaaS 化时 `<storage_root>`
|
|
替换 `workspace`,布局不变(§7.0)。
|
|
|
|
写入路径(agent 自管):memory_block 把 `.memory/` 的**可写绝对路径**(host 绝对路径 /
|
|
docker `/workspace/.memory`)连同「记忆维护契约」一起注进 prompt,agent 用已有
|
|
`write`/`edit`/`grep` 直接维护 —— 不引专用工具。契约 + 锚点即使记忆为空也常驻,
|
|
否则新用户冷启动永远不知道自己能记。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
from uuid import UUID
|
|
|
|
|
|
def _memory_dir(workspace_dir: Path, user_id: UUID) -> Path:
|
|
return workspace_dir / "users" / str(user_id) / ".memory"
|
|
|
|
|
|
def _parse_frontmatter_description(text: str) -> Optional[str]:
|
|
"""取 YAML frontmatter 里的 `description:` 一行;没有 frontmatter 返回 None。
|
|
|
|
只认文件最开头的 `---` ... `---` 块,块内首个 `description:` 行的值。
|
|
刻意不引 yaml 依赖 —— 记忆文件 frontmatter 就这一个字段够用,手解析最省。
|
|
"""
|
|
lines = text.splitlines()
|
|
if not lines or lines[0].strip() != "---":
|
|
return None
|
|
for raw in lines[1:]:
|
|
stripped = raw.strip()
|
|
if stripped == "---":
|
|
break
|
|
if stripped.startswith("description:"):
|
|
val = stripped[len("description:"):].strip()
|
|
# 去掉可能的引号
|
|
if len(val) >= 2 and val[0] in "'\"" and val[-1] == val[0]:
|
|
val = val[1:-1]
|
|
return val.strip() or None
|
|
return None
|
|
|
|
|
|
def _read_first_title(text: str, stem: str) -> str:
|
|
"""取文件第一个非空 h1/h2 行作为标题;没有就用文件名 stem。
|
|
|
|
legacy 兜底:存量 extended 文件没 frontmatter,退回首行当标题(平滑兼容)。
|
|
"""
|
|
for raw in text.splitlines():
|
|
line = raw.strip()
|
|
if line == "---":
|
|
continue
|
|
if line.startswith("#"):
|
|
return line.lstrip("#").strip()
|
|
if line:
|
|
return line[:60]
|
|
return stem
|
|
|
|
|
|
def _load_core(workspace_dir: Path, user_id: UUID) -> str:
|
|
p = _memory_dir(workspace_dir, user_id) / "core.md"
|
|
if not p.is_file():
|
|
return ""
|
|
try:
|
|
return p.read_text(encoding="utf-8").strip()
|
|
except (OSError, UnicodeDecodeError):
|
|
return ""
|
|
|
|
|
|
def _extended_index(workspace_dir: Path, user_id: UUID) -> List[Tuple[str, str]]:
|
|
"""返回 [(description_or_title, filename), ...],按文件名排序。
|
|
|
|
优先 frontmatter `description`;没有则退回首行标题(legacy 兼容)。
|
|
返回 filename(非绝对路径)—— 路径由 memory_block 按 backend 拼 display 前缀,
|
|
docker 下要给容器路径而非宿主路径。
|
|
"""
|
|
ext_dir = _memory_dir(workspace_dir, user_id) / "extended"
|
|
if not ext_dir.is_dir():
|
|
return []
|
|
items: List[Tuple[str, str]] = []
|
|
for p in sorted(ext_dir.glob("*.md")):
|
|
if not p.is_file():
|
|
continue
|
|
try:
|
|
text = p.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeDecodeError):
|
|
text = ""
|
|
label = _parse_frontmatter_description(text) or _read_first_title(text, p.stem)
|
|
items.append((label, p.name))
|
|
return items
|
|
|
|
|
|
_CONTRACT = """\
|
|
你可以**主动维护**这份记忆(用已有 `write`/`edit`/`grep`),把跨 task 复用的事实存下来,
|
|
下次别的 task 一开场就能用。规矩:
|
|
|
|
- **什么值得记**:用户稳定偏好、项目长期约定、反复用到的事实、踩过的坑(及为什么)。
|
|
**不要记**:只跟当前 task 有关的一次性信息、能从产物/代码里直接看到的东西。
|
|
- **core.md(常驻,贵)**:只放跨 task 高频、精炼的稳定事实 —— 它每轮都占 token。
|
|
- **extended/<slug>.md(按需,便宜)**:低频专题资料,一事一文件。文件开头写
|
|
frontmatter `description:` 一行(这行进上面索引,决定何时被召回),正文是事实本身:
|
|
```
|
|
---
|
|
description: <一句话说清这份资料是什么、何时该拉>
|
|
---
|
|
<内容>
|
|
```
|
|
- **写前先查重**:`grep`/`read` 看现有记忆有没有,有就 `edit` 更新、别堆重复;发现记错了就删。
|
|
- **用户让你"记住 / 改 / 忘掉"某事时,这是直接指令**:照办 —— "记住"就写、"改成"就 `edit`、
|
|
"忘掉 / 删掉"就把对应条目从 core.md 删掉或删掉那个 extended 文件。改完回一句确认即可。
|
|
- 记忆即时生效(每个新 task 重读),不用通知用户。"""
|
|
|
|
|
|
def memory_block(
|
|
workspace_dir: Path,
|
|
user_id: UUID,
|
|
mem_dir_display: Optional[str] = None,
|
|
) -> str:
|
|
"""构造注入 system prompt 的记忆段。
|
|
|
|
mem_dir_display: `.memory/` 在 agent 视角下的可写绝对路径前缀。host backend 传
|
|
None ⇒ 用宿主绝对路径;docker backend 传 `/workspace/.memory`(容器内路径)。
|
|
与旧版不同:契约 + 写入锚点常驻(即使记忆空),让 agent 知道自己能记;core /
|
|
extended 两段仍按有无内容才出现。
|
|
"""
|
|
core = _load_core(workspace_dir, user_id)
|
|
ext = _extended_index(workspace_dir, user_id)
|
|
|
|
real_dir = _memory_dir(workspace_dir, user_id)
|
|
base = mem_dir_display if mem_dir_display is not None else str(real_dir)
|
|
base = base.rstrip("/")
|
|
|
|
parts = ["\n\n## 记忆 (user 级,跨 task 共享)\n"]
|
|
parts.append(_CONTRACT)
|
|
parts.append(
|
|
f"\n\n**写到这里**:core → `{base}/core.md`;"
|
|
f"专题 → `{base}/extended/<slug>.md`\n"
|
|
)
|
|
# 快捷指令(与记忆是两套机制):触发词 → 完整指令的映射,存 shortcuts.md。**内容不注上下文**
|
|
# (入口层查表展开,不靠你召回),这里只给"能维护 + 格式",让你在用户要建/改快捷词时会写。
|
|
parts.append(
|
|
f"\n**快捷指令**:用户说\"记个快捷词 X → Y\"/\"把快捷词 X 改成/删掉\"时,维护 "
|
|
f"`{base}/shortcuts.md`(先 `read` 再 `edit`)。格式是两列 markdown 表 "
|
|
f"`| 触发词 | 完整指令 |`(表头 + `|---|---|` 分隔行 + 每条一行;触发词别含 `|`)。"
|
|
f"之后用户在任意入口(网页/微信/企业微信)整条打这个触发词,系统自动展开成完整指令 —— "
|
|
f"你无需在对话里替他执行触发,只负责把这行写对。\n"
|
|
)
|
|
if core:
|
|
parts.append("\n### Core (常驻 prompt)\n")
|
|
parts.append(core)
|
|
if ext:
|
|
parts.append("\n\n### Extended (按需用 `read` 加载)\n")
|
|
for label, name in ext:
|
|
parts.append(f"- `{base}/extended/{name}` — {label}\n")
|
|
return "".join(parts)
|
|
|
|
|
|
# ── 只读视图(web GUI 用) ─────────────────────────────────────────────
|
|
# 前端「记忆」弹框只读展示用。**故意不提供写/删 API** —— 改记忆全走对话(agent
|
|
# 自管,见 _CONTRACT),GUI 当"眼睛"不当"手":看全貌靠直接读 FS(便宜、是地面真相),
|
|
# 改靠模型(统一写入口、自然语言、能合并改写)。详见 DESIGN §3.7。
|
|
|
|
_EXTENDED_NAME_RE = re.compile(r"^[\w\-.]+\.md$")
|
|
|
|
|
|
def _is_safe_extended_name(name: str) -> bool:
|
|
"""防穿越:只许 `.memory/extended/` 下的扁平 `.md` 文件名。
|
|
|
|
拒斜杠 / `..` / dotfile / 非 .md。配合调用处 resolve 再兜一层子树校验。
|
|
"""
|
|
if not name or "/" in name or "\\" in name or name.startswith("."):
|
|
return False
|
|
return bool(_EXTENDED_NAME_RE.match(name))
|
|
|
|
|
|
def memory_view(workspace_dir: Path, user_id: UUID) -> Dict[str, Any]:
|
|
"""记忆全貌(只读):core 原文 + extended 列表(filename + description)。
|
|
|
|
一次填满前端弹框。core 给原文(非 strip 前的注入版)让用户看到真实落盘内容。
|
|
"""
|
|
return {
|
|
"core": _load_core(workspace_dir, user_id),
|
|
"extended": [
|
|
{"filename": name, "description": label}
|
|
for label, name in _extended_index(workspace_dir, user_id)
|
|
],
|
|
}
|
|
|
|
|
|
def read_extended_file(
|
|
workspace_dir: Path, user_id: UUID, filename: str
|
|
) -> Optional[str]:
|
|
"""读单篇 extended 原文;文件名非法 / 越界 / 不存在 → None(调用方转 404)。"""
|
|
if not _is_safe_extended_name(filename):
|
|
return None
|
|
ext_dir = (_memory_dir(workspace_dir, user_id) / "extended").resolve()
|
|
target = (ext_dir / filename).resolve()
|
|
if ext_dir not in target.parents or not target.is_file():
|
|
return None
|
|
try:
|
|
return target.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeDecodeError):
|
|
return None
|