在构建现代 AI Agent 系统时，记忆检索（Memory Retrieval）是决定智能助手响应质量的核心组件。当用户提出问题时，Agent 需要从大量历史对话、文档和上下文中快速找到最相关的内容，而向量相似度搜索（Vector Similarity Search）正是解决这一问题的关键技术。
为什么向量检索对 AI Agent 至关重要
传统的关键词搜索无法理解语义关系。例如，用户搜索"如何治疗感冒"时，关键词搜索只能找到包含"感冒"和"治疗"字样的文档，而向量检索能够理解"感冒"与"流感"、"药物治疗"与"家庭疗法"之间的语义关联，从而返回更相关的结果。
向量相似度算法对比
1. 余弦相似度（Cosine Similarity）
余弦相似度衡量两个向量在方向上的相似程度，取值范围为 [-1, 1]。当两个向量的方向完全相同时，值为 1；完全相反时，值为 -1。这种算法特别适合处理文本嵌入向量，因为文本的长度不会影响相似度计算结果。
import numpy as np
def cosine_similarity(vec_a, vec_b):
"""计算两个向量的余弦相似度"""
dot_product = np.dot(vec_a, vec_b)
norm_a = np.linalg.norm(vec_a)
norm_b = np.linalg.norm(vec_b)
if norm_a == 0 or norm_b == 0:
return 0.0
return dot_product / (norm_a * norm_b)
示例：计算两个句子的嵌入向量相似度
sentence1_embedding = np.array([0.1, 0.3, 0.5, 0.7])
sentence2_embedding = np.array([0.2, 0.4, 0.6, 0.8])
similarity = cosine_similarity(sentence1_embedding, sentence2_embedding)
print(f"余弦相似度: {similarity:.4f}") # 输出: 0.99472. 欧氏距离（Euclidean Distance）
欧氏距离是衡量两点之间直线距离的经典算法。在向量空间中，距离越小表示越相似。这种算法对向量的 magnitude（向量长度）敏感，适合需要考虑向量绝对值差异的场景。
def euclidean_distance(vec_a, vec_b):
"""计算两个向量的欧氏距离"""
return np.linalg.norm(vec_a - vec_b)
示例：计算两个嵌入向量的欧氏距离
distance = euclidean_distance(sentence1_embedding, sentence2_embedding)
print(f"欧氏距离: {distance:.4f}") # 输出: 0.4472
将距离转换为相似度（距离越小，相似度越高）
max_distance = np.sqrt(len(vec_a)) # 理论最大距离
similarity_from_distance = 1 - (distance / max_distance)
print(f"基于距离的相似度: {similarity_from_distance:.4f}")3. 点积相似度（Dot Product）
点积是两个向量对应元素相乘后求和的结果。当向量已经经过归一化处理时，点积等价于余弦相似度。这种算法计算效率高，是大多数向量数据库的默认选择。
def dot_product_similarity(vec_a, vec_b):
"""计算点积相似度"""
return np.dot(vec_a, vec_b)
示例：使用点积计算相似度
similarity = dot_product_similarity(sentence1_embedding, sentence2_embedding)
print(f"点积相似度: {similarity:.4f}") # 输出: 1.32提升召回率的三大策略
策略一：混合搜索（Hybrid Search）
混合搜索结合了向量搜索和关键词搜索的优点，能够同时捕获语义相关性和精确关键词匹配。这种方法特别适合需要查找特定术语或名称的场景。
from typing import List, Dict, Tuple
class HybridSearchEngine:
def __init__(self, vector_db, keyword_index):
self.vector_db = vector_db
self.keyword_index = keyword_index
self.alpha = 0.7 # 向量搜索权重
def search(self, query: str, top_k: int = 10) -> List[Dict]:
# 1. 向量搜索：获取语义相关结果
vector_results = self.vector_db.search(query, top_k * 2)
# 2. 关键词搜索：获取精确匹配结果
keyword_results = self.keyword_index.search(query, top_k * 2)
# 3. 结果融合
combined_scores = self._fuse_results(vector_results, keyword_results)
# 4. 返回 Top-K 结果
return sorted(combined_scores, key=lambda x: x['score'], reverse=True)[:top_k]
def _fuse_results(self, vector_results, keyword_results) -> List[Dict]:
"""使用 Reciprocal Rank Fusion 融合搜索结果"""
fused = {}
# 添加向量搜索结果
for i, result in enumerate(vector_results):
doc_id = result['id']
rank = i + 1
score = (1 / (60 + rank)) * self.alpha
fused[doc_id] = fused.get(doc_id, 0) + score
# 添加关键词搜索结果
for i, result in enumerate(keyword_results):
doc_id = result['id']
rank = i + 1
score = (1 / (60 + rank)) * (1 - self.alpha)
fused[doc_id] = fused.get(doc_id, 0) + score
return [{'id': doc_id, 'score': score} for doc_id, score in fused.items()]策略二：重排序（Reranking）
重排序使用更强大的模型对初步检索结果进行二次排序，能够显著提升结果的相关性。常见的重排序模型包括 Cross-Encoder 和 BERT-based 模型。
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: List[str], top_k: int = 5) -> List[Dict]:
"""
使用 Cross-Encoder 对文档进行重排序
Args:
query: 用户查询
documents: 待排序的文档列表
top_k: 返回的顶部结果数量
Returns:
重排序后的文档列表（包含相关性分数）
"""
# 创建查询-文档对
pairs = [[query, doc] for doc in documents]
# 批量计算相关性分数
scores = self.model.predict(pairs)
# 按分数排序并返回顶部结果
results = [
{'document': doc, 'score': float(score)}
for doc, score in zip(documents, scores)
]
return sorted(results, key=lambda x: x['score'], reverse=True)[:top_k]
使用示例
reranker = Reranker()
query = "如何预防心脏病"
documents = [
"保持健康的饮食习惯可以降低心脏病风险",
"每天运动30分钟有助于心脏健康",
"感冒时应多喝水休息"
]
reranked_results = reranker.rerank(query, documents, top_k=2)
for result in reranked_results:
print(f"分数: {result['score']:.4f} - 文档: {result['document']}")策略三：语义缓存（Semantic Caching）
语义缓存通过向量相似度判断新查询是否与历史查询相似，如果是则直接返回缓存结果，从而减少 API 调用次数和响应延迟。
import hashlib
from datetime import datetime, timedelta
class SemanticCache:
def __init__(self, similarity_threshold: float = 0.95, ttl_hours: int = 24):
self.cache = {} # {cache_key: {'result': ..., 'timestamp': ...}}
self.similarity_threshold = similarity_threshold
self.ttl = timedelta(hours=ttl_hours)
def _get_embedding(self, query: str) -> np.ndarray:
"""获取查询的向量表示（需要集成实际的嵌入模型）"""
# 简化示例：使用哈希作为占位符
# 实际应用中应使用: embedding_model.encode(query)
return np.random.rand(384) # 示例维度
def get(self, query: str) -> Tuple[any, bool]:
"""
从缓存中获取结果
Returns:
(result, cache_hit): 如果命中缓存则返回结果，否则返回 None
"""
query_embedding = self._get_embedding(query)
for cache_key, cached_item in self.cache.items():
# 检查 TTL
if datetime.now() - cached_item['timestamp'] > self.ttl:
del self.cache[cache_key]
continue
# 计算相似度
cached_embedding = cached_item['embedding']
similarity = cosine_similarity(query_embedding, cached_embedding)
if similarity >= self.similarity_threshold:
return cached_item['result'], True
return None, False
def set(self, query: str, result: any):
"""将查询和结果存入缓存"""
cache_key = hashlib.md5(query.encode()).hexdigest()
self.cache[cache_key] = {
'result': result,
'embedding': self._get_embedding(query),
'timestamp': datetime.now()
}
使用示例
cache = SemanticCache(similarity_threshold=0.95)
第一次查询
result1, hit1 = cache.get("人工智能的未来发展趋势")
if not hit1:
# 调用 API 获取结果
result1 = {"answer": "AI 将朝着通用人工智能方向发展..."}
cache.set("人工智能的未来发展趋势", result1)
print(f"缓存命中: {hit1}") # False实战案例：电商客服 Agent 的检索优化
某电商平台的客服 Agent 每天处理 10,000+ 咨询，问题重复率高达 60%。通过实施语义缓存，系统自动识别相似问题并返回缓存答案，在不影响响应质量的前提下，将 API 调用量降低 45%，每月节省成本约 $2,000。
性能优化最佳实践
- 向量维度选择：根据任务复杂度选择合适的嵌入维度。简单任务使用 384 维，高精度需求使用 1536 维。
- 批量处理：将多个查询批量处理以提高吞吐量。
- 索引优化：使用 HNSW、IVF 等索引算法加速大规模向量搜索。
- 异步处理：对非关键路径的结果进行异步加载，提升首屏响应速度。
总结
向量相似度搜索是 AI Agent 实现高质量记忆检索的基础。通过合理选择相似度算法、实施混合搜索、引入重排序机制和语义缓存，可以显著提升检索的准确性和效率，从而为用户提供更智能、更快速的响应体验。
👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน