用户记忆系统的本质是一个主动的、持续的学习过程，其目标是构建一个关于用户的、尽可能简洁而强大的预测模型。

工作记忆(working memory)容量有限但访问快速，用于当前任务的处理;
长期记忆(long-term memory)容量巨大但提取较慢，存储过去的经验和知识。

轨迹(Trajectory)
是 Agent 实例的工作记忆，记录了该实例从创建到当前时刻的所有事件，按时间顺序排列，形成一个完整的、不可变的事件序列。
提供了 Agent 决策所需的即时上下文

用户长期记忆(User Long-Term Memory)
是跨会话、跨实例的持久化存储，通常以键值对的形式存在。用户偏好，历史摘要。通过工具调用读取和更新。

业务状态(Business State)
是开发者定义的高层状态抽象，用于表示任务的逻辑阶段(如“需要澄清”、“处理请求中”“等待付款”“请求完成”)。
这不同于框架内部的实例生命周期状态(如“运行中”“等待中”)，而是从业务逻辑角度对任务进度的总结。
框架会将当前业务状态注入到 LLM 的输入上下文中，帮助 LLM 更好地理解任务进展。

记忆系统设计

Advanced JSON Cards 模式：情境知识
从信息存储到知识管理。这种模式的核心创新在于每个记忆卡片不仅记录事实信息本身，还加入了信息来源的叙事背景(backstory)、信息主体的身份标识(person)、主体与用户的关系(relationship)和时间戳

为了消除歧义，明确作用范围

评估

基础回忆
多会话检索
主动服务

压缩

重要性评分筛选（访问频率，时间衰减，情感强度，独特性）冲突的检测和解决
聚类代表
抽象泛化

SKILL动态提示词

算是一套模板，描述了某一类工作路线或者工作方式，和工具函数还是不一样的。

上下文压缩

三层压缩, 激进程度递增:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
Every turn:
+------------------+
| Tool call result |
+------------------+
        |
        v
[Layer 1: micro_compact]        (silent, every turn)
  Replace tool_result > 3 turns old
  with "[Previous: used {tool_name}]"
        |
        v
[Check: tokens > 50000?]
   |               |
   no              yes
   |               |
   v               v
continue    [Layer 2: auto_compact]
              Save transcript to .transcripts/
              LLM summarizes conversation.
              Replace all messages with [summary].
                    |
                    v
            [Layer 3: compact tool]
              Model calls compact explicitly.
              Same summarization as auto_compact.

工作原理

  1. 第一层 – micro_compact: 每次 LLM 调用前, 将旧的 tool result 替换为占位符。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    def micro_compact(messages: list) -> list:
        tool_results = []
        for i, msg in enumerate(messages):
            if msg["role"] == "user" and isinstance(msg.get("content"), list):
                for j, part in enumerate(msg["content"]):
                    if isinstance(part, dict) and part.get("type") == "tool_result":
                        tool_results.append((i, j, part))
        if len(tool_results) <= KEEP_RECENT:
            return messages
        for _, _, part in tool_results[:-KEEP_RECENT]:
            if len(part.get("content", "")) > 100:
                part["content"] = f"[Previous: used {tool_name}]"
        return messages
    这里 tool_results 是一个包含元组（Tuples）的列表，每个元组恰好有 3 个元素：(msg_idx, part_idx, part)。在 Python 中，字典（Dictionary）和列表（List）都是可变对象（Mutable Objects）。当你把一个列表里嵌套的字典拿出来放到一个新的列表（如 tool_results）里面时，你并没有“复制”或者“克隆”这个字典，你只是创建了一个指向原内存地址的引用（指针）。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    [
        (
            0,  # msg_idx: 外层 messages 列表中的索引
            0,  # part_idx: 当前消息的 content 列表中的索引
            {    # part: 实际的 Tool Result 字典对象
                "type": "tool_result",
                "tool_use_id": "toolu_01",
                "content": "cat file.txt 的超长输出内容..."
            }
        ),
  2. 第二层 – auto_compact: token 超过阈值时, 保存完整对话到磁盘, 让 LLM 做摘要。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    def auto_compact(messages: list) -> list:
        # Save transcript for recovery
        transcript_path = TRANSCRIPT_DIR / f"transcript_{int(time.time())}.jsonl"
        with open(transcript_path, "w") as f:
            for msg in messages:
                f.write(json.dumps(msg, default=str) + "\n")
        # LLM summarizes
        response = client.messages.create(
            model=MODEL,
            messages=[{"role": "user", "content":
                "Summarize this conversation for continuity..."
                + json.dumps(messages, default=str)[:80000]}],
            max_tokens=2000,
        )
        return [
            {"role": "user", "content": f"[Compressed]\n\n{response.content[0].text}"},
            {"role": "assistant", "content": "Understood. Continuing."},
        ]
  3. 第三层 – manual compact: compact 工具按需触发同样的摘要机制。
  4. 循环整合三层:
    1
    2
    3
    4
    5
    6
    7
    8
    9
    def agent_loop(messages: list):
        while True:
            micro_compact(messages)                        # Layer 1
            if estimate_tokens(messages) > THRESHOLD:
                messages[:] = auto_compact(messages)       # Layer 2
            response = client.messages.create(...)
            # ... tool execution ...
            if manual_compact:
                messages[:] = auto_compact(messages)       # Layer 3
    完整历史通过 transcript 保存在磁盘上。信息没有真正丢失, 只是移出了活跃上下文

TodoManager

多步任务中, 模型会丢失进度 – 重复做过的事、跳步、跑偏。对话越长越严重: 工具结果不断填满上下文, 系统提示的影响力逐渐被稀释。一个 10 步重构可能做完 1-3 步就开始即兴发挥, 因为 4-10 步已经被挤出注意力了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
+--------+      +-------+      +---------+
|  User  | ---> |  LLM  | ---> | Tools   |
| prompt |      |       |      | + todo  |
+--------+      +---+---+      +----+----+
                    ^                |
                    |   tool_result  |
                    +----------------+
                          |
              +-----------+-----------+
              | TodoManager state     |
              | [ ] task A            |
              | [>] task B  <- doing  |
              | [x] task C            |
              +-----------------------+
                          |
              if rounds_since_todo >= 3:
                inject <reminder> into tool_result

TodoManager 存储带状态的项目。同一时间只允许一个 in_progress。

1
2
3
4
5
6
7
8
9
10
11
12
13
class TodoManager:
    def update(self, items: list) -> str:
        validated, in_progress_count = [], 0
        for item in items:
            status = item.get("status", "pending")
            if status == "in_progress":
                in_progress_count += 1
            validated.append({"id": item["id"], "text": item["text"],
                              "status": status})
        if in_progress_count > 1:
            raise ValueError("Only one task can be in_progress")
        self.items = validated
        return self.render()

在TOOLS列表里加上todo,输入指令给下一轮LLM调用，让它调用todo工具进行更新

1
2
3
4
5
if block.name == "todo":
                    used_todo = True
        rounds_since_todo = 0 if used_todo else rounds_since_todo + 1
       if rounds_since_todo >= 3:
            results.insert(0, {"type": "text", "text": "<reminder>Update your todos.</reminder>"})

持久化任务记忆

 TodoManager 只是内存中的扁平清单: 没有顺序、没有依赖、状态只有做完没做完。真实目标是有结构的 – 任务 B 依赖任务 A, 任务 C 和 D 可以并行, 任务 E 要等 C 和 D 都完成。
 没有显式的关系, 智能体分不清什么能做、什么被卡住、什么能同时跑。而且清单只活在内存里, 上下文压缩 (s06) 一跑就没了。

还有也和上下文压缩有关系：对话流（messages list）不应该用来当“数据库”。这就比如人类没办法全凭心算记住 5 杯奶茶做到哪个步骤了，必须挂个订单流水水单。只要主框架把状态和TOOL results写在了那个 JSON 文件或日志板里，模型什么时候想看具体的，使用 task_get 或者 cat .tasks/task_B_err.log 主动索取（Pull in-demand）即可

解决方案

把扁平清单升级为持久化到磁盘的任务图。每个任务是一个 JSON 文件, 有状态、前置依赖 (blockedBy) 和后置依赖 (blocks)。任务图随时回答三个问题:

  • 什么可以做? – 状态为 pending 且 blockedBy 为空的任务。
  • 什么被卡住? – 等待前置任务完成的任务。
  • 什么做完了? – 状态为 completed 的任务, 完成时自动解锁后续任务。

初始化

1
2
3
4
5
6
7
8
9
10
class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)
        self._next_id = self._max_id() + 1
    def create(self, subject, description=""):
        task = {"id": self._next_id, "subject": subject,"status": "pending", "blockedBy": [],"blocks": [], "owner": ""}
        self._save(task)
        self._next_id += 1
        return json.dumps(task, indent=2)

ex:  task_4.json  {“id”:4, “blockedBy”:[2,3], “status”:”pending”}
2. 依赖解除: 完成任务时, 自动将其 ID 从其他任务的 blockedBy 中移除, 解锁后续任务

1
2
3
4
5
6
def _clear_dependency(self, completed_id):
    for f in self.dir.glob("task_*.json"):
        task = json.loads(f.read_text())
        if completed_id in task.get("blockedBy", []):
            task["blockedBy"].remove(completed_id)
            self._save(task)
  1. 状态变更 + 依赖关联: update 处理状态转换和依赖边。
    1
    2
    3
    4
    5
    6
    7
    8
    def update(self, task_id, status=None,
               add_blocked_by=None, add_blocks=None):
        task = self._load(task_id)
        if status:
            task["status"] = status
            if status == "completed":
                self._clear_dependency(task_id)
        self._save(task)
  2. 四个任务工具加入 dispatch map。task相关任务被驱动的唯一时机，是当 LLM 在它的推理过程（思考）中，认为有必要去持久化一个任务、更新进度，或者查看当前任务列表时。