Agent长线协作的关键：记忆分层与Agent Memory Layer工程实践

Agent 单独执行一次工具调用时看起来已经足够聪明能规划步骤、调用 API、处理报错甚至能在几轮对话里保持上下文连贯。可一旦把任务拉长比如让 Agent 连续跟进一个项目几天、在多个会话间记住用户的真实偏好、跨团队复用上一次的失败经验很多系统会迅速暴露出同一个短板——记忆。真正让 Agent 走向长线协作的难点不是单次推理的准确率而是模型能不能在需要的时候想起该想起的事在不需要的时候屏蔽噪声在旧经验与新事实冲突时做出正确取舍。围绕这个议题AML 首期揭榜出现在 Agent 工程讨论里的频率越来越高。如果没有额外上下文AML 在 Agent 语境里最自然的展开就是 Agent Memory Layer也就是把记忆从 Prompt 拼凑升级成独立、可检索、可压缩、可审计的基础设施层。本文不站队也不预测谁会赢。下面要做的是把问题摊开为什么上下文窗口不等于记忆一个可落地的 memory layer 应该长什么样如何用最小代码实现一套能跑通的基础版本以及上线后需要注意哪些坑。1. 为什么说长线协作的关键是记忆范式而不是模型参数1.1 上下文窗口不是记忆更像一张临时草稿纸很多初学者会默认只要模型上下文足够大Agent 就能记住所有历史。这个理解在实践中会吃亏原因有三层。第一层是容量和成本。把每一轮对话、每一次工具返回值都堆进 Prompttoken 消耗会随着任务时间线性增长。任务拖得越长每一轮推理的开销越大最终会遇到模型上下文上限只能靠截断或粗暴摘要解决问题。第二层是信息密度。Agent 执行一个长任务时90% 的中间过程并不需要原样出现在下一次推理里。真正需要保留的往往是用户的约束、已经完成的步骤、失败的原因、关键决策依据、约定俗成的术语。把这些全部塞进上下文会让模型把注意力分散到大量无关日志上反而降低回答质量。第三层是生命周期。上下文窗口的生命周期通常绑定在同一个会话或同一段连续对话里。Agent 一旦重启服务、切换会话、或者由另一个子 Agent 继续协作上下文就会断掉。真实业务里的“长线协作”恰恰要求 Agent 在会话结束后仍然记得任务背景。所以“上下文窗口等于记忆”是一个短期可用的错觉。长线协作需要的是结构化、可跨会话持久化、能被按需检索的记忆体系。1.2 把记忆拆成工作记忆、情景记忆、语义记忆和程序记忆要想设计记忆层第一步不是急着选向量库而是先把记忆类型分清楚。认知科学里的记忆分类法放到 Agent 工程里同样适用。记忆类型通俗理解Agent 场景举例生命周期典型写入时机工作记忆当前任务正在用的草稿当前用户输入、上一轮工具输出、本轮目标短随会话结束每个请求开始时组装情景记忆过去发生过哪些事昨天处理了一次线上库存同步失败原因是第三方接口超时长可跨会话任务完成后、关键失败后语义记忆从经验中提炼出的抽象规则用户偏好“对外报价默认不含税”项目命名规范采用全小写很长对话中识别到稳定偏好后程序记忆Agent 学会怎么做事调用某支付接口前必须先刷新令牌解析日志时先做格式校验很长可复用工具调用成功或失败后沉淀这种分层不是单纯做概念分类而是为了决定读写策略。程序记忆必须随着工具升级而更新情景记忆需要定期去重和压缩语义记忆需要支持冲突检测和版本覆盖工作记忆则完全不应该落到长期存储里否则会把大量无价值信息灌入检索结果。1.3 AML 在工程体系中的定位把 Agent 的记忆当成服务来建设如果把话题收敛到 AML我的理解是它不是一个单独的“大记忆模型”也不是简单的 Chroma 或 Redis而是介于 LLM 与存储系统之间的一层抽象服务。它负责解决 Agent 在记忆使用过程中反复出现的共性问题记忆如何写入才能避免重复和无意义的中间状态记忆如何检索才能把语义相似、时间有效、类型匹配的内容排到前面记忆如何更新才能处理“用户以前说 A现在改口 B”这类冲突记忆如何遗忘才能防止陈旧的、错误的、敏感的长期记忆持续污染后续决策记忆如何隔离才能保证 Agent A 的经验不会串到 Agent B 的上下文里。没有这层抽象你也能通过大量 Prompt 工程实现“假记忆”。但所有逻辑都会散落在业务代码里每接入一个新 Agent 就要重写一遍。只要团队里出现第二个 Agent记忆的写入路径、命名规则、存储位置就会开始失控。这也是 Agent Memory Layer 这类方案值得关注的根本原因。2. 落地 AML 前先设计统一记忆层和存储模型2.1 一个最小记忆层需要暴露哪些能力设计记忆层时我不建议一开始就写复杂配置。先把接口稳定下来后续换存储后端只是实现替换。一个最小可用的 Agent Memory Layer 至少需要五个方法from typing import Optional, Protocol, TypeAlias class MemoryItem(TypedDict): memory_id: str agent_id: str session_id: str memory_type: str content: str metadata: dict importance: float created_at: str updated_at: str class AgentMemoryLayer(Protocol): async def remember( self, agent_id: str, content: str, session_id: str , memory_type: str episodic, importance: float 0.5, metadata: Optional[dict] None, ) - str: 写入一条记忆返回 memory_id。 ... async def recall( self, query: str, agent_id: Optional[str] None, memory_type: Optional[str] None, top_k: int 5, threshold: float 0.0, ) - list[dict]: 根据当前输入检索相关记忆。 ... async def update( self, memory_id: str, content: Optional[str] None, importance: Optional[float] None, ) - None: 更新已有记忆的内容或重要度。 ... async def forget(self, memory_id: str) - None: 删除某条记忆用于遗忘。 ... async def compact( self, agent_id: str, strategy: str summarize, max_items: int 500, ) - None: 压缩、去重、归档控制记忆库体积。 ...remember负责写入recall负责读取update负责修正forget负责删除compact负责治理。有了这五个方法Agent 主体逻辑就不需要关心记忆到底存在 MySQL、PostgreSQL 还是对象存储里。这里还要强调一个容易被忽视的设计点一条记忆最好不要只存纯文本。它至少应该携带agent_id、memory_type、importance、created_at等元数据否则后续很难做权限隔离、过期清理和类型过滤。2.2 数据模型不只是一个 Embedding 字段很多刚接触向量检索的同学会把记忆设计成一张只有id, content, embedding三列的表。这个方案在单机 Demo 里能运行但真实业务里很快就会遇到问题你没法单独查“某个用户上周产生的语义记忆”也没法快速清理“某个 Agent 的所有临时记录”。更合理的最小数据模型应该类似这样CREATE TABLE IF NOT EXISTS memory_items ( id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, session_id TEXT NOT NULL, memory_type TEXT NOT NULL, content TEXT NOT NULL, metadata_json TEXT NOT NULL DEFAULT {}, embedding_json TEXT, importance REAL NOT NULL DEFAULT 0.5, access_count INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_mem_agent_type ON memory_items(agent_id, memory_type); CREATE INDEX IF NOT EXISTS idx_mem_updated ON memory_items(updated_at);其中agent_id是记忆的空间隔离维度memory_type是记忆的场景维度importance是记忆的重要度updated_at是时效性维度。content是给模型看的自然语言embedding_json是给检索用的向量表示。每条记忆既可以走向量相似度检索也可以按元数据做条件过滤两者组合才能把检索精度做上去。实际写入时一条记忆可能长这样{ id: mem-9f4c2a1e, agent_id: order-agent-prod, session_id: session-20250101-abc, memory_type: semantic, content: 客户 A 所在公司要求所有报价单必须包含增值税明细且默认不接受月结 60 天。, metadata: { source: user, source_url: , embedding_model: text-embedding-3-small, scope: customer-a }, importance: 0.85, created_at: 2025-01-01T09:30:00Z, updated_at: 2025-01-01T09:30:00Z }注意metadata里记录embedding_model。这个字段很容易被忽略但它在排查检索异常时非常重要。如果写入时用的向量模型和查询时不一致余弦相似度会明显失真表现为“相关记忆却检索不到”。2.3 后端存储选择向量库不一定解决所有检索问题选择合适的存储需要先看记忆的访问模式。存储方式优点缺点适合场景SQLite JSON零依赖方便本地调试不适合高并发和超大数据量单机 Demo、工程师本地验证PostgreSQL pgvectorSQL 生态完整能和业务表联合过滤海量向量检索能力不如专用库中小规模生产环境Redis reducers低延迟适合高频缓存持久化与复杂查询较弱会话级工作记忆图数据库能表达实体间关系写入和开发成本高多跳关系记忆、知识图谱对象存储 向量索引成本低容量大对查询链路要求更高大规模记忆慢速归档一个常见的误区是“选了向量库就等于解决了记忆”。实际上向量相似度擅长语义召回却不擅长精确过滤。比如“找出上周五失败的库存同步任务”需要时间字段做范围过滤“只查这个用户的偏好”需要agent_id和memory_type做精确匹配。所以生产环境通常是混合结构过滤条件交给关系型字段或标签索引相似度排序交给向量索引最终再融合一个打分函数。如果是本地起步先使用 SQLite 已经足够。它能验证写入、读取、过滤、删除全链路等数据量上来后再平滑切到 PostgreSQL pgvector 或其他向量数据库。3. 一个可运行的最小记忆层SQLite Embedding 双路检索3.1 环境准备和目录结构这里用一个最简 Python 示例说明实现思路。它不依赖重型框架只要本地安装 Python 3.10 以上版本并确保可选安装一个 embedding 服务。如果暂时不想接外部接口可以先使用embedding_fnNone退化成关键词匹配 重要度排序功能链路依然完整。环境清单如下项目要求说明Python3.10支持 dictSQLite内置用于本地持久化OpenAI SDK可选用于生成 embedding也可以换其他 SDK业务数据自定义建议先用 20 条左右事件测试建议按下面目录安排代码后续扩展时比较清晰agent-memory-demo/ requirements.txt memory_layer.py agent_loop.py eval_scripts/ recall_eval.py examples/ golden_cases.json3.2 实现 MemoryLayer 核心类先创建一个memory_layer.py里面实现一个基于 SQLite 的最小记忆层。import json import math import sqlite3 import time import uuid from typing import Callable, Optional def cosine_similarity(a: list[float], b: list[float]) - float: dot sum(x * y for x, y in zip(a, b)) norm_a math.sqrt(sum(x * x for x in a)) norm_b math.sqrt(sum(x * x for x in b)) if norm_a 0 or norm_b 0: return 0.0 return dot / (norm_a * norm_b) class AgentMemoryLayer: def __init__( self, db_path: str, embedding_fn: Optional[Callable[[str], list[float]]] None, ) - None: self.conn sqlite3.connect(db_path) self.conn.row_factory sqlite3.Row self.embedding_fn embedding_fn self._init_schema() def _init_schema(self) - None: self.conn.executescript( CREATE TABLE IF NOT EXISTS memory_items ( id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, session_id TEXT NOT NULL, memory_type TEXT NOT NULL, content TEXT NOT NULL, metadata_json TEXT NOT NULL DEFAULT {}, embedding_json TEXT, importance REAL NOT NULL DEFAULT 0.5, access_count INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_mem_agent_type ON memory_items(agent_id, memory_type); CREATE INDEX IF NOT EXISTS idx_mem_updated ON memory_items(updated_at); ) self.conn.commit() def remember( self, agent_id: str, content: str, session_id: str , memory_type: str episodic, importance: float 0.5, metadata: Optional[dict] None, ) - str: memory_id fmem-{uuid.uuid4().hex[:12]} now time.strftime(%Y-%m-%dT%H:%M:%SZ, time.gmtime()) embedding_json None if self.embedding_fn is not None: embedding self.embedding_fn(content) embedding_json json.dumps(embedding) self.conn.execute( INSERT INTO memory_items (id, agent_id, session_id, memory_type, content, metadata_json, embedding_json, importance, access_count, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( memory_id, agent_id, session_id, memory_type, content, json.dumps(metadata or {}), embedding_json, importance, 0, now, now, ), ) self.conn.commit() return memory_id这段代码的关键是写入时给每条记忆添加agent_id、memory_type、importance和embedding_json。agent_id是为了隔离不同 Agent 的记忆memory_type是为了在检索阶段缩小范围embedding_json是可选字段没接 embedding 服务时直接为空。接下来是检索方法。检索不是简单调一次向量查询就结束而是要经过“元数据过滤 - 粗排序 - 访问计数更新”三个阶段。def recall( self, query: str, agent_id: Optional[str] None, memory_type: Optional[str] None, top_k: int 5, threshold: float 0.0, ) - list[dict]: sql SELECT * FROM memory_items WHERE 1 1 params: list [] if agent_id: sql AND agent_id ? params.append(agent_id) if memory_type: sql AND memory_type ? params.append(memory_type) sql ORDER BY updated_at DESC LIMIT 2000 rows self.conn.execute(sql, params).fetchall() scored: list[tuple] [] if self.embedding_fn is not None: query_embedding self.embedding_fn(query) for row in rows: if not row[embedding_json]: continue doc_embedding json.loads(row[embedding_json]) sim cosine_similarity(query_embedding, doc_embedding) recency self._recency(row[updated_at]) score 0.75 * sim 0.15 * row[importance] 0.10 * recency scored.append((score, row)) else: # 无 embedding 时退化为关键词命中 重要度 keywords [k for k in query.lower().split() if len(k) 1] for row in rows: text row[content].lower() hit_count sum(1 for kw in keywords if kw in text) if hit_count 0: score hit_count / len(keywords) 0.2 * row[importance] scored.append((score, row)) scored.sort(keylambda x: x[0], reverseTrue) top_items [row for row in scored[:top_k] if row[0] threshold] # 命中后更新访问计数用于后续排序和清理 now time.strftime(%Y-%m-%dT%H:%M:%SZ, time.gmtime()) for _, row in top_items: self.conn.execute( UPDATE memory_items SET access_count access_count 1, updated_at ? WHERE id ? , (now, row[id]), ) self.conn.commit() return [ { memory_id: row[id], content: row[content], memory_type: row[memory_type], importance: row[importance], score: round(score, 4), } for score, row in top_items ] def _recency(self, updated_at: str) - float: # 这里按天做简单衰减完整实现可参考 RFC3339 解析 try: ts time.mktime( time.strptime(updated_at, %Y-%m-%dT%H:%M:%SZ)) now time.time() age_days max(0, (now - ts) / 86400) return 1.0 / (1.0 age_days / 7.0) except Exception: return 0.5在没有 embedding 服务时embedding_fnNone走的是关键词命中加重要度排序功能链路已经能通。当接入真正的 embedding 函数后检索就从“字面命中”升级为“语义命中”。3.3 把记忆层接入 Agent 主循环记忆层本身不产生智能真正的价值在于接入 Agent 推理循环。一个最常见的接入方式是每次收到用户输入时先从记忆层检索相关内容拼进 Prompt等 Agent 完成一个有价值节点后再把新经验写回记忆层。class AgentLoop: def __init__(self, llm_client, memory_layer: AgentMemoryLayer, agent_id: str): self.llm_client llm_client self.memory memory_layer self.agent_id agent_id async def handle_input(self, user_input: str) - str: candidates self.memory.recall( queryuser_input, agent_idself.agent_id, memory_typeNone, top_k6, threshold0.35, ) memory_context \n.join( f[{item[memory_type]}] {item[content]} for item in candidates ) messages [ { role: system, content: ( 你是一个有长期记忆的 Agent。以下是与当前任务相关的记忆 只在确实相关时使用不要编造记忆中没有的事实。\n f{memory_context} ), }, {role: user, content: user_input}, ] response await self.llm_client.chat(messages) # 这里只写入关键结果不写入每一条中间日志 if self._is_memory_worthy(response, user_input): self.memory.remember( agent_idself.agent_id, contentself._build_memory_content(user_input, response), memory_typeepisodic, importance0.6, ) return response核心原则是写入记忆不要贪多。只在用户的稳定偏好、工具调用成功/失败、重要任务结果三类事件发生时写入。高频中间日志一旦写入长期记忆后续检索的噪声会迅速上升。4. 参数调节、写入时机和记忆压缩4.1 必须搞明白的四个检索参数记忆层能不能用好往往不是模型决定而是参数设置决定。下面这几个参数在接入时一定要逐个理解。参数含义常见值调大影响调小影响top_k最多进入 Prompt 的记忆条数3 到 10Prompt 更长可能引入噪声容易漏掉关键上下文threshold相关度阈值0.2 到 0.5只保留高相关记忆可能召回不足低相关记忆也会被拼进 Promptimportance记忆重要度参与融合排序0 到 1高重要度记忆难以被新经验覆盖重要经验可能沉底recency衰减周期控制旧记忆多久后降权7 天到 30 天旧记忆长期保留但可能过时新记忆优先但旧经验容易被遗忘在生产系统里不要把threshold设得太高。一旦阈值过高Agent 会频繁出现“明明写过某条记忆检索却为空”的现象。推荐先设一个较低的阈值把日志跑起来再根据检索命中率和下游任务准确率逐步上调。4.2 写入时机决定记忆质量写入时机比写入算法更重要。过早写入会积累垃圾过晚写入又会丢信息。建议按三条规则判断用户显式表达的偏好和禁令立即写入语义记忆重要度给高。工具调用失败并导致重试后成功将失败原因和规避方法写入程序记忆。一个可以被复述、可以复用的任务结果在任务闭环时写入情景记忆。相反的不要写入模型每轮推理过程、工具返回的原始 JSON 大字段、以及包含密钥或敏感 token 的请求体。这些内容一旦进入长期记忆检索时很可能被当成上下文拼给模型造成信息泄露或无效 token 浪费。4.3 长期记忆必须能压缩和遗忘没有压缩和遗忘的记忆层本质上是一个“只进不出的 Context 垃圾桶”。引入一个简单的compact策略就可以把记忆库控制在一个可管理的范围内。def compact_agent_memory(self, agent_id: str, max_items: int 500) - None: rows self.conn.execute( SELECT * FROM memory_items WHERE agent_id ? ORDER BY importance DESC, access_count DESC LIMIT ? , (agent_id, int(max_items * 2)), ).fetchall() keep [] seen_similar {} for row in rows: content row[content] # 只看近似重复可以用 embedding 余弦阈值这里演示用前 20 字 key content[:20] if key in seen_similar: continue seen_similar[key] row[id] keep.append(row[id]) # 只保留去重后 top max_items 条过期未访问的删除 self.conn.execute( DELETE FROM memory_items WHERE agent_id ? AND id NOT IN ({}) .format(,.join([?] * len(keep))), [agent_id] keep, ) self.conn.commit()实际生产中的压缩通常由定时任务执行比如每天凌晨跑一次。更高阶的做法是调用 LLM将大量情景记忆提炼成语义记忆。例如把“昨天支付失败因为余额不足”“今天支付失败因为接口超时”压缩成一条“支付失败时先查余额与上游健康状态”。不过这类压缩不能盲目执行压缩后需要保存源记忆标识方便溯源审计。5. 如何验证长线协作能力真的变强5.1 设计一个跨会话的记忆验证场景给 Agent 加记忆层后最忌讳只看“感觉它记得更清楚了”。要验证需要设计一个冷启动评测集。核心思路是在会话 A 中给出一批任务约束在会话 B 中要求 Agent 完成一个隐藏依赖这些约束的任务并检查结果是否命中。一个最小 golden case 可以长这样[ { agent_id: meeting-agent, session_a: [ 用户说以后安排会议尽量避开每周五下午因为周五需要写周报。, 用户说最近的会议室优选 3 楼靠窗那间。 ], session_b_query: 请帮我安排下一次产品评审会的时间和会议室。, expected: 不能推荐周五下午应优先考虑三楼的靠窗会议室。 } ]测试时会话 A 的内容不能出现在会话 B 的 Prompt 里会话 B 只能依赖 memory layer。这样才能证明记忆确实持久化了而不是上下文恰好没有断开。5.2 用三类基线做横向对比为了证明“AML 记忆层有效”至少要对比三种方案。方案会话 A 记录下来后如何给会话 B预期问题无记忆会话 B 只收到当前输入跨会话约束全部丢失全文回放把会话 A 全部日志塞进会话 B长任务下 token 爆炸噪声多AML 分层记忆检索后只注入相关记忆