为人工智能代理提供外部内存,而非填满上下文窗口
用基于混合向量与图内存的“读取-推理-行动-写作”循环,再加上整合、剪枝、评估及隐私控制机制,来替代金鱼智能体。
来源: Your AI Agent’s Memory Problem: The New Architecture You Need
无状态聊天机器人会在每次会话结束后重置。那些负责编写代码、分析市场或协调工作流的智能体也是如此:当流程结束之时,它们的偏好设置、错误记录以及项目上下文都会消失。解决之道并非不断延长提示词长度,而是将内存视为一级外部子系统,从而使智能体能够在数千次交互中持续学习。
金鱼问题:为何当今的AI智能体会遗忘
可将其称为金鱼式智能体:每次交互都从零开始,无法保留以往的经验、用户偏好或失败记录。如果聘请的精英顾问每次合上笔记本电脑就忘记你的名字、代码库及目标,那么一半的预算将浪费在重新说明任务上。无状态智能体架构也会造成同样的浪费。产品演示因为每次评估时间较短而掩盖了这些问题;但当用户几天后返回,智能体却再次询问它早已收集过的偏好,或是重新提出团队在上次迭代中已拒绝的设计时,问题就会暴露出来。多步骤工具智能体则会进一步增加成本:由于缺乏数据回写功能,每一个漫长的工作流程都不得不重新去发现那些之前已经通过多次运行才了解到的环境特性。
+-------------------------------------------------------------+
| User Session Ends |
| | |
| ▼ |
| +-----------------------------------------------------+ |
| | Stateless LLM Call Clears | |
| | - System Prompts: Reset | |
| | - Session Context: Erased | |
| | - Learned Preferences: Forgotten | |
| +-----------------------------------------------------+ |
| | |
| ▼ |
| Next Run = Day One Again |
+-------------------------------------------------------------+
大型上下文窗口的错觉
团队常常通过在每轮对话中将大量历史记录和文件塞进上下文窗口来伪造“召回”功能。百万级token的窗口确实容易让人产生这种念头,但将窗口用作长期存储其实是一种反模式。立即会出现三种弊端:
- 成本不断上升,因为每次调用都需要重新处理庞大的提示词
- 延迟增加,首次生成token的时间会随着提示词长度的增加而延长
- 上下文噪声,关键信息会被大量文本淹没而无法被识别
更大的窗口有助于实现交互式聊天,但无法替代持久化的状态存储。此外,它们还会让成本和延迟与历史记录的长度挂钩,从而导致使用频繁的用户成为服务成本最高的群体。
从更大上下文转向更智能的上下文
自主系统需要将内存作为外部架构组件,而非临时应对的技巧。这种转变使智能体从无状态函数变为具有状态、能够随时间不断读取、写入和整合信息的结构。下一代智能体工程的重点不再是扩展注意力范围,而是实现能够超越单次请求而持续存在的解耦式内存。
记忆循环:智能体认知的架构蓝图
智能体无法将每一次交互都视为全新的起点。它需要一个能够持续获取、过滤并保留信息的认知循环——这是让大型语言模型从预测工具转变为具有状态的工作节点的基础。
想象一下经验丰富的工程师一天的工作:阅读代码库、规划、实现、提交代码。智能体内存则将这一人类工作流程转化为可重复的循环结构。
读取-推理-行动-写入循环
阅读-推理-行动-写作(RRAW)将执行过程分为四个连续的阶段:
- 阅读——在构建提示词之前,从长期存储中检索相关的历史记录、偏好设置及过往决策
- 推理——模型将当前请求与检索到的信息结合以制定计划
- 行动——根据该计划调用工具、API或数据库
- 写作——整合结果与反馈,然后将其重新索引到内存中
内存写入操作应异步执行。在向量或图结构更新时阻塞面向用户的响应会带来难以接受的延迟。一种实用的方案是立即通过原因/操作路径确认用户的操作,将原始交互数据放入队列以执行整合任务,然后让工作进程通过重试和死信处理机制在后台更新索引。
架构数据流
实现该循环意味着要明确数据在用户、协调器、模型和存储之间的传输方式:
[User Input]
│
▼
[Agent Orchestrator] <─────────────────────────────┐
│ │ (Loopback)
├─────────────────► [Memory Retrieval] │
│ │ │
▼ ▼ │
[LLM Prompt Assembler] <─── [Context] │
│ │
▼ │
[Large Language Model (LLM)] │
│ │
▼ │
[Action Executor] │
│ │
▼ │
[Memory Writeback Module] ─────────────────────────┘
│
▼
[(Episodic / Semantic / Procedural Stores)]
智能记忆的三大支柱
单一的数据库会导致检索混乱和架构僵化。应将长期记忆分为三种角色:
- 情景记忆 — 按时间顺序排列的交互日志;即叙事性自传
将大脑与记忆解耦
其工程上的优势在于将计算部分(大语言模型)与状态存储部分分离。这样既能避免上下文窗口的限制,又能减少频繁微调的成本。随着历史数据的增加,存储部分可以水平扩展,而模型则仍保持为有界且基本无状态的执行器。团队无需重新训练聊天模型即可更换嵌入模型或图结构,还可以按租户划分内存,从而避免某个客户的使用增长影响其他客户的查询效率。
构建记忆存储:混合技术栈
人类认知将情景回忆与语义知识区分开来。将二者混存于同一系统中会导致系统要么过于模糊,要么过于僵化。因此,生产环境通常会将用于关联检索的向量数据库与用于处理确定性关系的图数据库结合使用。
用于语义回忆与情景回忆的向量数据库
Pinecone、Milvus、Weaviate等平台用于存储事件和文档的嵌入向量。余弦相似度等距离度量指标可用于筛选出相关的记忆内容:
import numpy as np
from openai import OpenAI
client = OpenAI()
def get_embedding(text: str) -> list[float]:
"""Generates a 1536-dimensional vector embedding for semantic search."""
response = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
# Store an episodic memory with its vector representation
episodic_memory = "Met with Sarah yesterday. She greenlit the budget for Project Titan."
memory_vector = get_embedding(episodic_memory)
# This vector allows us to find this memory later using natural language queries
# like 'What did we agree on with Sarah?' even if 'budget' is omitted.
在“上个季度我们讨论了哪些关于云迁移的内容?”这类开放性问题上,向量搜索表现优异。但当需要精确的结构化事实作为答案时,它的效果就会减弱——概率性相似度并不能提供确定性保证。相似度工具适合用于“提醒我关于迁移的讨论内容”,却不适用于那些必须得出唯一正确结果的薪资审批或访问控制关系。
用于确定性知识映射的图数据库
“谁负责泰坦项目?”这样的查询可能会返回一份同时提及相关人员的文档,但却无法证明他们之间的关联。像Neo4j这样的图数据库将实体视为节点,将关系视为边,从而通过遍历实现确定性:
(Project: Titan) <-[:LEADS]- (Employee: Sarah) -[:REPORTS_to]-> (VP: Marcus)
系统会沿着经过验证的LEADS路径进行查询,而非依靠相似度来猜测。
混合解决方案:协同式的向量到图检索
生产内存通常负责协调这两个引擎。向量模型将模糊的意图转化为实体ID,随后图结构会返回精确的关系与边界信息:
# Real-world query: "Who is the lead on that document-indexing project?"
def hybrid_memory_lookup(user_query: str) -> dict:
# Step 1: Query Vector DB to find the most contextually relevant project
query_vector = get_embedding(user_query)
# Mock Vector DB search returning the closest matching entity ID
matched_project_id = "project_titan_404" # Found via match on "document-indexing"
# Step 2: Query Graph DB using the strict entity ID to find the exact lead
# Cypher Query: MATCH (p:Project {id: $id})<-[:LEADS]-(e:Employee) RETURN e.name
graph_record = mock_neo4j_driver.run(
"MATCH (p:Project {id: $id})<-[:LEADS]-(e:Employee) RETURN e.name",
id=matched_project_id
)
return {
"project_id": matched_project_id,
"lead_name": graph_record["e.name"]
}
# This hybrid flow eliminates vector hallucination by anchoring
# fuzzy natural language inputs to hard, deterministic graph relationships.
应先从用户查询中提取实体,并以这些ID作为图结构的起点,而非将原始文本直接放入类似Cypher的查询语句中。通过结合关联存储与结构化存储,智能体能够在保持对话上下文的同时确保对企业事实的精确把握。在实际应用中,协调器可能会先执行一次低成本的向量查询来获取候选实体ID,再对照图结构进行验证,最后才将经过核实的三元组组合注入提示信息中。
从原始数据到可用内存:整合流程
将每一条对话内容直接写入向量索引会导致内存膨胀和检索噪声——桌上堆满来自各类会议的便签,甚至包括问候语。想要找到三个月前的决策几乎不可能。若不进行整合,检索延迟会上升,前几条结果中充斥着问候语,模型还会开始引用过时或矛盾的片段。异步整合流程能将原始日志转化为高质量的知识:
[Raw Interaction Data] → [Extraction] → [Synthesis] → [Pruning] → [Long-Term Memory]
记忆整合的三个阶段
在关键请求路径上——通过cron任务或会话结束事件——分三个阶段进行整合:
- 提取——从冗长的对话内容中筛选出事实、偏好和决策信息
- 合成——与现有的长期记录合并,解决冲突,更新过时的字段
利用结构化输出合成对话
提取与合成过程需要遵循模式约束的生成规则。类似Pydantic风格的模型能使数据以适合数据库存储的形式被写入:
import os
from typing import List
from pydantic import BaseModel, Field
from openai import OpenAI
# Define the target structure for consolidated memory
class MemoryItem(BaseModel):
fact: str = Field(description="A single, atomic statement of fact about the user or their environment.")
category: str = Field(description="The functional domain (e.g., tech_stack, preference, constraint).")
importance: int = Field(description="A subjective importance rating from 1 (low) to 5 (high).")
class ConsolidatedMemory(BaseModel):
memories: List[MemoryItem]
def synthesize_transcript(transcript: str) -> ConsolidatedMemory:
"""
Parses a raw transcript and extracts high-signal, deduplicated facts.
This structured parsing prevents raw noise from polluting the vector store.
"""
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
prompt = f"""
Analyze the following chat transcript between a user and an AI.
Extract key facts, user preferences, and technical constraints.
Ignore conversational noise, greetings, and transition phrases.
Transcript:
{transcript}
"""
# Use structured outputs to guarantee the response matches the Pydantic schema
completion = client.beta.chat.completions.parse(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are an expert data consolidation engine."},
{"role": "user", "content": prompt}
],
response_format=ConsolidatedMemory,
)
return completion.choices[0].message.parsed
# Example Usage
raw_chat = """
User: Hey! I'm starting a new web project. I want to use FastAPI and PostgreSQL.
AI: That's a great stack! Do you want to use Docker?
User: Yes, Docker is a must. And remember, I hate using MongoDB, so let's avoid that completely.
"""
structured_output = synthesize_transcript(raw_chat)
for item in structured_output.memories:
print(f"[{item.category.upper()}] Fact: {item.fact} (Importance: {item.importance}/5)")
算法剪枝与相关性评分
并非所有合成后的记录都值得永久保留。可通过重要性、最新性和出现频率等指标来评估其效用:
效用 = (重要性 * 访问次数) / (1 + (衰减率 * 自上次访问以来的天数))
低于阈值的数据会被删除,以保持索引的规模适中且具有相关性。在业务低峰期进行批量整合,从而降低令牌消耗,并避免对高频使用的索引造成写入竞争。应定期将大语言模型给出的重要性评分与人工审核样本进行对比校准,确保系统不会过度保留琐碎信息,也不会遗漏那些罕见但至关重要的决策。
实际应用场景
无状态模型就像患有健忘症的顾问——单独工作时表现出色,但日复一日的运作却令人疲惫不堪。持久化内存则能让生成器变成能够持续创造价值的合作伙伴:
- AI同事:能够记住代码库结构、API协议以及以往的设计决策
- 客服代理:能在不同会话之间记住产品相关信息和用户情绪状态
- CRM助手:能够追踪关系中的细微变化,从而实现更有效的跟进
根据需求选择检索路径:支持性功能可使用会话级本地状态,而代码架构则需依赖深层结构上下文。应避免使用单一的存储机制。支持类机器人通常需要最近的几轮会话记录及账户元数据;编码类智能体则需要代码库结构图和问题解决记录摘要。将不同类型的查询定向到不同的存储系统——或具有不同保留策略的不同数据集合中——有助于保持各索引的完整性。
生产环境规范:从原型到可靠系统
仅仅连接向量数据库是不够的。一种常见的故障模式是只读内存——所有工具日志和回复都被存储下来,导致在海量数据干扰下检索功能失效。生产环境中的系统需要具备包含过滤、整合以及定期归档功能的严格RRAW循环;否则上下文噪声和向量漂移就会占据主导地位。可观测性指标应包括整合任务的队列深度、数据裁剪率,以及在架构或嵌入模型变更后命中率下降时的警报阈值。
利用评估指标量化内存价值
应像管理其他服务一样监控内存使用情况:
- 内存命中率(MHR)——实际用于生成答案的检索片段所占比例
- 响应质量得分(RQS)——用于评估基于内存的回复与零状态回复的质量差异
成本、延迟与隐私管理
每次内存访问都会耗费时间和金钱。应将数据整合操作设为异步进行,为数据检索设置严格的超时时间,以避免缓慢的查询阻塞主循环,并将内存存储视为敏感数据加以保护:
import asyncio
import logging
from typing import List
logging.basicConfig(level=logging.INFO)
async def fetch_memories_from_db(user_id: str, query: str) -> List[str]:
# Simulate a vector database retrieval latency
await asyncio.sleep(0.1)
return ["User prefers Python for data engineering tasks."]
async def retrieve_memory_safely(
user_id: str,
query: str,
timeout_seconds: float = 0.15
) -> List[str]:
"""
Retrieves memories asynchronously with a strict timeout guardrail.
This prevents slow vector DB lookups from blocking the main agent response loop.
"""
try:
# Enforce a strict latency budget using asyncio.wait_for
memories = await asyncio.wait_for(
fetch_memories_from_db(user_id, query),
timeout=timeout_seconds
)
return memories
except asyncio.TimeoutError:
logging.warning(f"Memory timeout for user {user_id}. Falling back to empty memory.")
# Fallback to preserve user experience latency
return []
# Example execution of the guarded memory retrieval
if __name__ == "__main__":
memories = asyncio.run(retrieve_memory_safely("usr_90210", "Get programming language"))
print(f"Retrieved Memories: {memories}")
在数据导入时进行匿名化处理,实施角色基访问控制,确保代理A无法读取代理B的内存数据,并符合GDPR/CCPA的相关要求。绝不能嵌入未经处理的原始用户文本,必须在建立索引之前对个人信息进行脱敏处理。
首次正式上线时的操作检查清单
在建立完善的分类体系之前,先传输薄的垂直数据切片。从阶段性会话日志以及用于存储用户持久偏好的语义集合开始,只有当产品问题需要确定性关联时才添加用于组织实体的小型图结构,并从第一天起就对MHR和RQS进行监控。当内存性能不佳时,优先选择超时处理和降级响应而非阻断聊天。需记录每个租户的数据保留策略,演练从备份中恢复的数据流程,并明确谁有权限根据请求清除用户的记忆。这些习惯比选择最流行的向量数据库品牌更为重要,因为内存系统在概念层面出现故障之前,早已会在实际运行中出现问题。一个拥有明确服务水平指标、数据脱敏功能以及定期清理任务的普通技术组合,其表现会优于那种没人监控、命中率逐渐下降的时髦数据库。
架构优化的目标
该架构更重视可演化性与状态保持能力,而非一次性提示的巧妙性。大语言模型是运行时引擎,而外部化存储则构成系统状态。通过结构化的反馈循环,智能体能够在每个迭代中不断优化发展路径——从临时的命令行工具逐步演变为持续运行的生产系统。
它还解决了成本、延迟和相关性问题:迫使模型在庞大的上下文海中搜索会消耗大量的注意力资源与Token。外部记忆可以预先过滤这些海量信息。真正的智能体价值并非来自上下文窗口的大小,而更多取决于为模型提供数据的检索系统的精准度。那些在混合记忆、信息整合、评估机制以及隐私保护方面投入力量的团队,其智能体能够随着使用不断改进,而不会因自身历史数据而性能下降。若缺乏这些控制措施,历史数据反而会成为负担:提示词需要更强烈,响应速度更慢,用户也会逐渐不再信任那些每周都会自相矛盾的助手。只有精心设计的外部记忆,才能让智能体产品积累优势而非杂乱信息。