智能体记忆系统

智能体记忆系统

记忆是智能体的基石。没有它，每次交互都将从零开始。本技能涵盖智能体记忆的架构：短期记忆（上下文窗口）、长期记忆（向量存储）以及组织这些记忆的认知架构。

核心洞察：记忆不仅仅是存储，更是检索。如果你无法找到正确的信息，存储一百万个事实也毫无意义。分块（Chunking）、嵌入（Embedding）和检索策略决定了你的智能体是记得还是忘记。

该领域目前较为碎片化，术语不统一。我们采用 CoALA 认知架构框架：语义记忆（事实）、情景记忆（经历）和程序记忆（操作知识）。

原则

• 记忆质量 = 检索质量，而非存储数量

• 为了检索而分块，而非为了存储而分块

• 上下文隔离是记忆的大敌

• 为正确的信息选择正确的记忆类型

• 衰减旧记忆 —— 并非所有内容都应永久保存

• 在投入生产前测试检索准确率

• 后台记忆形成优于实时形成

能力

• agent-memory (智能体记忆)

• long-term-memory (长期记忆)

• short-term-memory (短期记忆)

• working-memory (工作记忆)

• episodic-memory (情景记忆)

• semantic-memory (语义记忆)

• procedural-memory (程序记忆)

• memory-retrieval (记忆检索)

• memory-formation (记忆形成)

• memory-decay (记忆衰减)

范围

• vector-database-operations (向量数据库操作) → data-engineer (数据工程师)

• rag-pipeline-architecture (RAG 流水线架构) → llm-architect (LLM 架构师)

• embedding-model-selection (嵌入模型选择) → ml-engineer (机器学习工程师)

• knowledge-graph-design (知识图谱设计) → knowledge-engineer (知识工程师)

工具链

记忆框架 (Memory_frameworks)

• LangMem (LangChain) - 适用场景：需要持久化记忆的 LangGraph 智能体。注：支持语义、情景、程序记忆类型。

• MemGPT / Letta - 适用场景：虚拟上下文管理，操作系统风格的记忆。注：分层记忆体系，自动分页。

• Mem0 - 适用场景：用于个性化的用户记忆层。注：专为用户偏好和历史记录设计。

向量存储 (Vector_stores)

• Pinecone - 适用场景：托管式、企业级规模（数十亿向量）。注：查询性能最佳，成本最高。

• Qdrant - 适用场景：复杂的元数据过滤，开源。注：基于 Rust，过滤能力极强。

• Weaviate - 适用场景：混合搜索，知识图谱特性。注：GraphQL 接口，擅长处理关系。

• ChromaDB - 适用场景：原型开发，中小规模应用。注：开发者友好，10万向量时 p50 延迟约 20ms。

• pgvector - 适用场景：已在使用 PostgreSQL，追求简单部署。注：适用于 100万以下向量，工具链熟悉。

嵌入模型 (Embedding_models)

• OpenAI text-embedding-3-large - 适用场景：最高质量，3072 维。注：$0.13/1M tokens。

• OpenAI text-embedding-3-small - 适用场景：平衡之选，1536 维。注：$0.02/1M tokens，成本降低 5 倍。

• nomic-embed-text-v1.5 - 适用场景：开源，本地部署。注：768 维，质量良好。

• all-MiniLM-L6-v2 - 适用场景：轻量级，快速本地嵌入。注：384 维，延迟最低。

模式

记忆类型架构

为不同信息选择正确的记忆类型

适用场景：设计智能体记忆系统

记忆类型架构 (CoALA 框架):

"""
三种用于不同目的的记忆类型：

1. 语义记忆 (Semantic Memory)：事实和知识
- 你对世界的认知
- 用户偏好
"""
1. 语义记忆 (Semantic Memory)：事实、概念、领域知识
- 存储在配置文件（结构化）或集合（非结构化）中

2. 情节记忆 (Episodic Memory)：经历与事件
- 发生了什么（带时间戳的事件）
- 过往对话、任务结果
- 用于从经验中学习

3. 程序记忆 (Procedural Memory)：如何操作
- 规则、技能、工作流
- 通常以 few-shot 示例形式实现
- “我之前是如何解决这个问题的？”

LangMem 实现

""" from langmem import MemoryStore from langgraph.graph import StateGraph

初始化记忆存储

memory = MemoryStore( connection_string=os.environ["POSTGRES_URL"] )

语义记忆：用户配置文件

await memory.semantic.upsert( namespace="user_profile", key=user_id, content={ "name": "Alice", "preferences": ["dark mode", "concise responses"], "expertise_level": "developer", } )

情节记忆：过往交互

await memory.episodic.add( namespace="conversations", content={ "timestamp": datetime.now(), "summary": "Helped debug authentication issue", "outcome": "resolved", "key_insights": ["Token expiry was root cause"], }, metadata={"user_id": user_id, "topic": "debugging"} )

程序记忆：习得的模式

await memory.procedural.add( namespace="skills", content={ "task_type": "debug_auth", "steps": ["Check token expiry", "Verify refresh flow"], "example_interaction": few_shot_example, } ) """

运行时记忆检索

""" async def prepare_context(user_id, query): # 获取用户配置文件 (语义) profile = await memory.semantic.get( namespace="user_profile", key=user_id )

# 查找相关的过往经历 (情节)
similar_experiences = await memory.episodic.search(
namespace="conversations",
query=query,
filter={"user_id": user_id},
limit=3
)

# 查找相关的技能 (程序)
relevant_skills = await memory.procedural.search(
namespace="skills",
query=query,
limit=2
)

return {
"profile": profile,
"past_experiences": similar_experiences,
"relevant_skills": relevant_skills,
}
"""

向量数据库选择模式

为你的用例选择合适的向量数据库

适用场景：搭建持久化记忆存储

向量数据库选择：

"""
决策矩阵：

| | Pinecone | Qdrant | Weaviate | ChromaDB | pgvector |
|------------|----------|--------|----------|----------|----------|
| 规模 | 十亿级 | 亿级+ | 亿级+ | 百万级 | 百万级 |
| 托管服务 | 是 | 两者均有 | 两者均有 | 自托管 | 自托管 |
| 过滤能力 | 基础 | 最佳 | 良好 | 基础 | SQL |
| 混合检索 | 否 | 是 | 最佳 | 否 | 是 |
| 成本 | 高 | 中 | 中 | 免费 | 免费 |
| 延迟 | 5ms | 7ms | 10ms | 20ms | 15ms |
"""

Pinecone (企业级规模)

""" from pinecone import Pinecone

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("agent-memory")

带有元数据的 Upsert

index.upsert( vectors=[ { "id": f"memory-{uuid4()}", "values": embedding, "metadata": { "user_id": user_id, "timestamp": datetime.now().isoformat(), "type": "episodic", "content": memory_text, } } ], namespace=n amespace )

带过滤条件的查询

results = index.query( vector=query_embedding, filter={"user_id": user_id, "type": "episodic"}, top_k=5, include_metadata=True ) """

Qdrant (复杂过滤)

""" from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition

client = QdrantClient(url="http://localhost:6333")

使用 Qdrant 进行复杂过滤

results = client.search( collection_name="agent_memory", query_vector=query_embedding, query_filter=Filter( must=[ FieldCondition(key="user_id", match={"value": user_id}), FieldCondition(key="type", match={"value": "semantic"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "security"]}), ] ), limit=5 ) """

ChromaDB (快速原型)

""" import chromadb

client = chromadb.PersistentClient(path="./memory_db")
collection = client.get_or_create_collection("agent_memory")

适用于原型的简单快速方案

collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[memory_text], metadatas=[{"user_id": user_id, "type": "episodic"}] )

results = collection.query(
query_embeddings=[query_embedding],
n_results=5,
where={"user_id": user_id}
)
"""

分块策略模式 (Chunking Strategy Pattern)

将文档拆分为可检索的块

适用场景：为内存存储处理文档

分块策略：

"""
分块的两难境地：

• 过大：向量失去特异性

• 过小：丢失上下文

最佳分块大小取决于：

• 文档类型（代码 vs 散文 vs 数据）

• 查询模式（事实性 vs 探索性）

• 嵌入模型（每个模型都有其最佳区间）

通用指南：大多数场景建议 256-512 个 token
"""

固定大小分块 (基准方案)

""" from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 字符数
chunk_overlap=50, # 重叠部分防止句子被截断
separators=["\n\n", "\n", ". ", " ", ""] # 优先级顺序
)

chunks = splitter.split_text(document)
"""

语义分块 (更高质量)

""" from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings

基于语义相似度进行拆分

splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )

chunks = splitter.split_text(document)
"""

结构感知分块 (具有层级结构的文档)

""" from langchain.text_splitter import MarkdownHeaderTextSplitter

遵循文档结构

splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] )

chunks = splitter.split_text(markdown_doc)

每个分块都包含用于提供上下文的标题元数据

"""

上下文分块 (Anthropic 的方法)

"""

在嵌入之前为每个分块添加上下文

可将检索失败率降低 35%

def add_context_to_chunk(chunk, document_summary):
context_prompt = f'''
文档摘要: {document_summary}

以下是该文档的一个分块:
{chunk}
'''
return context_prompt

对上下文增强后的分块进行嵌入，而非原始分块

for chunk in chunks: contextualized = add_context_to_chunk(chunk, summary) embedding = embed(contextualized) store(chunk, embedding) # 存储原始文本，但使用增强后的向量 """

代码特定 (Code-Spe)

特定分块 (Specific Chunking) """ from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

语言感知分块

python_splitter = RecursiveCharacterTextSplitter.from_language( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )

尊重函数/类边界

chunks = python_splitter.split_text(python_code) """

背景记忆形成 (Background Memory Formation)

异步处理记忆以提高质量

适用场景：希望在不降低交互速度的前提下提高召回率

背景记忆形成：

"""
实时提取记忆会降低对话速度，并增加 Agent 工具调用的复杂度。
在对话结束后进行后台处理可以获得更高质量的记忆。

模式：潜意识记忆形成
"""

LangGraph 后台处理

""" from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver

async def background_memory_processor(thread_id: str):
# 在对话结束或进入闲置状态后运行
conversation = await load_conversation(thread_id)

# 在无时间压力的情况下提取洞察
insights = await llm.invoke('''
分析这段对话并提取：
1. 关于用户的关键事实
2. 揭示的用户偏好
3. 已完成或待办的任务
4. 用户行为模式

请详尽分析 - 此过程在后台运行。

对话内容：
{conversation}
''')

# 存储到长期记忆
for insight in insights:
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(insight),
content=insight,
metadata={"source_thread": thread_id}
)

在对话结束或闲置超时时触发

@on_conversation_idle(timeout_minutes=5) async def process_conversation(thread_id): await background_memory_processor(thread_id) """

记忆巩固（类比睡眠）

"""

定期巩固并去重记忆

async def consolidate_memories(user_id: str):
# 获取该用户的所有记忆
memories = await memory.semantic.list(
namespace="user_insights",
filter={"user_id": user_id}
)

# 寻找相似记忆（潜在重复项）
clusters = cluster_by_similarity(memories, threshold=0.9)

# 合并相似记忆
for cluster in clusters:
if len(cluster) > 1:
merged = await llm.invoke(f'''
将这些相关的记忆合并为一条：
{cluster}

保留所有重要信息。
''')
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(merged),
content=merged
)
# 删除原记忆
for old in cluster:
await memory.semantic.delete(old.id)
"""

记忆衰减模式 (Memory Decay Pattern)

遗忘陈旧且无关的记忆

适用场景：记忆量过大导致检索速度下降

记忆衰减：

"""
并非所有记忆都应永久保存：

• 旧的偏好可能已过时

• 任务细节会失去相关性

• 冲突的记忆会干扰检索

基于以下维度实现智能衰减：

• 时效性（创建/访问时间？）

• 频率（检索次数？）

• 重要性（是核心事实还是细节？）

"""

基于时间的衰减

""" from datetime import datetime, timedelta

async def decay_old_memories(namespace: str, max_age_days: int):
cutoff = datetime.now() - timedel
ta(days=max_age_days)

old_memories = await memory.episodic.list(
namespace=namespace,
filter={"last_accessed": {"$lt": cutoff.isoformat()}}
)

for mem in old_memories:
# 软删除（标记为已归档）
await memory.episodic.update(
id=mem.id,
metadata={"archived": True, "archived_at": datetime.now()}
)
"""

基于效用的衰减 (MIRIX 方法)

""" def calculate_memory_utility(memory): ''' 受认知科学启发的综合效用评分： - 近时性 (Recency)：上次访问是什么时候？ - 频率 (Frequency)：访问频率如何？ - 重要性 (Importance)：该信息有多关键？ ''' now = datetime.now()

# 近时性评分（半衰期为 72 小时的指数衰减）
hours_since_access = (now - memory.last_accessed).total_seconds() / 3600
recency_score = 0.5 ** (hours_since_access / 72)

# 频率评分
frequency_score = min(memory.access_count / 10, 1.0)

# 重要性（来自元数据或启发式算法）
importance = memory.metadata.get("importance", 0.5)

# 加权组合
utility = (
0.4 * recency_score +
0.3 * frequency_score +
0.3 * importance
)

return utility

async def prune_low_utility_memories(threshold=0.2):
all_memories = await memory.list_all()
for mem in all_memories:
if calculate_memory_utility(mem) < threshold:
await memory.archive(mem.id)
"""

潜在痛点 (Sharp Edges)

分块导致信息脱离上下文

严重程度：极高 (CRITICAL)

场景：为向量存储处理文档

症状：
检索到了分块，但分块本身缺乏意义。智能体的回答缺失全局视角。例如，检索到“该函数返回 X”，但不知道是指哪个函数；或者出现“这个”之类的指代，但不知道具体指什么。

失效原因：
在为 AI 处理进行分块时，我们切断了原有的连接，将整体叙事简化为孤立的片段，这往往会导致丢失全局信息。一个关于“配置”的分块如果没有关于“正在配置哪个系统”的上下文，几乎毫无用处。

建议修复方案：

上下文分块 (Anthropic 的方法)

在嵌入之前为每个分块添加文档上下文

可将检索失败率降低 35%

def contextualize_chunk(chunk, document):
summary = summarize(document)

# LLM 为分块生成上下文
context = llm.invoke(f'''
文档摘要: {summary}

请为该分块生成一段简短的上下文陈述，
以帮助读者理解其指代的内容：

{chunk}
''')

return f"{context}\n\n{chunk}"

对增强上下文后的版本进行嵌入

for chunk in chunks: contextualized = contextualize_chunk(chunk, full_doc) embedding = embed(contextualized) # 存储原始分块，但使用增强上下文后的向量 store(original=chunk, embedding=embedding)

层级分块

以多种粒度存储

chunks_small = split(doc, size=256) chunks_medium = split(doc, size=512) chunks_large = split(doc, size=1024)

根据查询类型在相应的层级进行检索

分块大小与查询模式不匹配

严重程度：高 (HIGH)

场景：配置内存存储的分块策略

症状：
高质量的文档产生了低质量的检索结果。简单问题漏掉相关信息，复杂问题只能得到碎片化答案而非完整回答。

失效原因：
最佳分块大小取决于查询模式：

• 事实性查询需要小而具体的分块

• 概念性查询需要...

• 需要更大的上下文

• 代码需要函数级的边界

最佳切分大小取决于文档类型和嵌入模型。
默认的 1000 个字符并不适用于所有场景。

建议修复方案：

测试不同的切分大小

from sklearn.metrics import recall_score

def evaluate_chunk_size(documents, test_queries, chunk_size):
chunks = split_documents(documents, size=chunk_size)
index = build_index(chunks)

correct_retrievals = 0
for query, expected_chunk in test_queries:
results = index.search(query, k=5)
if expected_chunk in results:
correct_retrievals += 1

return correct_retrievals / len(test_queries)

测试多种大小

for size in [256, 512, 768, 1024]: recall = evaluate_chunk_size(docs, test_queries, size) print(f"Size {size}: Recall@5 = {recall:.2%}")

按内容类型推荐的大小

CHUNK_SIZES = { "documentation": 512, # 完整概念 "code": 1000, # 函数级 "conversation": 256, # 对话轮次级 "articles": 768, # 段落级 }

使用重叠（overlap）防止边界问题

splitter = RecursiveCharacterTextSplitter( chunk_size=512, chunk_overlap=50, # 10% 重叠 )

语义搜索返回无关结果

严重程度：高

场景：查询内存以获取上下文

症状：
Agent 检索到的记忆看似相关但并无用处。
例如，“告诉我用户的偏好”返回了关于通用偏好的对话，而非该用户的偏好。错误内容的相似度得分很高。

失效原因：
语义相似度并不等同于相关性。“用户喜欢 Python”和“Python 是一种编程语言”在语义上相似，但信息类型截然不同。如果没有元数据过滤，检索就变成了简单的词汇匹配。

建议修复方案：

始终先通过元数据过滤

不要仅依赖语义相似度

错误做法：仅使用语义搜索

results = index.query( vector=query_embedding, top_k=5 )

正确做法：先过滤再搜索

results = index.query( vector=query_embedding, filter={ "user_id": current_user.id, "type": "preference", "created_after": cutoff_date, }, top_k=5 )

使用混合搜索（语义 + 关键词）

from qdrant_client import QdrantClient

client = QdrantClient(...)

使用融合技术的混合搜索

results = client.search( collection_name="memories", query_vector=semantic_embedding, query_text=query, # 同时进行关键词匹配 fusion={"method": "rrf"}, # 倒数排名融合 (Reciprocal Rank Fusion) )

使用 Cross-Encoder 对结果进行重排序

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

初始检索（侧重召回率）

candidates = index.query(query_embedding, top_k=20)

重排序（侧重精准率）

pairs = [(query, c.text) for c in candidates] scores = reranker.predict(pairs) reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)

旧记忆覆盖当前信息

严重程度：高

场景：用户偏好或事实随时间发生变化

症状：
Agent 使用过时的偏好。例如，6 个月前的“用户偏好深色模式”覆盖了最近的“切换到浅色模式”请求。Agent 自信地使用陈旧数据。

失效原因：
向量数据库默认不具备时间感知能力。一年前的记忆与今天的记忆在检索权重上相同。对于偏好和可变事实，近期信息通常应覆盖旧信息。

建议方案：
修复方案：

添加时间权重评分

from datetime import datetime, timedelta

def time_decay_score(memory, half_life_days=30):
age = (datetime.now() - memory.created_at).days
decay = 0.5 ** (age / half_life_days)
return decay

def retrieve_with_recency(query, user_id):
# 获取候选集
candidates = index.query(
vector=embed(query),
filter={"user_id": user_id},
top_k=20
)

# 应用时间衰减
for candidate in candidates:
time_score = time_decay_score(candidate)
candidate.final_score = candidate.similarity * 0.7 + time_score * 0.3

# 按最终得分重新排序
return sorted(candidates, key=lambda x: x.final_score, reverse=True)[:5]

对偏好设置执行更新而非追加

async def update_preference(user_id, category, value): # 删除旧偏好 await memory.delete( filter={"user_id": user_id, "type": "preference", "category": category} )

# 存储新偏好
await memory.upsert(
id=f"pref-{user_id}-{category}",
content={"category": category, "value": value},
metadata={"updated_at": datetime.now()}
)

为事实添加显式版本控制

await memory.upsert( id=f"fact-{fact_id}-v{version}", content=new_fact, metadata={ "version": version, "supersedes": previous_id, "valid_from": datetime.now() } )

矛盾记忆被同时检索

严重程度：中等

场景：用户更改了偏好或提供了冲突的信息

症状：
智能体在同一上下文中同时检索到“用户偏好深色模式”和“用户偏好浅色模式”。导致回答不一致，给用户一种困惑或健忘的感觉。

失效原因：
缺乏冲突解决机制，导致新旧信息共存。由于两者讨论的主题相同（偏好），语义搜索可能会同时返回两者。智能体无法判断哪个是当前有效的。

推荐修复方案：

在存储时检测冲突

async def store_with_conflict_check(memory, user_id): # 查找潜在的冲突记忆 similar = await index.query( vector=embed(memory.content), filter={"user_id": user_id, "type": memory.type}, threshold=0.9, # 高相似度 top_k=5 )

for existing in similar:
if is_contradictory(memory.content, existing.content):
# 请求解决冲突
resolution = await resolve_conflict(memory, existing)
if resolution == "replace":
await index.delete(existing.id)
elif resolution == "version":
await mark_superseded(existing.id, memory.id)

await index.upsert(memory)

冲突检测启发式算法

def is_contradictory(new_content, old_content): # 使用 LLM 检测矛盾 result = llm.invoke(f''' 这两句话是否矛盾？

语句 1: {old_content}
语句 2: {new_content}

请仅回答 YES 或 NO。
''')
return result.strip().upper() == "YES"

定期整合

async def consolidate_memories(user_id): all_memories = await index.list(filter={"user_id": user_id}) clusters = cluster_by_topic(all_memories)

for cluster in clusters:
if has_conflicts(cluster):
resolved = await llm.invoke(f'''
这些记忆可能存在冲突。请创建一个整合后的记忆，以代表当前的真实情况：
{cluster}
''')
awai
t replace_cluster(cluster, resolved)

检索记忆超出上下文窗口

严重程度：中

场景：一次性检索过多记忆

症状：
出现 Token 限制错误。智能体截断重要信息。
系统提示词（System Prompt）被截断。检索到的记忆与用户查询争抢空间。

失效原因：
检索通常返回 top-k 个结果。如果 k 值过高或分块（chunk）过大，检索到的上下文将撑满窗口。关键信息（系统提示词、近期消息）会被挤出。

建议修复方案：

为不同类型的记忆分配 Token 预算

TOKEN_BUDGET = { "system_prompt": 500, "user_profile": 200, "recent_messages": 2000, "retrieved_memories": 1000, "current_query": 500, "buffer": 300, # 安全余量 }

def budget_aware_retrieval(query, context_limit=4000):
remaining = context_limit - TOKEN_BUDGET["system_prompt"] - TOKEN_BUDGET["buffer"]

# 优先保证近期消息
recent = get_recent_messages(limit=TOKEN_BUDGET["recent_messages"])
remaining -= count_tokens(recent)

# 其次是用户画像
profile = get_user_profile(limit=TOKEN_BUDGET["user_profile"])
remaining -= count_tokens(profile)

# 最后用剩余预算检索记忆
memories = retrieve_memories(query, max_tokens=remaining)

return build_context(profile, recent, memories)

根据分块大小动态调整 k 值

def retrieve_with_budget(query, max_tokens=1000): avg_chunk_tokens = 150 # 基于数据统计 max_k = max_tokens // avg_chunk_tokens

results = index.query(query, top_k=max_k)

# 如果仍然超出预算则进行裁剪
total_tokens = 0
filtered = []
for result in results:
tokens = count_tokens(result.text)
if total_tokens + tokens <= max_tokens:
filtered.append(result)
total_tokens += tokens
else:
break

return filtered

查询与文档的 Embedding 模型不一致

严重程度：中

场景：升级 Embedding 模型或混合使用不同供应商

症状：
检索质量突然下降。找不到相关文档。
返回结果随机。新文档正常，旧文档失效。

失效原因：
不同的 Embedding 模型产生不同的向量空间。使用 text-embedding-3 编码的查询无法匹配使用 text-ada-002 编码的文档。混合模型会导致相似度分数失效。

建议修复方案：

在元数据中记录 Embedding 模型

await index.upsert( id=doc_id, vector=embedding, metadata={ "embedding_model": "text-embedding-3-small", "embedding_version": "2024-01", "content": content } )

检索时按模型版本过滤

results = index.query( vector=query_embedding, filter={"embedding_model": current_model}, top_k=10 )

模型升级的迁移策略

async def migrate_embeddings(old_model, new_model): # 获取所有使用旧模型的文档 old_docs = await index.list(filter={"embedding_model": old_model})

for doc in old_docs:
# 使用新模型重新编码
new_embedding = await embed(doc.content, model=new_model)

# 原地更新
await index.update(
id=doc.id,
vector=new_embedding,
metadata={"embedding_model": new_model}
)

迁移期间使用独立的集合（Collection）

旧集合：处理生产环境查询

新集合：进行重新编码

完成后统一切换

验证检查

生产环境中使用内存存储 C

严重程度：错误 (ERROR)

内存存储在重启时会丢失数据

消息：检测到内存存储。生产环境请使用持久化存储（如 Postgres, Qdrant, Pinecone）。

向量更新缺失元数据

严重程度：警告 (WARNING)

向量应包含用于过滤的元数据

消息：向量更新时未提供元数据。请添加 user_id、type、timestamp 以实现正确过滤。

查询缺失用户过滤

严重程度：错误 (ERROR)

查询应按用户过滤以防止数据泄露

消息：向量查询未进行用户过滤。请务必通过 user_id 过滤以防止数据泄露。

硬编码分块大小且缺乏依据

严重程度：信息 (INFO)

分块大小应经过测试并有据可依

消息：分块大小为硬编码。请针对您的内容类型测试不同的大小并衡量检索准确率。

分块缺失重叠

严重程度：警告 (WARNING)

分块重叠可防止边界问题

消息：文本分割未设置重叠。请添加 chunk_overlap（10-20%）以防止边界问题。

语义搜索缺失过滤器

严重程度：警告 (WARNING)

纯语义搜索经常返回无关结果

消息：纯语义搜索。请添加元数据过滤器（用户、类型、时间）以提高相关性。

检索缺失结果限制

严重程度：警告 (WARNING)

无限制检索可能会导致上下文溢出

消息：检索未设置限制。请设置 top_k 以防止上下文溢出。

嵌入缺失模型版本追踪

严重程度：警告 (WARNING)

追踪嵌入模型以处理迁移

消息：在元数据中存储嵌入模型版本，以便处理模型迁移。

文档与查询使用不同嵌入模型

严重程度：错误 (ERROR)

文档和查询必须使用相同的嵌入模型

消息：确保索引和查询使用相同的嵌入模型。

协作

委派触发条件

• 用户需要大规模向量数据库 -> data-engineer (生产级向量存储运维)

• 用户需要嵌入模型优化 -> ml-engineer (自定义嵌入、微调)

• 用户需要知识图谱 -> knowledge-engineer (基于图的记忆结构)

• 用户需要 RAG 流水线 -> llm-architect (端到端检索增强生成)

• 用户需要多智能体共享内存 -> multi-agent-orchestration (智能体间的内存共享)

相关技能

协同工作：autonomous-agents, multi-agent-orchestration, llm-architect, agent-tool-builder

使用场景

• 用户提及或暗示：智能体记忆 (agent memory)

• 用户提及或暗示：长期记忆 (long-term memory)

• 用户提及或暗示：记忆系统 (memory systems)

• 用户提及或暗示：跨会话记忆 (remember across sessions)

• 用户提及或暗示：记忆检索 (memory retrieval)

• 用户提及或暗示：情景记忆 (episodic memory)

• 用户提及或暗示：语义记忆 (semantic memory)

• 用户提及或暗示：向量存储 (vector store)

• 用户提及或暗示：RAG

• 用户提及或暗示：langmem

• 用户提及或暗示：memgpt

• 用户提及或暗示：对话历史 (conversation history)

局限性

• 仅在任务明确符合上述范围时使用此技能。

• 不要将输出视为针对特定环境的验证、测试或专家评审的替代方案。

• 如果缺失必要输入、权限、安全边界或成功标准，请停止并请求澄清。