记忆与状态管理

API 是无状态的，「记忆」全靠你维护的 messages 数组。会话一长，三个问题必然出现：塞不下、变贵、变慢。本篇讲工程解法。

上下文管理的运行时原理（为什么压缩、记忆分层）在 Harness 上下文工程篇；本篇聚焦应用侧怎么实现。

一、全景：四层记忆

工作记忆    当前 messages 数组（模型唯一看得见的）
会话记忆    落库的完整历史（可恢复、可翻页）
压缩记忆    历史摘要（老对话的代理表示）
长期记忆    跨会话事实库（用户偏好、关键事件）

层	存储	生命周期	典型实现
工作记忆	内存	单次任务	messages 列表
会话记忆	Redis/DB	单会话	按 session_id 存取
压缩记忆	会话记录里	会话内滚动	LLM 生成摘要
长期记忆	向量库/DB	永久	embedding 检索注入

二、会话存储：从内存到 Redis

多实例部署（或进程重启）下，内存存历史必丢。会话状态外置：

import json, redis

r = redis.Redis()

class SessionStore:
    def __init__(self, session_id: str):
        self.key = f"session:{session_id}"

    def load(self) -> list[dict]:
        raw = r.get(self.key)
        return json.loads(raw) if raw else []

    def save(self, messages: list[dict]):
        r.set(self.key, json.dumps(messages, ensure_ascii=False))
        r.expire(self.key, 60 * 60 * 24 * 7)   # 7 天过期

    def append(self, message: dict):
        messages = self.load()
        messages.append(message)
        self.save(messages)

生产注意点：

□ key 设计带用户维度：session:{user_id}:{session_id}，便于隔离与审计
□ 大会话（几百 KB）别每轮全量覆写：用 Redis list RPUSH 追加
□ 工具结果里的大块文本，落库前先截断（日志膨胀比想象快）
□ 结构化存储（Postgres JSONB）比 Redis 更适合审计与回放场景

三、上下文压缩：滑动窗口 + 摘要

上下文逼近上限时的标准三步：

1. 保留：系统提示 + 最近 N 轮（模型正在处理的活儿）
2. 摘要：更早的历史 → 压成一段 summary（LLM 生成）
3. 重组：[system, summary_as_context, 最近 N 轮, 当前输入]

实现：

COMPRESS_AT = 40_000      # token 阈值（留 buffer，别顶格）

async def compress_if_needed(messages: list[dict], system: str) -> list[dict]:
    if count_tokens(messages) < COMPRESS_AT:
        return messages

    old, recent = messages[:-6], messages[-6:]   # 最近 3 轮

    summary_resp = await call_llm(system="你是对话压缩器。把历史对话压缩成要点，保留：用户目标、已做决定、关键事实（数字/名称/路径）、未完成事项。用条目列表。",
                                  messages=[{"role": "user", "content": render(old)}])
    summary = summary_resp.text

    return [
        {"role": "user", "content": f"[此前对话摘要]\n{summary}"},
        *recent,
    ]

摘要提示词的取舍是关键：

必须保留：任务目标、约束条件、关键数字与路径、已确认的决定
可以丢弃：寒暄、试错过程、工具返回的原始大段数据
容易翻车：摘要丢了某个约束 → 模型「忘了」用户说过的话 → 用户爆炸
对策：约束类信息单独结构化保存（见长期记忆），不依赖摘要

四、长期记忆：跨会话记住用户

方案 A：事实抽取（结构化，适合明确偏好）

class UserFact(BaseModel):
    fact: str = Field(description="关于用户的持久事实")
    category: Literal["preference", "profile", "constraint"]

# 会话结束时跑一次抽取，去重后入库
facts = await extract_structured(conversation, schema=UserFact)
# "用户偏好简洁回复" / "用户的项目用 Python 3.12"

下次会话启动时注入：

[用户画像]
- 偏好简洁、直接给结论
- 技术栈：Python 3.12 / FastAPI
- 时区：UTC+8

方案 B：向量检索（语义记忆，适合海量历史）

# 写入：把值得记住的对话/结论 embedding 后入库
await vec_store.upsert(id, text=chunk, vector=await embed(chunk))

# 读取：新任务来时，检索相关记忆注入上下文
hits = await vec_store.search(await embed(user_input), top_k=3)
context = "\n".join(h.text for h in hits)

选型：事实抽取先行（便宜、可控、够用），语义检索等记忆规模真的上来了再加。RAG 全流程详见 RAG 教程——长期记忆本质上就是「对记忆做 RAG」。

方案 C：文件记忆（agent 自己维护）

编码类 agent 常用：给 agent 一个 write_memory 工具，让它自己把重要事实写进 MEMORY.md，下次会话自动加载。简单粗暴，效果出奇地好——Claude Code 的 memory 机制就是这个思路。

五、并发会话的坑

# 两个请求同时读写同一 session → 历史互相覆盖 / 消息错乱
# 解法：按 session_id 加锁（单机 asyncio.Lock / 分布式 Redis SETNX）

LOCKS: dict[str, asyncio.Lock] = {}

async def handle_turn(session_id: str, user_input: str):
    async with LOCKS.setdefault(session_id, asyncio.Lock()):
        store = SessionStore(session_id)
        messages = store.load()
        messages.append({"role": "user", "content": user_input})
        resp = await run_agent_loop(messages)
        store.save(messages)
        return resp

更彻底的方案：同一会话内的消息改成队列串行处理，用户连续发三条消息时按序执行，避免上下文竞争。

六、记忆策略速查表

场景	推荐组合
一次性任务 agent（CI 审查 bot）	无记忆，任务级上下文即可
客服 agent（单会话）	会话存储 + 滑动窗口压缩
个人助手（跨会话）	+ 事实抽取注入画像
编码 agent（长任务）	文件记忆 + 会话存储（可恢复）
海量用户历史	+ 向量检索记忆

本篇小结

• 四层记忆：工作 / 会话 / 压缩 / 长期，按需叠加，从会话存储起步
• 压缩 = 保留最近 N 轮 + 老历史摘要素要；约束类信息结构化保存别进摘要
• 长期记忆先事实抽取后向量检索；agent 自维护文件是被验证的轻量方案
• 并发会话必须按 session 加锁或排队

下一篇：多 Agent 编排模式。