Пам’ять гібридного агента: поєднання 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 — це стандартна константа; вона зменшує вплив будь-якого окремого топового рангу, тож документи, які посідають високі місця у обох списках, перевершують ті, що посідають перше місце лише в одному з них.