77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
"""双层记忆: `workspace/memory/`。
|
|
|
|
core.md —— 注 system prompt,每次都看到。装稳定事实
|
|
(用户偏好 / 常用命令 / 项目约定 / 模型 quirk 备忘等)
|
|
extended/<x>.md —— 索引(标题+路径)注 prompt,内容 agent 用 `read` 按需拉。
|
|
装少数任务才用的专题资料(某 API 速查 / 某历史事件等)
|
|
|
|
为什么这样切:
|
|
core 一直挂在上下文里,token 成本固定 ⇒ 只放跨任务高频用的精炼内容
|
|
extended 索引只占几行,内容按需付费 ⇒ 适合大量低频专题
|
|
|
|
memory 是 workspace 级别(不是 task 级别)。同一 workspace 的所有 task 共享。
|
|
SaaS 化(§7)后会按 tenant 隔离 —— 接口不变,只换 storage backend。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import List, Tuple
|
|
|
|
|
|
def _memory_dir(workspace_dir: Path) -> Path:
|
|
return workspace_dir / "memory"
|
|
|
|
|
|
def _read_first_title(p: Path) -> str:
|
|
"""取文件第一个非空 h1/h2 行作为标题;没有就用文件名 stem。"""
|
|
try:
|
|
for raw in p.read_text(encoding="utf-8").splitlines():
|
|
line = raw.strip()
|
|
if line.startswith("#"):
|
|
return line.lstrip("#").strip()
|
|
if line:
|
|
return line[:60]
|
|
except (OSError, UnicodeDecodeError):
|
|
pass
|
|
return p.stem
|
|
|
|
|
|
def _load_core(workspace_dir: Path) -> str:
|
|
p = _memory_dir(workspace_dir) / "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) -> List[Tuple[str, Path]]:
|
|
"""返回 [(title, abs_path), ...],按文件名排序。"""
|
|
ext_dir = _memory_dir(workspace_dir) / "extended"
|
|
if not ext_dir.is_dir():
|
|
return []
|
|
items: List[Tuple[str, Path]] = []
|
|
for p in sorted(ext_dir.glob("*.md")):
|
|
if p.is_file():
|
|
items.append((_read_first_title(p), p.resolve()))
|
|
return items
|
|
|
|
|
|
def memory_block(workspace_dir: Path) -> str:
|
|
"""构造注入 system prompt 的记忆段;两块都空就返回空串。"""
|
|
core = _load_core(workspace_dir)
|
|
ext = _extended_index(workspace_dir)
|
|
if not core and not ext:
|
|
return ""
|
|
|
|
parts = ["\n\n## 记忆 (workspace 级,跨 task 共享)"]
|
|
if core:
|
|
parts.append("\n### Core (常驻 prompt)\n")
|
|
parts.append(core)
|
|
if ext:
|
|
parts.append("\n\n### Extended (按需用 `read` 加载)\n")
|
|
for title, path in ext:
|
|
parts.append(f"- `{path}` — {title}\n")
|
|
return "".join(parts)
|