Agrega memoria y almacenamiento en caché semántico con LangChain y MongoDB

Configura el entorno.

Configura el entorno para este tutorial. Cree un cuaderno interactivo de Python guardando un archivo con la extensión .ipynb. Este cuaderno te permite ejecutar snippets de código Python de forma individual y lo utilizarás para ejecutar el código en este tutorial.

Para configurar tu entorno de notebook:

1. Ejecuta el siguiente comando en tu notebook:

   pip install --quiet --upgrade langchain langchain-community langchain-core langchain-mongodb langchain-voyageai langchain-openai langgraph langgraph-checkpoint-mongodb pypdf

2. Configure variables de entorno.

   Ejecute el siguiente código para configurar las variables de entorno de este tutorial. Proporcione su clave de API de Voyage, la clave de API de OpenAI y SRV del clúster de MongoDB cadena de conexión.

   import os
   os.environ["OPENAI_API_KEY"] = "<openai-key>"
   os.environ["VOYAGE_API_KEY"] = "<voyage-key>"
   MONGODB_URI = "<connection-string>"

   Nota

   Se debe sustituir <connection-string> por la cadena de conexión del clúster Atlas o de la implementación local de Atlas.

   Atlas Cluster

   Local o autogestionado

   Su cadena de conexión debe usar el siguiente formato:

   mongodb+srv://<db_username>:<db_password>@<clusterName>.<hostname>.mongodb.net

   Para obtener más información, consulta Conectar a un clúster a través de bibliotecas de clientes.

   Su cadena de conexión debe usar el siguiente formato:

   mongodb://localhost:<port-number>/?directConnection=true

   Para obtener más información, consulta Cadenas de conexión.

Instanciar la tienda de vectores.

Pega y ejecuta el siguiente código en su cuaderno para crear una instancia de vector store llamada vector_store utilizando el namespace langchain_db.rag_with_memory en MongoDB:

from langchain_mongodb import MongoDBAtlasVectorSearch
from langchain_voyageai import VoyageAIEmbeddings
# Use the voyage-3-large embedding model
embedding_model = VoyageAIEmbeddings(model="voyage-3-large")
# Create the vector store
vector_store = MongoDBAtlasVectorSearch.from_connection_string(
   connection_string = MONGODB_URI,
   embedding = embedding_model,
   namespace = "langchain_db.rag_with_memory"
)

Añade datos al almacén vectorial.

Pega y ejecuta el siguiente código en tu notebook para ingerir un PDF de muestra que contiene un informe de ganancias reciente de MongoDB en la base de datos vectorial.

Este código utiliza un divisor de texto para fragmentar los datos del PDF en documentos padre más pequeños. Especifica el tamaño del fragmento (número de caracteres) y la superposición de fragmentos (número de caracteres superpuestos entre fragmentos consecutivos) para cada documento.

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# Load the PDF
loader = PyPDFLoader("https://investors.mongodb.com/node/13176/pdf")
data = loader.load()
# Split PDF into documents
text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20)
docs = text_splitter.split_documents(data)
# Add data to the vector store
vector_store.add_documents(docs)

Crea el índice de búsqueda vectorial de MongoDB.

Ejecute el siguiente código para crear el índice MongoDB Vector Search para la tienda de vectores y permitir búsquedas vectoriales en sus datos:

# Use LangChain helper method to create the vector search index
vector_store.create_vector_search_index(
   dimensions = 1024 # The dimensions of the vector embeddings to be indexed
)

El índice debería tardar alrededor de un minuto en compilarse. Mientras se crea, el índice se encuentra en un estado de sincronización inicial. Cuando se termine de construir, se podrá empezar a consultar los datos de tu colección.

Configure el verificador de puntos de control de MongoDB.

Para mantener el historial de conversaciones en múltiples interacciones, utilice el punto de control MongoDBSaver de LangGraph. Un punto de control guarda el estado de una conversación en MongoDB después de cada paso y lo restaura en la siguiente interacción, lo que permite que su aplicación RAG gestione el contexto de la conversación.

LangGraph identifica cada conversación mediante un thread_id. El punto de control guarda el estado en las colecciones checkpoints y checkpoint_writes de la base de datos que especifique.

Pegue y ejecute el siguiente código en su notebook para crear un punto de control llamado checkpointer que almacene el estado de la conversación en la base de datos langchain_db.

from langgraph.checkpoint.mongodb import MongoDBSaver
from pymongo import MongoClient
# Connect to your MongoDB cluster
mongo_client = MongoClient(MONGODB_URI)
# Create the checkpointer to persist conversation state
checkpointer = MongoDBSaver(mongo_client, db_name="langchain_db")

Crea una cadena RAG que gestione el historial de mensajes de chat.

Pega y ejecuta los siguientes fragmentos de código para crear la cadena RAG:

1. Especifique el LLM a utilizar.

   from langchain_openai import ChatOpenAI
   # Define the model to use for chat completion
   llm = ChatOpenAI(model = "gpt-4o")

2. Define un prompt que resuma el historial de chat para el recuperador.

   from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
   from langchain_core.output_parsers import StrOutputParser
   # Create a prompt to generate standalone questions from follow-up questions
   standalone_system_prompt = """
   Given a chat history and a follow-up question, rephrase the follow-up question to be a standalone question.
   Do NOT answer the question, just reformulate it if needed, otherwise return it as is.
   Only return the final standalone question.
   """
   standalone_question_prompt = ChatPromptTemplate.from_messages(
       [
           ("system", standalone_system_prompt),
           MessagesPlaceholder(variable_name="history"),
           ("human", "{question}"),
       ]
   )
   # Parse output as a string
   parse_output = StrOutputParser()
   question_chain = standalone_question_prompt | llm | parse_output

3. Compila una cadena de recuperadores que procese el historial de chat y recupere documentos.

   from langchain_core.runnables import RunnablePassthrough
   # Create a retriever
   retriever = vector_store.as_retriever(search_type="similarity", search_kwargs={ "k": 5 })
   # Create a retriever chain that processes the question with history and retrieves documents
   retriever_chain = RunnablePassthrough.assign(context=question_chain | retriever | (lambda docs: "\n\n".join([d.page_content for d in docs])))

4. Define una solicitud para generar una respuesta basada en el historial de chat y el contexto recuperado.

   # Create a prompt template that includes the retrieved context and chat history
   rag_system_prompt = """Answer the question based only on the following context:
   {context}
   """
   rag_prompt = ChatPromptTemplate.from_messages(
       [
           ("system", rag_system_prompt),
           MessagesPlaceholder(variable_name="history"),
           ("human", "{question}"),
       ]
   )

5. Implementar RAG con memoria.

   Combina los componentes que has definido en una cadena RAG completa:

   # Build the RAG chain
   rag_chain = (
       retriever_chain
       | rag_prompt
       | llm
       | parse_output
   )

   Luego, envuelva la cadena en un grafo de LangGraph y compílela con el checkpointer. El grafo contiene un solo nodo que lee el estado de la conversación, pasa la última pregunta y los mensajes anteriores a la cadena, y agrega la respuesta al estado:

   from langgraph.graph import StateGraph, START, END, MessagesState
   from langchain_core.messages import AIMessage
   # Define the node that runs the RAG chain
   def call_rag_chain(state: MessagesState):
       question = state["messages"][-1].content
       history = state["messages"][:-1]
       answer = rag_chain.invoke(
           {"question": question, "history": history}
       )
       return {"messages": [AIMessage(content=answer)]}
   # Build the graph
   workflow = StateGraph(MessagesState)
   workflow.add_node("rag", call_rag_chain)
   workflow.add_edge(START, "rag")
   workflow.add_edge("rag", END)
   # Compile the graph with the checkpointer
   rag_with_memory = workflow.compile(checkpointer=checkpointer)

Prueba tu implementación de RAG.

Invoque la cadena para responder preguntas. Esta cadena mantiene el contexto de la conversación y devuelve respuestas relevantes que tienen en cuenta las interacciones anteriores. Tus respuestas pueden variar.

# First question
response_1 = rag_with_memory.invoke(
    {"messages": [("human", "What was MongoDB's latest acquisition?")]},
    {"configurable": {"thread_id": "user_1"}}
)
print(response_1["messages"][-1].content)

MongoDB's latest acquisition was Voyage AI, a pioneer in state-of-the-art embedding and reranking models for next-generation AI applications.

# Follow-up question that references the previous question
response_2 = rag_with_memory.invoke(
    {"messages": [("human", "Why did they do it?")]},
    {"configurable": {"thread_id": "user_1"}}
)
print(response_2["messages"][-1].content)

MongoDB acquired Voyage AI to enable organizations to easily build trustworthy AI applications by integrating advanced embedding and reranking models into their technology. This acquisition aligns with MongoDB's goal of helping businesses innovate at "AI speed" using its flexible document model and seamless scalability.

Configura la caché semántica.

Ejecute el siguiente código para configurar la caché semántica usando la clase MongoDBAtlasSemanticCache:

from langchain_mongodb.cache import MongoDBAtlasSemanticCache
from langchain_core.globals import set_llm_cache
# Configure the semantic cache
set_llm_cache(MongoDBAtlasSemanticCache(
    connection_string = MONGODB_URI,
    database_name = "langchain_db",
    collection_name = "semantic_cache",
    embedding = embedding_model,
    index_name = "vector_index",
    similarity_threshold = 0.5  # Adjust based on your requirements
))

Prueba el caché semántico con tu cadena RAG.

La caché semántica almacena automáticamente tus avisos. Ejecute las siguientes consultas de muestra, donde debería ver una reducción significativa en el tiempo de respuesta para la segunda consulta. Tus respuestas y tiempos de respuesta podrían variar.

%%time
# First query (not cached)
rag_with_memory.invoke(
  {"messages": [("human", "What was MongoDB's latest acquisition?")]},
  {"configurable": {"thread_id": "user_2"}}
)

CPU times: user 54.7 ms, sys: 34.2 ms, total: 88.9 ms
Wall time: 7.42 s
"MongoDB's latest acquisition was Voyage AI, a pioneer in state-of-the-art embedding and reranking models that power next-generation AI applications."

%%time
# Second query (cached)
rag_with_memory.invoke(
  {"messages": [("human", "What company did MongoDB acquire recently?")]},
  {"configurable": {"thread_id": "user_2"}}
)

CPU times: user 79.7 ms, sys: 24 ms, total: 104 ms
Wall time: 3.87 s
'MongoDB recently acquired Voyage AI.'