Pamięć agenta hybrydowego: łączenie BM25 i wyszukiwania wektorowego z RRF w Pythonie

Agent, który przetwarza długie procesy przez dni lub tygodnie, potrzebuje pamięci, do której może w sposób niezawodny uzyskiwać dostęp, a domyślny wybór – magazyn wektorów wyszukiwany za pomocą podobieństwa embeddingów – szybko zawodzi w dziedzinach technicznych. Kody błędów i nazwy funkcji giną, pytania dotyczące różnych dokumentów pozostają bez odpowiedzi, a prompty wypełniają się słabo powiązanymi fragmentami tekstu. Ten przewodnik pokazuje, jak połączyć wyszukiwanie według słów kluczowych z gęstym wyszukiwaniem za pomocą Reciprocal Rank Fusion (RRF), omawia zwięzłą implementację w Pythonie oraz jej ograniczenia, a także wyjaśnia, w jakich przypadkach pomocna jest warstwa GraphRAG.

Dlaczego samo wyszukiwanie wektorowe nie wystarcza

Czyste gęste wyszukiwanie, często nazywane prostym RAG, ma trzy powtarzające się wady jako pamięć długoterminowa agenta:

• Dokładne identyfikatory stają się niewyraźne. Embeddingi potrafią uchwycić znaczenie, ale słabo radzą sobie z precyzyjnymi tokenami, takimi jak identyfikatory entytetów, numery modeli, identyfikatory użytkowników i nazwy funkcji. Zapytanie dotyczące jednego kodu błędu może zwrócić fragmenty informacji o podobnym kodzie.
• Brak ogólnej perspektywy. Proces wydobywania informacji zwraca izolowane fragmenty bliskie zapytaniu i nie jest w stanie połączyć relacji rozproszonych w całym korpusie.
• Nadmierna ilość kontekstu. Wklejanie surowych wyników do promptu marnuje tokeny bez żadnej gwarancji ich istotności.

Zwykła odpowiedź producenta jest wielowarstwowa: hybrydowe wyszukiwanie wykorzystuje BM25 razem z gęstymi wektorami, łączy je za pomocą RRF oraz dodaje graf wiedzy GraphRAG do obsługi szerokiego zakresu pytań. Aby ponownie zapoznać się z aspektem wektorowym, sprawdź jak faktycznie działają bazy danych wektorowych, od embeddingów po hybrydowe wyszukiwanie.

Trzy metody wydobywania informacji i to, w czym każda z nich się sprawdza

• Hybrydowe wyszukiwanie wektorowe (HNSW lub proste indeksy) przekształca tekst do przestrzeni embeddingów i porównuje intencję oraz podobieństwo koncepcyjne, nawet gdy sformułowanie się różni.
• Szare wyszukiwanie słów kluczowych (BM25 lub TF-IDF) ocenia dokładne dopasowania tokenów, co jest niezbędne dla kodów błędów, unikalnych identyfikatorów, sygnatur funkcji oraz nazw własnych.

GraphRAG wydobywa entity i relacje do grafu wiedzy oraz nadaje się do syntetyzowania pytań takich jak „Jak wszystkie usługi w Klasterze B radzą sobie z timeoutami bazy danych?”. Takie grafy to ogólne grafy skierowane, niekoniecznie acykliczne.

Kształt zapytania do hybrydowej pamięci

Agent konsultujący swoją pamięć wysyła jedno żądanie, które parametryzuje oba silniki:

{
  "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"
  }
}

Oprócz tekstu zapytania i top_k, treść żądania zawiera wagi dla każdego silnika, stałą RRF k oraz filtr metadanych ograniczający wyniki do jednego środowiska i usługi. Filtrowanie jest ważne: incydent w produkcji nie powinien być odpowiadany na podstawie notatek z środowiska testowego.

Fuzjonowany odpowiedź zwraca fragmenty ocenione pod kątem jakości wraz z ich źródłem:

{
  "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 jest celowo niski. Przy k = 60 element, który zajmuje pierwsze miejsce w jednej liście i trzecie w drugiej, otrzymuje wynik 1/61 + 1/63, co odpowiada mniej więcej 0,032. Te wartości mają sens tylko w odniesieniu do siebie w ramach jednego zapytania, dlatego należy unikać bezwzględnych progów. (Przykład pokazuje total_retrieved jako 2, ale podaje tylko jeden wynik; należy to traktować jako ilustrację.)

Wdrożenie hybrydowego wyszukiwania i RRF w Pythonie

Poniższa implementacja wykorzystuje Pydantic i asyncio wraz z symulowanymi backendami, więc pokazuje logikę fuzji, a nie gotowy komponent do bezpośredniego użycia.

Modele danych dla elementów i wyników posortowanych

SearchResult to wynik zwracany przez każdy silnik, zawierający jego pozycję (rank) oraz oryginalną ocenę; MemoryChunk to połączony wynik przekazywany agentowi. Zachowanie wartości raw_score ułatwia debugowanie, mimo że proces fuzji ją ignoruje.

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

Silnik fuzji

RRF ocenia każdy dokument jako sumę wartości 1 / (k + rank) dla wszystkich list, w których się pojawił. Klasa ta stosuje tę metodę do list wektorowych i BM25, zapamiętuje treść każdego fragmentu, sortuje według łącznej oceny i zwraca najlepsze wyniki:

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

Fragment znaleziony przez oba silniki otrzymuje dwa wkłady, dzięki czemu zgodność jest automatycznie nagradzana. Należy zauważyć, że wagi zawarte w danych nie są wykorzystywane – to zwykły RRF. Aby dense_weight i sparse_weight miały znaczenie, należy pomnożyć wkład każdej listy przez jej wagę; taka wersja nazywana jest zazwyczaj RRF z wagami.

Koordynacja obu wyszukiwań

Narzędzie pobierające dane prosi każdy silnik o dwa razy więcej kandydatów, niż jest potrzebne, co daje narzędziu łączącemu więcej materiału do przetworzenia, a następnie przekazuje obie listy silnikowi 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)
        ]

Docstring obiecuje wykonywanie zadań równolegle, ale dwie funkcje await są wykonywane sekwencyjnie; należy użyć asyncio.gather, aby skrócić czas oczekiwania. W rzeczywistym systemie metody _fetch_ wywoływałyby bazę danych wektorową i indeks BM25, zastosowały filtr metadanych oraz uwzględniły wartość fetch_k. W symulowanych danych doc_883 pojawia się w obu listach, dlatego znajduje się na szczycie łączonej listy rankingowej, mimo że w jednej z list jest pierwszy.

Dodanie GraphRAG do pytań obejmujących cały korpus

Hybrydowe wyszukiwanie nadal pobiera lokalne fragmenty tekstu. Aby umożliwić rozumowanie na podstawie tysięcy dokumentów, GraphRAG buduje ustrukturyzowaną warstwę w trzech etapach:

• Ekstrakcja entytetów. Tło procesowe identyfikuje takie entytety jak usługi, typy błędów i schematy SQL, wyodrębnia jawne relacje oraz przechowuje je w bazie danych grafowych takich jak Neo4j lub w bibliotekach pamięci operacyjnej, np. NetworkX.
• Detekcja społeczności. Algorytmy klastrujące, takie jak Leiden, grupują ściśle powiązane węzły w hierarchię społeczności.
• Sumaryzacja hierarchiczna. Sztuczna inteligencja pisze raport dla każdej społeczności, a szerokie pytania są odpowiadane na podstawie tych streszczeń, zamiast na podstawie milionów surowych fragmentów tekstu.

Kompromisem jest początkowy koszt: indeksowanie wymaga wielu wywołań sztucznej inteligencji, a streszczenia stają się przestarzałe wraz ze zmianami dokumentów. GraphRAG okazuje się korzystny przy obsłudze dużych, względnie stabilnych zbiorków danych, do których zadawane są pytania ogólne.

Częste pytania

Dlaczego używać RRF zamiast średnicy ocen?

Wartości gęste i rzadkie istnieją na niepowiązanych skalach. W praktyce podobieństwo kosinowe zazwyczaj mieści się w przedziale od 0 do 1, natomiast wartości BM25 są nieograniczone i często przekraczają 50. Ich normalizacja wprowadza kruche parametry, które ulegają zmianie wraz ze zmianami danych. RRF wykorzystuje jedynie pozycje, co zapewnia jego odporność na różne zbiory danych bez konieczności dostosowywania.

Jak GraphRAG zmniejsza rozmiar promptu?

Zamiast wstawiać dziesiątki surowych fragmentów, powiedzmy 50, agent czyta jeden lub dwa uprzednio obliczone streszczenia społeczności. Wspomina się o oszczędnościach sięgających około 70 procent mniej tokenów, ale wyniki zależą od Twojego korpusu i zapytań, więc zmierz je sam.

Główne wnioski

• W dziedzinach technicznych łącz embeddingi z BM25 oraz łącz rankingi za pomocą RRF.

k = 60 to standardowa stała; tłumi ona wpływ jakiegokolwiek pojedynczego najwyższego rankingu, dzięki czemu dokumenty dobrze sklasyfikowane w obu listach przewyższają te, które zajmują pierwsze miejsce tylko w jednej z nich.

RRF ignoruje surowe wyniki; należy dodać wyraźne wagi, jeśli jeden silnik powinien mieć większy wpływ, oraz przefiltrować metadane przed fuzją.

Należy przeprowadzić obie wyszukiwania jednocześnie i sprawdzić, czy kod rzeczywiście jest „paralelny”.

Dla rozumowania na poziomie całego systemu nad dużymi zbiorkami danych należy korzystać ze streszczeń społeczności GraphRAG, uwzględniając przy tym koszt indeksowania.