混合智能体记忆机制：在Python中结合BM25、向量搜索与RRF技术
了解为何纯向量搜索无法用作智能体记忆，Reciprocal Rank Fusion如何在Python中整合BM25与密集型搜索结果，以及何时使用GraphRAG摘要会更有效。
来源: State Checkpointing & Resiliency: Fault-Tolerant ReAct Engines
那些需要处理持续数天或数周的复杂工作流的智能体，必须拥有一个能够可靠查询的记忆系统。虽然向量存储通过嵌入相似度搜索这一默认方案看似可行，但在技术领域中很快就会出现问题：错误代码和函数名称会丢失，跨文档的查询得不到回应，提示词中也充斥着彼此关联度不高的内容。本指南将介绍如何结合关键词检索与密集检索技术，并运用互反排名融合（RRF）方法，同时展示一个简洁的Python实现方案及其局限性，还会说明在何处加入GraphRAG层能起到辅助作用。
为何仅靠向量搜索不够
纯粹的密集检索方式，通常被称为简单的RAG模型，作为智能体的长期记忆系统时存在三个反复出现的缺陷：
- 精确标识符会被模糊处理。嵌入模型虽能捕捉含义，但对实体ID、型号、用户ID及函数名等精确标记的处理能力较差。查询某个错误代码时，可能会返回与它外观相似的其他错误代码的相关信息。
- 缺乏全局视角。检索结果仅返回与查询内容相近的孤立片段，无法整合整个语料库中的关联信息。
- 上下文冗余问题。将原始检索结果直接粘贴到提示词中会占用大量标记，且无法保证这些内容与查询主题相关。
常见的生产环境解决方案是分层处理的：混合搜索同时使用BM25和密集向量，通过RRF将二者合并，并为宽泛的问题添加GraphRAG知识图谱。如需了解向量方面的相关内容，请参阅向量数据库的运作原理，从嵌入到混合搜索。
三种检索方法及其优势
- 密集向量搜索（HNSW或扁平索引）将文本映射到嵌入空间，从而识别意图和概念上的相似性，即便表述有所不同也能实现匹配。
- 稀疏关键词搜索（BM25或TF-IDF）通过精确匹配词元来评分，这对于错误代码、唯一标识符、函数签名和专有名词而言至关重要。
混合内存查询的形态
智能体在查询其内存时会发送一个请求，该请求为两种引擎都设置了参数：
{
"query_id": "mem_req_99214",
"search_text": "HTTP 504 gateway timeout postgres connection pool",
"top_k": 5,
"hybrid_params": {
"dense_weight": 0.6,
"sparse_weight": 0.4,
"rrf_k_constant": 60
},
"metadata_filter": {
"environment": "production",
"service": "billing-service"
}
}
除了查询文本和top_k之外，请求数据中还包含每种引擎的权重、RRF常数k，以及用于将结果限制在特定环境和服务范围内的元数据过滤器。过滤非常重要：生产环境中的问题不应用测试环境的说明来回答。
融合后的响应会返回带有来源信息的得分片段：
{
"query_id": "mem_req_99214",
"total_retrieved": 2,
"results": [
{
"chunk_id": "doc_8832_p4",
"rrf_score": 0.0322,
"text": "PostgreSQL connection pool exhaustion causes HTTP 504 timeouts. Increase max_connections to 200.",
"source": "wiki/infrastructure_troubleshooting.md"
}
]
}
rrf_score的设计值本来就较小。当k = 60时，某个数据块在其中一个列表中排名第一，在另一个列表中排名第三，其得分将为1/61 + 1/63，约为0.032。这些分数仅在同一查询内部相互比较时才有意义，因此应避免使用绝对阈值。（示例中total_retrieved显示为2，但实际上只列出了一个结果；此数值仅用于说明。）
在Python中实现混合搜索与RRF算法
下面的实现使用了Pydantic和asyncio以及模拟的后端，因此它展示的是融合逻辑，而非可直接使用的组件。
数据块与排序结果的数据模型
SearchResult 是每个搜索引擎返回的结果，其中包含其排名（rank）和原始得分；而 MemoryChunk 则是传递给智能体的融合后输出。尽管融合过程会忽略 raw_score，但保留它仍有助于调试。
import math
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
class MemoryChunk(BaseModel):
chunk_id: str
content: str
metadata: Dict[str, Any] = Field(default_factory=dict)
score: float = 0.0
class SearchResult(BaseModel):
chunk_id: str
content: str
rank: int
raw_score: float
融合引擎
RRF 通过计算每篇文档在所有出现列表中 1 / (k + rank) 的总和来为其评分。该类将这一方法应用于向量列表和 BM25 列表，记住每个数据块的内容，按综合得分排序后返回最佳结果：
class ReciprocalRankFusionEngine:
"""
Combines ranked result lists from Dense Vector and BM25 Sparse Search
using the RRF algorithm: RRF_Score = sum(1 / (k + rank))
"""
def __init__(self, k_constant: int = 60):
self.k_constant = k_constant
def fuse_rankings(
self,
vector_results: List[SearchResult],
bm25_results: List[SearchResult],
top_k: int = 5
) -> List[MemoryChunk]:
rrf_scores: Dict[str, float] = {}
content_map: Dict[str, str] = {}
# Process Vector Search Ranks
for item in vector_results:
rrf_scores[item.chunk_id] = rrf_scores.get(item.chunk_id, 0.0) + (
1.0 / (self.k_constant + item.rank)
)
content_map[item.chunk_id] = item.content
# Process BM25 Search Ranks
for item in bm25_results:
rrf_scores[item.chunk_id] = rrf_scores.get(item.chunk_id, 0.0) + (
1.0 / (self.k_constant + item.rank)
)
content_map[item.chunk_id] = item.content
# Sort by combined RRF Score descending
sorted_chunks = sorted(
rrf_scores.items(), key=lambda x: x[1], reverse=True
)
output: List[MemoryChunk] = []
for chunk_id, score in sorted_chunks[:top_k]:
output.append(
MemoryChunk(
chunk_id=chunk_id,
content=content_map[chunk_id],
score=round(score, 5)
)
)
return output
如果某个数据块同时被两个引擎识别，它将获得两次评分，因此一致性会自动得到奖励。需要注意的是，此处并未使用有效载荷的权重：这仅仅是普通的 RRF 算法。若要让 dense_weight 和 sparse_weight 发挥作用，需将每个列表的评分结果乘以其对应的权重，这种变体通常被称为加权 RRF。
协调两种搜索方式
检索器会向每个引擎请求所需数量两倍的候选项，为融合排序提供更多数据，随后将两份列表一并传递给RRF引擎：
class HybridMemoryRetriever:
"""
Orchestrates Vector and BM25 search queries in parallel and applies RRF.
"""
def __init__(self, vector_db_client: Any, bm25_index: Any):
self.vector_db = vector_db_client
self.bm25_index = bm25_index
self.rrf_engine = ReciprocalRankFusionEngine(k_constant=60)
async def retrieve_memory(self, query: str, top_k: int = 5) -> List[MemoryChunk]:
# Fetch rankings from vector space and BM25 index
vector_hits = await self._fetch_vector_results(query, fetch_k=top_k * 2)
bm25_hits = await self._fetch_bm25_results(query, fetch_k=top_k * 2)
# Merge and re-rank using Reciprocal Rank Fusion
fused_memory = self.rrf_engine.fuse_rankings(
vector_results=vector_hits,
bm25_results=bm25_hits,
top_k=top_k
)
return fused_memory
async def _fetch_vector_results(self, query: str, fetch_k: int) -> List[SearchResult]:
# Simulated vector DB query
return [
SearchResult(chunk_id="doc_102", content="Postgres timeout handling guide", rank=1, raw_score=0.92),
SearchResult(chunk_id="doc_883", content="Database connection pooling configuration", rank=2, raw_score=0.85)
]
async def _fetch_bm25_results(self, query: str, fetch_k: int) -> List[SearchResult]:
# Simulated BM25 keyword query
return [
SearchResult(chunk_id="doc_883", content="Database connection pooling configuration", rank=1, raw_score=14.2),
SearchResult(chunk_id="doc_401", content="HTTP 504 status codes and gateways", rank=2, raw_score=11.1)
]
文档说明承诺实现并行执行，但实际上两个await调用是顺序执行的；应使用asyncio.gather来让多次请求同时进行。真实系统中的_fetch_方法会调用向量数据库和BM25索引，应用元数据过滤条件，并按照fetch_k的设定获取结果。在模拟数据中，doc_883同时出现在两份列表中，因此尽管它只在一份列表中排在首位，仍会在融合后的排序结果中处于最顶端。
加入GraphRAG以处理全语料库查询
混合搜索仍然只能获取局部片段。而为数千份文档进行推理，GraphRAG会分三个阶段构建结构化层：
- 实体提取。后台流程可识别服务、错误类型及SQL架构等实体，提取其中的显式关联关系，并将它们存储在Neo4j这样的图数据库或NetworkX这样的内存库中。
- 社区检测。Leiden聚类等算法会将紧密相连的节点聚合为不同层级的社区。
- 分层总结。大型语言模型会为每个社区撰写报告，从而从这些总结中回答宏观问题，而无需处理数百万条原始数据。
其代价在于前期成本：建立索引需要大量调用大型语言模型，且随着文档变化，总结内容也会过时。对于需要用全局性问题查询的大型、相对稳定的语料库，GraphRAG能发挥显著优势。
常见问题
为何选择RRF而非平均分值？
密集分数与稀疏分数处于不同的量级上。实际应用中，余弦相似度通常介于0和1之间，而BM25分数则没有上限，常常超过50。对它们进行归一化会引入易受数据变化影响的敏感参数。RRF仅使用位置信息，因此无需调整即可在不同数据集中保持稳定性。
GraphRAG是如何减少提示词大小的？
代理程序无需插入数十个原始片段（比如50个），只需读取一两个预先计算好的社区摘要。据称这样可以减少约70%的标记数量，但实际效果取决于你的语料库和查询内容，需自行测试。
关键要点
- 在技术领域，可将嵌入向量与BM25结合使用，并用RRF合并排序结果。
k = 60 是常用的常数；它能够削弱任何单一的最高排名，因此同时在两个列表中排名靠前的文档会优于仅在其中一个列表中排第一的文档。