一、长期记忆的核心设计要素
长期记忆的向量数据库设计需围绕 “经验如何表示、如何存储、如何检索” 三个核心问题，具体要素如下：
|
|
|
|
|---|---|---|
| 经验内容定义 |
|
|
| 向量表示方式 |
|
|
| 元数据设计 |
|
scene（场景标签）、timestamp（时间戳）、priority（优先级）等。
|
| 检索策略 |
|
|
| 记忆演化机制 |
|
|
二、具体设计方案（以 Milvus 为例）
1. 经验数据结构定义
每条长期记忆（经验）需包含 “向量 + 文本内容 + 元数据” 三部分，示例结构如下：
|
|
|
|
|---|---|---|
id |
|
exp_12345）
|
vector |
|
|
content |
|
|
scene |
|
["data_cleaning", "missing_value_processing"]）
|
timestamp |
|
|
priority |
|
|
source |
|
"user_feedback"“task_success”）
|
2. 向量数据库集合（Collection）设计
以 Milvus 为例，创建用于存储长期记忆的集合，需定义字段类型和索引（加速检索）：
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility# 1. 连接Milvus服务connections.connect("default", host="localhost", port="19530")# 2. 定义字段fields = [# 唯一IDFieldSchema(name="id", dtype=DataType.VARCHAR, max_length=50, is_primary=True),# 经验向量（768维，与嵌入模型输出维度一致）FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=768),# 经验文本内容FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=2000),# 适用场景（数组类型）FieldSchema(name="scene", dtype=DataType.ARRAY, element_type=DataType.VARCHAR, max_length=100),# 时间戳FieldSchema(name="timestamp", dtype=DataType.INT64),# 优先级FieldSchema(name="priority", dtype=DataType.INT32),# 来源FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=100)]# 3. 创建集合 schemaschema = CollectionSchema(fields, description="AI Agent长期记忆：历史经验存储")# 4. 创建集合（如已存在则删除重建）collection_name = "agent_long_term_memory"if utility.has_collection(collection_name):utility.drop_collection(collection_name)collection = Collection(name=collection_name, schema=schema)# 5. 创建向量索引（加速相似度检索）index_params = {"index_type": "IVF_FLAT", # 适合中小规模数据的精确检索"metric_type": "COSINE", # 用余弦相似度衡量向量距离"params": {"nlist": 1024} # 聚类数量，影响检索速度和精度}collection.create_index(field_name="vector", index_params=index_params)collection.load() # 加载集合到内存，准备检索
3. 经验存储：将历史经验写入向量数据库
需先通过嵌入模型将经验文本转化为向量，再插入数据库：
from sentence_transformers import SentenceTransformer # 嵌入模型# 初始化嵌入模型（将文本转为768维向量）embedder = SentenceTransformer('all-MiniLM-L6-v2')def add_experience(collection, experience):"""向长期记忆添加一条经验:param collection: Milvus集合对象:param experience: 经验字典（含content、scene等字段）"""# 1. 生成唯一IDexperience_id = f"exp_{experience['timestamp']}"# 2. 将经验文本转为向量vector = embedder.encode(experience["content"]).tolist()# 3. 组装插入数据data = [[experience_id], # id[vector], # vector[experience["content"]], # content[experience["scene"]], # scene[experience["timestamp"]], # timestamp[experience["priority"]], # priority[experience["source"]] # source]# 4. 插入向量数据库collection.insert(data)print(f"已添加经验：{experience_id}")# 示例：添加一条用户反馈的经验new_experience = {"content": "在处理用户年龄数据时，若缺失值比例超过10%，直接删除字段会导致样本量减少过多，用户建议改用中位数填充并标记缺失记录","scene": ["data_cleaning", "missing_value_processing", "age_field"],"timestamp": 1690000000,"priority": 4, # 较高优先级（用户明确反馈）"source": "user_feedback"}add_experience(collection, new_experience)
4. 经验检索：根据当前任务查询相似经验
结合 “当前任务描述” 生成向量，检索最相似的历史经验，并可通过元数据（如场景）过滤：
def retrieve_similar_experiences(collection, query_text, scene_filters=None, top_k=3):"""检索与当前任务相似的历史经验:param query_text: 当前任务描述（如“如何处理年龄字段的高比例缺失值”）:param scene_filters: 场景过滤条件（如["missing_value_processing"]）:param top_k: 返回最相似的前k条:return: 相似经验列表"""# 1. 将查询文本转为向量query_vector = embedder.encode(query_text).tolist()# 2. 构建检索参数（结合向量相似度和元数据过滤）search_params = {"metric_type": "COSINE","params": {"nprobe": 10} # 检索时探查的聚类数量，平衡速度和精度}# 3. 元数据过滤条件（如仅检索“缺失值处理”场景的经验）expr = Noneif scene_filters:# Milvus中数组包含条件的表达式（如scene包含"missing_value_processing"）expr = " && ".join([f'"{s}" in scene' for s in scene_filters])# 4. 执行检索results = collection.search(data=[query_vector],anns_field="vector",param=search_params,limit=top_k,expr=expr, # 元数据过滤output_fields=["content", "scene", "priority", "source"] # 需要返回的字段)# 5. 整理结果（提取相似度分数和经验内容）similar_experiences = []for hit in results[0]:similar_experiences.append({"id": hit.id,"similarity": hit.score, # 余弦相似度（0-1，越大越相似）"content": hit.entity.get("content"),"scene": hit.entity.get("scene"),"priority": hit.entity.get("priority")})return similar_experiences# 示例：查询与“处理年龄字段高比例缺失值”相关的经验query = "如何处理年龄字段中超过10%的缺失值？"similar_exps = retrieve_similar_experiences(collection,query_text=query,scene_filters=["missing_value_processing", "age_field"],top_k=2)print("相似经验检索结果：")for exp in similar_exps:print(f"相似度：{exp['similarity']:.2f}，内容：{exp['content']}")
5. 记忆演化：更新与维护长期记忆
长期记忆需定期优化（如淘汰过时经验、强化高频复用经验）：
def update_experience_priority(collection, exp_id, new_priority):"""更新经验优先级（如被多次复用的经验提升优先级）"""collection.update(expr=f"id == '{exp_id}'",partition_name=None,params={"priority": new_priority})def prune_old_experiences(collection, days_threshold=365):"""删除超过阈值天数的低优先级经验（如1年未复用的优先级1经验）"""# 计算阈值时间戳（当前时间 - 天数*86400秒）current_ts = 1690000000 # 实际使用时用time.time()threshold_ts = current_ts - days_threshold * 86400# 删除条件：时间戳早于阈值且优先级<=2expr = f"timestamp < {threshold_ts} && priority <= 2"collection.delete(expr=expr)print(f"已删除过时低优先级经验")
三、设计要点与技术选型建议
嵌入模型选择：
-
轻量级场景：用 all-MiniLM-L6-v2（768 维，速度快，适合中小数据量）； -
高精度场景：用 text-embedding-ada-002（OpenAI，1536 维，语义理解能力强）。 向量数据库选型：
-
开源部署：Milvus（支持分布式，适合大规模数据）、Qdrant（轻量，易部署）； -
云服务：Pinecone、Weaviate（无需自建，按使用量计费）。 检索策略优化：
-
基础：“向量相似度 + 场景标签过滤”（平衡召回率和精准度）； -
进阶：结合经验优先级（如优先返回优先级 > 3 的结果）、时间衰减（近期经验权重更高）。 性能考量：
-
向量维度不宜过高（建议 512-1536 维），避免检索耗时； -
对高频访问的经验建立缓存（如 Redis），减少数据库压力。
总结
长期记忆的向量数据库设计核心是 “将经验语义转化为可计算的向量，并通过相似性检索实现跨场景复用”。通过结构化的经验定义、合理的元数据设计和灵活的检索策略，AI Agent 能从历史经验中快速学习，逐步提升决策能力 —— 这正是 Agent “智能化” 和 “持续进化” 的关键支撑。