Hybride Agenten-Memory-Strategie: Kombination von BM25 und Vektorabfrage mit RRF in Python
Erfahren Sie, warum reine Vektorabfragen als Agentenmemorie versagen, wie Reciprocal Rank Fusion BM25 und dichte Ergebnisse in Python kombiniert, und wann GraphRAG-Zusammenfassungen hilfreich sind.
Quellen: State Checkpointing & Resiliency: Fault-Tolerant ReAct Engines
Ein Agent, der langwierige Workflows über Tage oder Wochen ausführt, benötigt ein Speichersystem, auf das er zuverlässig zugreifen kann. Die Standardlösung, ein Vektorlagertyp, der durch ähnlichkeitsbasierte Suche abgerufen wird, versagt schnell in technischen Bereichen. Fehlercodes und Funktionsnamen gehen verloren, Fragen zwischen Dokumenten bleiben unbeantwortet, und die Eingaben füllen sich mit nur schwach miteinander verbundenen Inhalten. Dieser Leitfaden zeigt, wie Schlüsselwort- und dichte Suchverfahren mit Reciprocal Rank Fusion (RRF) kombiniert werden können, führt durch eine kompakte Python-Implementierung sowie deren Einschränkungen und erläutert, wo eine GraphRAG-Schicht hilft.
Warum reine Vektor Suche nicht ausreicht
Pure dichte Suchverfahren, oft als naives RAG bezeichnet, weisen als langfristiges Speichersystem eines Agents drei wiederkehrende Schwächen auf:
- Genaue Identifikatoren werden unklar. Embeddings erfassen die Bedeutung, können aber präzise Token wie Entity-IDs, Modellnummern, Benutzer IDs und Funktionsnamen nur unzureichend verarbeiten. Eine Abfrage nach einem Fehlercode kann Teileinformationen zu einem ähnlich aussehenden Code zurückgeben.
- Keine globale Sichtweise. Die Auswertung liefert isolierte Teileinheiten in der Nähe der Abfrage und kann keine Beziehungen zusammenfassen, die sich über ein ganzes Korpus erstrecken.
- Kontextüberlastung. Das Einfügen unverarbeiteter Ergebnisse in die Anfrage verbraucht Token, ohne dass eine Relevanz garantiert ist.
Die übliche Produktlösung ist schichtweise aufgebaut: Die hybride Suche verwendet BM25 zusammen mit dichten Vektoren, fügt sie mithilfe von RRF zusammen und ergänzt sie durch ein GraphRAG-Wissensgraph für umfassendere Fragen. Für einen Überblick zum Vektoraspekt siehe wie Vektordatenbanken tatsächlich funktionieren, von Embeddings bis hin zur hybriden Suche.
Drei Suchmethoden und ihre Stärken
- Dichte Vektorsuche (HNSW oder flache Indizes) überträgt Text in einen Embeddings-Raum und erkennt Absichten sowie konzeptionelle Ähnlichkeiten – auch dann, wenn die Formulierung unterschiedlich ist.
- Sparse-Schlüsselwortsuche (BM25 oder TF-IDF) bewertet exakte Tokenübereinstimmungen, was für Fehlercodes, eindeutige IDs, Funktionssignaturen und Eigennamen unverzichtbar ist.
Die Struktur einer hybriden Speicherabfrage
Ein Agent, der sein Gedächtnis abruft, sendet eine Anfrage, die beide Engines parametriert:
{
"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"
}
}
Neben dem Abfragetext und top_k enthält die Nutzlast Gewichte für jede Engine, die RRF-Konstante k sowie einen Metadatenfilter, der die Ergebnisse auf ein bestimmtes Umfeld und einen Dienst beschränkt. Das Filtern ist wichtig: Ein Produktionsproblem sollte nicht mit Notizen aus der Staging-Umgebung beantwortet werden.
Die zusammengeführte Antwort liefert bewertete Abschnitte zusammen mit ihrer Quelle:
{
"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 ist aus Designgründen klein. Bei k = 60 erhält ein Datensatz, der in einer Liste an erster Stelle und in einer anderen an dritter Stelle rangiert, die Punktzahl 1/61 + 1/63, was etwa 0,032 entspricht. Diese Werte sind nur im Verhältnis zueinander innerhalb einer einzigen Abfrage sinnvoll, daher sollten absolute Schwellenwerte vermieden werden. (Das Beispiel gibt total_retrieved mit 2 an, listet aber nur ein Ergebnis auf; es dient lediglich zur Veranschaulichung.)
Implementierung von Hybrid-Suche und RRF in Python
Die untenstehende Implementierung verwendet Pydantic und asyncio zusammen mit simulierten Backend-Systemen, wodurch sie eher die Fusionlogik demonstriert als eine direkt einsetzbare Komponente.
Datentypen für Datensätze und gerangierte Ergebnisse
SearchResult ist das, was jede Suchmaschine zurückgibt – zusammen mit ihrer Position (rank) und dem ursprünglichen Score; MemoryChunk ist die zusammengeführte Ausgabe, die an den Agenten übergeben wird. Die Beibehaltung von raw_score hilft beim Debuggen, obwohl die Fusion diesen Wert ignoriert.
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
Der Fusion-Engine
RRF bewertet jedes Dokument als Summe von 1 / (k + rank) für alle Listen, in denen es vorkommt. Die Klasse wendet dies auf die Vektor- und BM25-Listen an, speichert den Inhalt jedes Chunks, sortiert nach dem kombinierten Score und gibt die besten Ergebnisse zurück:
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
Ein Chunk, der von beiden Engines gefunden wird, erhält zwei Beiträge, wodurch eine Übereinstimmung automatisch belohnt wird. Beachten Sie, dass die Gewichte des Payloads nicht verwendet werden – es handelt sich dabei um reines RRF. Um dense_weight und sparse_weight relevant zu machen, muss der Beitrag jeder Liste mit ihrem Gewicht multipliziert werden; eine solche Variante wird in der Regel als gewichtetes RRF bezeichnet.
Orchestrierung beider Suchvorgänge
Der Retriever bittet jede Suchmaschine um doppelt so viele Ergebnisse, wie er benötigt, wodurch dem Fusionssystem mehr Material zur Verfügung steht, und gibt anschließend beide Listen an den RRF-Engine weiter:
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)
]
Die Dokumentation verspricht eine parallele Ausführung, doch die beiden await-Aufrufe laufen sequenziell; verwenden Sie asyncio.gather, um die Ausführungen zu überlappen. Die _fetch_-Methoden eines echten Systems würden Ihre Vektordatenbank sowie den BM25-Index aufrufen, den Metadatenfilter anwenden und fetch_k berücksichtigen. In den simulierten Daten erscheint doc_883 in beiden Listen, weshalb es trotzdem an erster Stelle der gefusionierten Rangliste steht, obwohl es in nur einer Liste an erster Stelle war.
Hinzufügen von GraphRAG für Fragen im gesamten Korpus
Die hybride Suche holt weiterhin lokale Datenschnitte ab. Für das Schlussfolgern über Tausende von Dokumenten erstellt GraphRAG in drei Schritten eine strukturierte Ebene:
- Entitätenextraktion. Ein Hintergrundprozess identifiziert Entitäten wie Dienste, Fehlerarten und SQL-Schemata, extrahiert explizite Beziehungen und speichert sie in einer Graph-Datenbank wie Neo4j oder in einer In-Memory-Bibliothek wie NetworkX.
- Gemeinschaftserkennung. Clustering-Algorithmen wie Leiden gruppieren eng miteinander verbundene Knoten zu einer Hierarchie von Gemeinschaften.
- Hierarchische Zusammenfassung. Ein LLM erstellt einen Bericht für jede Gemeinschaft, wodurch umfassende Fragen anhand dieser Zusammenfassungen statt anhand von Millionen von Rohdaten beantwortet werden können.
Der Kompromiss liegt in den Anfangskosten: Das Indexieren erfordert viele Aufrufe des LLM, und die Zusammenfassungen veralten, wenn sich die Dokumente ändern. GraphRAG lohnt sich bei großen, relativ stabilen Korpora, die mit globalen Fragen abgefragt werden.
Häufige Fragen
Warum RRF statt der Durchschnittsberechnung der Scores verwenden?
Dichte und spärliche Bewertungswerte existieren auf unverwandten Skalen. Die Kosinusähnlichkeit liegt in der Praxis in der Regel zwischen 0 und 1, während die BM25-Bewertungen unbegrenzt sind und oft über 50 liegen. Die Normalisierung führt zu empfindlichen Parametern, die bei Datenänderungen fehlschlagen. RRF verwendet ausschließlich Positionen, wodurch es ohne Anpassungen in verschiedenen Datensätzen robust bleibt.
Wie verringert GraphRAG die Größe der Anfragen?
Anstelle von Dutzenden roher Textabschnitte, sagen wir 50, liest der Agent ein oder zwei im Voraus berechnete Zusammenfassungen. Es werden Einsparungen von bis zu etwa 70 Prozent an Token angegeben, doch die Ergebnisse hängen vom jeweiligen Korpus und den Anfragen ab, daher sollten Sie sie selbst messen.
Kernpunkte
- In technischen Bereichen sollten Embeddings mit BM25 kombiniert und die Ranglisten mit RRF verschmolzen werden.
k = 60 ist die übliche Konstante; sie dämpft jede einzelne Spitzenplatzierung, sodass Dokumente, die in beiden Listen gut platziert sind, jene übertrumpfen, die nur in einer der Listen an erster Stelle stehen.