Память гибридного агента: объединение BM25 и векторного поиска с RRF на Python
Узнайте, почему чистый векторный поиск не подходит в качестве памяти агента, как метод Reciprocal Rank Fusion объединяет результаты BM25 и плотных поисков в Python, и когда резюме GraphRAG бывают полезны.
Источники: State Checkpointing & Resiliency: Fault-Tolerant ReAct Engines
Агенту, который выполняет длительные рабочие процессы в течение дней или недель, необходима память, к которой можно было бы надежно обращаться, однако стандартный вариант — хранилище векторов с поиском по сходству эмбеддингов — быстро теряет эффективность в технических областях. Коды ошибок и названия функций теряются, вопросы, касающиеся нескольких документов, остаются без ответа, а промпты заполняются нерелевантными фрагментами информации. В этом руководстве показано, как сочетать поиск по ключевым словам и плотный поиск с методом Reciprocal Rank Fusion (RRF), приводится краткая реализация на Python вместе с её ограничениями, а также объясняется, где полезен слой GraphRAG.
Почему один только векторный поиск недостаточен
Чистый плотный поиск, часто называемый примитивным RAG, имеет три постоянных недостатка в качестве долгосрочной памяти агента:
- Точные идентификаторы становятся нечитаемыми. Методы эмбеддингов улавливают смысл, но плохо справляются с точными токенами, такими как идентификаторы сущностей, номера моделей, идентификаторы пользователей и названия функций. Запрос на один код ошибки может вернуть информацию, связанную с похожим кодом.
- Отсутствие общего вида. Методы поиска возвращают изолированные фрагменты, близкие к запросу, и не могут собрать информацию о взаимосвязях, распределенных по всему корпусу.
- Чрезмерная нагрузка на контекст. Вставка необработанных результатов в промпт тратит токены без гарантии их релевантности.
Стандартный подход к обработке запросов основан на многоуровневой структуре: гибридный поиск использует алгоритм 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, но приводится всего один результат; рассматривайте это как иллюстративный пример.)
Реализация гибридного поиска и алгоритма RRF на Python
Приведённая ниже реализация использует 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 Group, объединяют тесно связанные узлы в иерархию сообществ.
- Иерархическое краткое изложение. Большая языковая модель пишет отчет для каждого сообщества, и на основе этих кратких изложений отвечают на общие вопросы, вместо того чтобы анализировать миллионы исходных фрагментов текста.
Компромисс заключается в начальных затратах: индексация требует множества вызовов больших языковых моделей, а краткие изложения устаревают по мере изменения документов. GraphRAG оправдывает себя при работе с большими, относительно стабильными корпусами данных, на которые задаются глобальные вопросы.
Часто задаваемые вопросы
Почему использовать RRF вместо среднего значения оценок?
Показатели плотности и разреженности существуют на несвязанных шкалах. В практике коэффициент косинусного сходства обычно находится в диапазоне от 0 до 1, тогда как значения BM25 неограничены и часто превышают 50. Нормализация этих показателей приводит к появлению чувствительных параметров, которые меняются при изменении данных. RRF использует только позиции, что обеспечивает его устойчивость к различным наборам данных без необходимости настройки.
Как GraphRAG сокращает размер запросов?
Вместо вставки десятков необработанных фрагментов, скажем, 50, агент читает один или два заранее вычисленных кратких обзора сообществ. Утверждается, что это позволяет сократить количество токенов примерно на 70 процентов, однако результаты зависят от вашего корпуса данных и запросов, поэтому измеряйте их сами.
Основные выводы
- В технических областях следует сочетать эмбеддинги с BM25 и объединять рейтинги с помощью RRF.
k = 60 — это стандартная константа; она снижает влияние любого отдельного высокого ранга, поэтому документы, занимающие высокие позиции в обоих списках, превосходят те, что на первом месте только в одном из них.