Agent 的状态、短期记忆与持久化：让任务可以继续执行

前面我们已经用一个受限循环实现了单 Agent：模型提出动作，程序执行工具，再把观察结果交回下一轮。但这个循环一旦结束，状态也随进程消失；如果服务重启，用户就只能从头开始。本篇只聚焦一个核心问题：如何设计 Agent 的状态、短期记忆，以及把必要信息持久化，让任务能够暂停后继续。示例不调用在线模型，使用确定性的规则决策器验证恢复流程。

先区分状态、记忆与持久化

这三个词经常被混用，但职责不同。

状态（state）是一次任务当前的完整事实，例如任务 ID、目标、当前阶段、已完成的工具结果和更新时间。
短期记忆（working memory）是帮助当前决策的有限上下文，通常包括最近消息、最近观察或经过压缩的摘要。
持久化（persistence）是把状态写入进程之外的介质，例如 JSON 文件、关系数据库或键值存储，使重启后仍能恢复。

状态是业务对象，记忆是给决策器使用的视图，持久化是保存状态的手段。把全部聊天历史当作状态并无限追加，既会让上下文变长，也会增加隐私和恢复成本。应先明确“任务恢复必须知道什么”，只保存这些字段。

为任务定义可保存的状态

下面的示例把任务拆成三个阶段：pending 表示尚未处理，done 表示已经完成，failed 表示执行失败。history 只记录必要的事件，而不是保存模型的所有原始输入。使用数据类可以让字段和默认值更清楚。

from dataclasses import asdict, dataclass, field
import json
from pathlib import Path


@dataclass
class TaskState:
    task_id: str
    goal: str
    status: str = "pending"
    result: str | None = None
    history: list[str] = field(default_factory=list)

    def save(self, path: Path) -> None:
        path.write_text(
            json.dumps(asdict(self), ensure_ascii=False, indent=2),
            encoding="utf-8",
        )

    @classmethod
    def load(cls, path: Path) -> "TaskState":
        data = json.loads(path.read_text(encoding="utf-8"))
        return cls(**data)

save 和 load 只负责序列化，不负责决定任务能否执行。真实项目中还应加入状态版本号，例如 schema_version，这样字段变化时可以迁移旧数据。加载外部文件后也不能盲信内容：应检查任务 ID、状态值、字符串长度和权限范围。

短期记忆应该有限且可重建

Agent 不一定需要把完整历史传给每一次决策。可以把事件分为用户目标、工具观察和最终结果，并提供一个只取最近若干条的视图：

def recent_memory(state: TaskState, limit: int = 3) -> list[str]:
    if limit < 1:
        raise ValueError("limit 必须大于 0")
    return state.history[-limit:]

这个函数返回的是短期记忆视图，而不是另一份需要长期维护的副本。这样做有两个好处：第一，历史增长不会直接导致每次请求都携带全部内容；第二，丢失缓存后仍可以从持久化状态重新构造记忆。若历史很长，可以定期把旧事件压缩成摘要，但摘要也应带来源和更新时间，避免把不确定的模型总结当成事实。

实现一个可恢复的最小任务

为了演示“暂停后继续”，让规则决策器每次只处理一个单词。第一次运行完成两个单词后主动暂停；第二次从文件加载状态，继续处理剩余内容。这里的规则函数只是替代真实模型，重点是状态如何流动。

def next_item(state: TaskState) -> str | None:
    completed = set(state.history)
    for item in state.goal.split():
        if item not in completed:
            return item
    return None


def run_once(state: TaskState, max_items: int = 2) -> None:
    if state.status == "done":
        return
    if max_items < 1:
        raise ValueError("max_items 必须大于 0")

    for _ in range(max_items):
        item = next_item(state)
        if item is None:
            state.status = "done"
            state.result = "已处理：" + "、".join(state.history)
            return
        # 这里代表一次受控的工具或模型动作
        state.history.append(item)

    if next_item(state) is None:
        state.status = "done"
        state.result = "已处理：" + "、".join(state.history)


if __name__ == "__main__":
    path = Path("task.json")
    if path.exists():
        task = TaskState.load(path)
    else:
        task = TaskState(task_id="demo-1", goal="整理 发票 合同 报告")

    run_once(task, max_items=2)
    task.save(path)
    print(task.status, task.history)

第一次执行会把 整理 和 发票 写入 task.json，状态仍是 pending；再次执行时会加载文件，继续处理 合同 和 报告，最终变成 done。这类示例输出取决于实际运行时文件是否已经存在，因此开发时应删除 task.json 后再验证全新流程。文章示例不把文件加入博客仓库；真实应用还要把任务文件放在有访问控制的位置。

持久化时最容易忽略的边界

不要保存密钥。 环境变量、访问令牌和完整授权凭据不属于任务记忆。状态中只保存必要的资源 ID，恢复时由服务端重新读取权限。

保存要尽量原子化。 直接覆盖文件时，进程可能在写到一半时崩溃。简单场景可以先写临时文件，再使用 Path.replace 替换目标文件；多进程场景则应使用数据库事务或专门的存储服务。

恢复前重新检查权限。 用户在任务暂停期间可能已经失去权限。不能因为状态中记录了“允许执行”，恢复时就跳过当前权限检查。

区分可重试动作和已完成动作。 如果工具执行后、状态保存前进程崩溃，重启可能重复执行。对发送通知、扣款等副作用操作，需要幂等键、执行记录或人工确认，而不能只依靠列表去重。

设置过期和容量。 短期记忆应有条数或字数上限，持久化任务应有过期时间和清理策略。日志、工具返回值和用户内容也要脱敏，避免“为了恢复任务”无限保存隐私数据。

常见问题

短期记忆和持久化是不是二选一？ 不是。短期记忆服务于当前一次决策，持久化服务于跨请求恢复；持久化的数据加载后仍可以裁剪成短期记忆。

为什么不直接保存完整消息历史？ 完整历史可能很长、含敏感内容，也可能包含已经失效的指令。优先保存可验证的业务状态，必要时再保存有限的审计事件。

恢复时能不能直接让模型接着聊天？ 可以，但恢复前要重新构造系统约束、检查权限和验证工具结果。历史记录不是安全边界，模型生成的动作仍必须经过程序校验。

只用 JSON 文件是否适合生产？ 单进程、低并发的原型可以使用。并发写入、故障恢复、查询和权限要求增加后，应迁移到支持事务和约束的持久化系统。

小结

Agent 的状态描述任务事实，短期记忆提供有限决策上下文，持久化让状态跨越进程生命周期。可靠的恢复流程不是“把聊天记录存起来”这么简单，而是定义最小状态、限制记忆、校验加载数据，并在恢复时重新检查权限和副作用边界。先用可序列化的纯 Python 状态验证暂停与继续，再根据并发、审计和可靠性要求选择数据库等存储方案。

搜索时光笔记