Milvus Hybrid Search Retriever

La ricerca ibrida combina i punti di forza di diversi paradigmi di ricerca per migliorare l'accuratezza e la robustezza del recupero. Sfrutta le capacità della ricerca vettoriale densa e della ricerca vettoriale rada, nonché le combinazioni di più strategie di ricerca vettoriale densa, assicurando un recupero completo e preciso per diverse query.

Questo diagramma illustra lo scenario di ricerca ibrida più comune, ovvero la ricerca ibrida densa + rada. In questo caso, i candidati vengono recuperati utilizzando sia la similarità semantica dei vettori sia la corrispondenza precisa delle parole chiave. I risultati di questi metodi vengono uniti, riclassificati e passati a un LLM per generare la risposta finale. Questo approccio bilancia precisione e comprensione semantica, rendendolo molto efficace per diversi scenari di interrogazione.

Oltre alla ricerca ibrida densa + rada, le strategie ibride possono anche combinare più modelli vettoriali densi. Ad esempio, un modello vettoriale denso potrebbe essere specializzato nella cattura delle sfumature semantiche, mentre un altro si concentra sulle incorporazioni contestuali o sulle rappresentazioni specifiche del dominio. Unendo i risultati di questi modelli e riclassificandoli, questo tipo di ricerca ibrida garantisce un processo di recupero più sfumato e consapevole del contesto.

L'integrazione di LangChain Milvus offre un modo flessibile di implementare la ricerca ibrida, supportando un numero qualsiasi di campi vettoriali e di modelli di incorporamento densi o sparsi personalizzati, il che consente a LangChain Milvus di adattarsi in modo flessibile a vari scenari di utilizzo della ricerca ibrida e allo stesso tempo di essere compatibile con le altre funzionalità di LangChain.

In questo tutorial, inizieremo con il caso più comune di dense + sparse, per poi introdurre una serie di approcci generali all'uso della ricerca ibrida.

Prerequisiti

Prima di eseguire questo notebook, assicurarsi di avere installato le seguenti dipendenze:

$ pip install --upgrade --quiet  langchain langchain-core langchain-community langchain-text-splitters langchain-milvus langchain-openai bs4 pymilvus[model] #langchain-voyageai

Utilizzeremo i modelli di OpenAI. È necessario preparare le variabili d'ambiente OPENAI_API_KEY da OpenAI.

import os

os.environ["OPENAI_API_KEY"] = "sk-***********"

Specificare il server Milvus URI (e facoltativamente TOKEN). Per sapere come installare e avviare il server Milvus, seguire questa guida.

URI = "http://localhost:19530"
# TOKEN = ...

Preparare alcuni documenti di esempio, che sono riassunti di storie di fantasia classificate per tema o genere.

from langchain_core.documents import Document

docs = [
    Document(
        page_content="In 'The Whispering Walls' by Ava Moreno, a young journalist named Sophia uncovers a decades-old conspiracy hidden within the crumbling walls of an ancient mansion, where the whispers of the past threaten to destroy her own sanity.",
        metadata={"category": "Mystery"},
    ),
    Document(
        page_content="In 'The Last Refuge' by Ethan Blackwood, a group of survivors must band together to escape a post-apocalyptic wasteland, where the last remnants of humanity cling to life in a desperate bid for survival.",
        metadata={"category": "Post-Apocalyptic"},
    ),
    Document(
        page_content="In 'The Memory Thief' by Lila Rose, a charismatic thief with the ability to steal and manipulate memories is hired by a mysterious client to pull off a daring heist, but soon finds themselves trapped in a web of deceit and betrayal.",
        metadata={"category": "Heist/Thriller"},
    ),
    Document(
        page_content="In 'The City of Echoes' by Julian Saint Clair, a brilliant detective must navigate a labyrinthine metropolis where time is currency, and the rich can live forever, but at a terrible cost to the poor.",
        metadata={"category": "Science Fiction"},
    ),
    Document(
        page_content="In 'The Starlight Serenade' by Ruby Flynn, a shy astronomer discovers a mysterious melody emanating from a distant star, which leads her on a journey to uncover the secrets of the universe and her own heart.",
        metadata={"category": "Science Fiction/Romance"},
    ),
    Document(
        page_content="In 'The Shadow Weaver' by Piper Redding, a young orphan discovers she has the ability to weave powerful illusions, but soon finds herself at the center of a deadly game of cat and mouse between rival factions vying for control of the mystical arts.",
        metadata={"category": "Fantasy"},
    ),
    Document(
        page_content="In 'The Lost Expedition' by Caspian Grey, a team of explorers ventures into the heart of the Amazon rainforest in search of a lost city, but soon finds themselves hunted by a ruthless treasure hunter and the treacherous jungle itself.",
        metadata={"category": "Adventure"},
    ),
    Document(
        page_content="In 'The Clockwork Kingdom' by Augusta Wynter, a brilliant inventor discovers a hidden world of clockwork machines and ancient magic, where a rebellion is brewing against the tyrannical ruler of the land.",
        metadata={"category": "Steampunk/Fantasy"},
    ),
    Document(
        page_content="In 'The Phantom Pilgrim' by Rowan Welles, a charismatic smuggler is hired by a mysterious organization to transport a valuable artifact across a war-torn continent, but soon finds themselves pursued by deadly assassins and rival factions.",
        metadata={"category": "Adventure/Thriller"},
    ),
    Document(
        page_content="In 'The Dreamwalker's Journey' by Lyra Snow, a young dreamwalker discovers she has the ability to enter people's dreams, but soon finds herself trapped in a surreal world of nightmares and illusions, where the boundaries between reality and fantasy blur.",
        metadata={"category": "Fantasy"},
    ),
]

Incorporamento denso + Incorporamento sparso

Opzione 1 (consigliata): incorporazione densa + funzione incorporata Milvus BM25

Utilizzare l'incorporazione densa + la funzione incorporata di Milvus BM25 per assemblare l'istanza di archivio vettoriale ibrido.

from langchain_milvus import Milvus, BM25BuiltInFunction
from langchain_openai import OpenAIEmbeddings


vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    builtin_function=BM25BuiltInFunction(),  # output_field_names="sparse"),
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`). See https://milvus.io/docs/tune_consistency.md#Consistency-Level for more details.
    drop_old=False,
)

Nel codice qui sopra, definiamo un'istanza di BM25BuiltInFunction e la passiamo all'oggetto Milvus. BM25BuiltInFunction è una classe leggera per il wrapper di Function in Milvus. Possiamo usarla con OpenAIEmbeddings per inizializzare un'istanza di archivio vettoriale Milvus a ricerca ibrida densa + rada.

BM25BuiltInFunction La classe non richiede al cliente di passare il corpus o l'addestramento; tutto viene elaborato automaticamente dal server Milvus, quindi gli utenti non devono preoccuparsi del vocabolario e del corpus. Inoltre, gli utenti possono anche personalizzare l'analizzatore per implementare l'elaborazione del testo personalizzato in BM25.

Per ulteriori informazioni su BM25BuiltInFunction, consultare le sezioni Full-Text-Search e Using Full-Text Search with LangChain and Milvus.

Opzione 2: Usare l'incorporamento denso e personalizzato di LangChain

È possibile ereditare la classe BaseSparseEmbedding da langchain_milvus.utils.sparse e implementare i metodi embed_query e embed_documents per personalizzare il processo di incorporazione rada. Ciò consente di personalizzare qualsiasi metodo di incorporazione rada sia basato su statistiche di frequenza dei termini (ad esempio BM25) sia su reti neurali (ad esempio SPADE).

Ecco un esempio:

from typing import Dict, List
from langchain_milvus.utils.sparse import BaseSparseEmbedding


class MyCustomEmbedding(BaseSparseEmbedding):  # inherit from BaseSparseEmbedding
    def __init__(self, model_path): ...  # code to init or load model

    def embed_query(self, query: str) -> Dict[int, float]:
        ...  # code to embed query
        return {  # fake embedding result
            1: 0.1,
            2: 0.2,
            3: 0.3,
            # ...
        }

    def embed_documents(self, texts: List[str]) -> List[Dict[int, float]]:
        ...  # code to embed documents
        return [  # fake embedding results
            {
                1: 0.1,
                2: 0.2,
                3: 0.3,
                # ...
            }
        ] * len(texts)

Abbiamo una classe demo BM25SparseEmbedding ereditata da BaseSparseEmbedding in langchain_milvus.utils.sparse. È possibile passarla nell'elenco di inizializzazione dell'istanza di Milvus vector store proprio come le altre classi di incorporazione densa di langchain.

# BM25SparseEmbedding is inherited from BaseSparseEmbedding
from langchain_milvus.utils.sparse import BM25SparseEmbedding

embedding1 = OpenAIEmbeddings()

corpus = [doc.page_content for doc in docs]
embedding2 = BM25SparseEmbedding(
    corpus=corpus
)  # pass in corpus to initialize the statistics

vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=[embedding1, embedding2],
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`). See https://milvus.io/docs/tune_consistency.md#Consistency-Level for more details.
    drop_old=False,
)

Sebbene questo sia un modo per usare BM25, richiede che l'utente gestisca il corpus per le statistiche di frequenza dei termini. Si consiglia invece di utilizzare la funzione incorporata di BM25 (opzione 1), che gestisce tutto sul lato server di Milvus. Questo elimina la necessità di gestire il corpus o di addestrare un vocabolario. Per ulteriori informazioni, consultare la sezione Utilizzo della ricerca full-text con LangChain e Milvus.

Definire più campi vettoriali arbitrari

Quando si inizializza l'archivio vettoriale di Milvus, è possibile passare l'elenco degli embeddings (e in futuro anche l'elenco delle funzioni integrate) per implementare il reperimento a più vie e quindi classificare i candidati. Ecco un esempio:

# from langchain_voyageai import VoyageAIEmbeddings

embedding1 = OpenAIEmbeddings(model="text-embedding-ada-002")
embedding2 = OpenAIEmbeddings(model="text-embedding-3-large")
# embedding3 = VoyageAIEmbeddings(model="voyage-3")  # You can also use embedding from other embedding model providers, e.g VoyageAIEmbeddings


vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=[embedding1, embedding2],  # embedding3],
    builtin_function=BM25BuiltInFunction(output_field_names="sparse"),
    # `sparse` is the output field name of BM25BuiltInFunction, and `dense1` and `dense2` are the output field names of embedding1 and embedding2
    vector_field=["dense1", "dense2", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`). See https://milvus.io/docs/tune_consistency.md#Consistency-Level for more details.
    drop_old=False,
)

vectorstore.vector_fields

['dense1', 'dense2', 'sparse']

In questo esempio, abbiamo tre campi vettoriali. Tra questi, sparse è usato come campo di output per BM25BuiltInFunction, mentre gli altri due, dense1 e dense2, sono assegnati automaticamente come campi di output per i due modelli OpenAIEmbeddings (in base all'ordine).

Specificare i parametri degli indici per i campi multivettoriali

Per impostazione predefinita, i tipi di indice di ciascun campo vettoriale sono determinati automaticamente dal tipo di incorporazione o dalla funzione incorporata. Tuttavia, è possibile specificare il tipo di indice per ogni campo vettoriale per ottimizzare le prestazioni della ricerca.

dense_index_param_1 = {
    "metric_type": "COSINE",
    "index_type": "HNSW",
}
dense_index_param_2 = {
    "metric_type": "IP",
    "index_type": "HNSW",
}
sparse_index_param = {
    "metric_type": "BM25",
    "index_type": "AUTOINDEX",
}

vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=[embedding1, embedding2],
    builtin_function=BM25BuiltInFunction(output_field_names="sparse"),
    index_params=[dense_index_param_1, dense_index_param_2, sparse_index_param],
    vector_field=["dense1", "dense2", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`). See https://milvus.io/docs/tune_consistency.md#Consistency-Level for more details.
    drop_old=False,
)

vectorstore.vector_fields

['dense1', 'dense2', 'sparse']

Riclassificazione dei candidati

Dopo la prima fase di recupero, è necessario riclassificare i candidati per ottenere un risultato migliore. Si può scegliere WeightedRanker o RRFRanker a seconda delle esigenze. Per ulteriori informazioni, consultare la sezione Reranking.

Ecco un esempio di reranking ponderato:

vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    builtin_function=BM25BuiltInFunction(),
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`). See https://milvus.io/docs/tune_consistency.md#Consistency-Level for more details.
    drop_old=False,
)

query = "What are the novels Lila has written and what are their contents?"

vectorstore.similarity_search(
    query, k=1, ranker_type="weighted", ranker_params={"weights": [0.6, 0.4]}
)

[Document(metadata={'pk': 454646931479252186, 'category': 'Heist/Thriller'}, page_content="In 'The Memory Thief' by Lila Rose, a charismatic thief with the ability to steal and manipulate memories is hired by a mysterious client to pull off a daring heist, but soon finds themselves trapped in a web of deceit and betrayal.")]

Ecco un esempio di RRF reranking:

vectorstore.similarity_search(query, k=1, ranker_type="rrf", ranker_params={"k": 100})

[Document(metadata={'category': 'Heist/Thriller', 'pk': 454646931479252186}, page_content="In 'The Memory Thief' by Lila Rose, a charismatic thief with the ability to steal and manipulate memories is hired by a mysterious client to pull off a daring heist, but soon finds themselves trapped in a web of deceit and betrayal.")]

Se non si passa alcun parametro per il rerank, viene utilizzata per default la strategia di rerank medio ponderato.

Utilizzo della ricerca ibrida e del reranking in RAG

Nello scenario di RAG, l'approccio più diffuso per la ricerca ibrida è il dense + sparse retrieval, seguito dal reranking. L'esempio seguente mostra un codice end-to-end semplice.

Preparare i dati

Utilizziamo il WebBaseLoader di Langchain per caricare i documenti dalle fonti web e dividerli in pezzi utilizzando il RecursiveCharacterTextSplitter.

import bs4
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Create a WebBaseLoader instance to load documents from web sources
loader = WebBaseLoader(
    web_paths=(
        "https://lilianweng.github.io/posts/2023-06-23-agent/",
        "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    ),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
# Load documents from web sources using the loader
documents = loader.load()
# Initialize a RecursiveCharacterTextSplitter for splitting text into chunks
text_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

# Split the documents into chunks using the text_splitter
docs = text_splitter.split_documents(documents)

# Let's take a look at the first document
docs[1]

Document(metadata={'source': 'https://lilianweng.github.io/posts/2023-06-23-agent/'}, page_content='Fig. 1. Overview of a LLM-powered autonomous agent system.\nComponent One: Planning#\nA complicated task usually involves many steps. An agent needs to know what they are and plan ahead.\nTask Decomposition#\nChain of thought (CoT; Wei et al. 2022) has become a standard prompting technique for enhancing model performance on complex tasks. The model is instructed to “think step by step” to utilize more test-time computation to decompose hard tasks into smaller and simpler steps. CoT transforms big tasks into multiple manageable tasks and shed lights into an interpretation of the model’s thinking process.\nTree of Thoughts (Yao et al. 2023) extends CoT by exploring multiple reasoning possibilities at each step. It first decomposes the problem into multiple thought steps and generates multiple thoughts per step, creating a tree structure. The search process can be BFS (breadth-first search) or DFS (depth-first search) with each state evaluated by a classifier (via a prompt) or majority vote.\nTask decomposition can be done (1) by LLM with simple prompting like "Steps for XYZ.\\n1.", "What are the subgoals for achieving XYZ?", (2) by using task-specific instructions; e.g. "Write a story outline." for writing a novel, or (3) with human inputs.\nAnother quite distinct approach, LLM+P (Liu et al. 2023), involves relying on an external classical planner to do long-horizon planning. This approach utilizes the Planning Domain Definition Language (PDDL) as an intermediate interface to describe the planning problem. In this process, LLM (1) translates the problem into “Problem PDDL”, then (2) requests a classical planner to generate a PDDL plan based on an existing “Domain PDDL”, and finally (3) translates the PDDL plan back into natural language. Essentially, the planning step is outsourced to an external tool, assuming the availability of domain-specific PDDL and a suitable planner which is common in certain robotic setups but not in many other domains.\nSelf-Reflection#')

Caricare il documento nell'archivio vettoriale Milvus

Come nell'introduzione precedente, inizializziamo e carichiamo i documenti preparati nell'archivio vettoriale Milvus, che contiene due campi vettoriali: dense è per l'incorporamento OpenAI e sparse è per la funzione BM25.

vectorstore = Milvus.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    builtin_function=BM25BuiltInFunction(),
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`). See https://milvus.io/docs/tune_consistency.md#Consistency-Level for more details.
    drop_old=False,
)

Costruire la catena RAG

Prepariamo l'istanza LLM e il prompt, quindi li combiniamo in una pipeline RAG utilizzando il linguaggio LangChain Expression.

from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Initialize the OpenAI language model for response generation
llm = ChatOpenAI(model_name="gpt-4o", temperature=0)

# Define the prompt template for generating AI responses
PROMPT_TEMPLATE = """
Human: You are an AI assistant, and provides answers to questions by using fact based and statistical information when possible.
Use the following pieces of information to provide a concise answer to the question enclosed in <question> tags.
If you don't know the answer, just say that you don't know, don't try to make up an answer.
<context>
{context}
</context>

<question>
{question}
</question>

The response should be specific and use statistics or numbers when possible.

Assistant:"""

# Create a PromptTemplate instance with the defined template and input variables
prompt = PromptTemplate(
    template=PROMPT_TEMPLATE, input_variables=["context", "question"]
)
# Convert the vector store to a retriever
retriever = vectorstore.as_retriever()


# Define a function to format the retrieved documents
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

Utilizzare LCEL (LangChain Expression Language) per costruire una catena RAG.

# Define the RAG (Retrieval-Augmented Generation) chain for AI response generation
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# rag_chain.get_graph().print_ascii()

Invochiamo la catena RAG con una domanda specifica e recuperiamo la risposta.

query = "What is PAL and PoT?"
res = rag_chain.invoke(query)
res

'PAL (Program-aided Language models) and PoT (Program of Thoughts prompting) are approaches that involve using language models to generate programming language statements to solve natural language reasoning problems. This method offloads the solution step to a runtime, such as a Python interpreter, allowing for complex computation and reasoning to be handled externally. PAL and PoT rely on language models with strong coding skills to effectively perform these tasks.'

Congratulazioni! Avete costruito una catena RAG ibrida (vettore denso + funzione rada bm25) con Milvus e LangChain.