LangChain と MongoDB によるメモリとセマンティックキャッシュの追加

環境を設定します。

このチュートリアルの環境を設定します。 .ipynb 拡張機能のファイルを保存して、インタラクティブPythonノートを作成します。 このノートはPythonコード スニペットを個別に実行でき、このチュートリアルのコードを実行するために使用します。

ノートク環境を設定するには、次の手順に従います。

1. ノートブックで次のコマンドを実行します。

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

2. 環境変数を設定してください。

   このチュートリアルの環境変数を設定するには、次のコードを実行します。Vorage APIキー、OpenAI APIキー、およびMongoDBクラスターの SRV 接続文字列を指定します。

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

   注意

   <connection-string> を Atlas クラスターまたはローカル Atlas 配置の接続文字列に置き換えます。

   Atlas Cluster

   ローカルまたは自己管理型

   接続stringには、次の形式を使用する必要があります。

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

   詳細については、クライアント ライブラリを使用したクラスターへの接続 を参照してください。

   接続stringには、次の形式を使用する必要があります。

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

   「接続文字列」を参照してください。

ベクトル ストアをインスタンス化します。

次のコードをノート PC に貼り付けて実行し、 MongoDBの langchain_db.rag_with_memory名前空間を使用して vector_store という名前のベクトルストアインスタンスを作成します。

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"
)

ベクトルストアにデータを追加します。

次のコードをノートに貼り付けて実行し、最近のMongoDB収益レポートを含むサンプルPDF をベクトルストアに取り込みます。

このコードは、テキストスプリッターを使用して、PDFデータを小さな親ドキュメントに分割します。各ドキュメントのチャンクサイズ（文字数）とチャンクオーバーラップ（連続するチャンク間で重なる文字数）を指定します。

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)

MongoDB ベクトル検索インデックスを作成します。

次のコードを実行してベクトルストアのMongoDB ベクトル検索インデックスを作成し、データに対してベクトル検索を有効にします。

# 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
)

The index should take about one minute to build. While it builds, the index is in an initial sync state. When it finishes building, you can start querying the data in your collection.

MongoDB チェックポインタを設定します。

To maintain conversation history across multiple interactions, use the LangGraph MongoDBSaver checkpointer. A checkpointer saves the state of a conversation to MongoDB after each step and restores it on the next interaction, which lets your RAG application handle conversation context.

LangGraphはthread_idによって各会話を識別します。チェックポインターは、指定したデータベースのcheckpointsとcheckpoint_writesコレクションに状態を書き込みます。

ノートブックに次のコードを貼り付けて実行するし、langchain_db データベースに会話状態を保存する checkpointer という名前のチェックポインターを作成します。

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")

チャット メッセージ履歴を処理する RG チェーンを作成します。

次のコード スニペットを貼り付けて実行し、RAM チェーンを作成します。

1. 使用する LLM を指定します。

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

2. 検索者のチャット履歴を要約するプロンプトを定義します。

   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. チャット履歴を処理し、ドキュメントを取得する取得チェーンをビルドします。

   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. チャット履歴と検索したコンテキストに基づいて答えるためのプロンプトを定義します。

   # 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. メモリを使用して RAG を実装します。

   定義したコンポーネントを完全な RAG チェーンに結合します。

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

   次に、チェーンを LangGraph グラフでラップし、チェックポインターでコンパイルします。グラフには、会話の状態を読み取り、最新の質問と以前のメッセージをチェーンに渡し、状態に回答を追加する単一のノードが含まれています。

   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)

RAG実装をテストしてください。

質問に答えるためにチェーンを呼び出します。この連鎖は対話のコンテキストを維持し、以前のインタラクションを考慮した関連する回答を返します。応答は異なる場合があります。

# 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.

セマンティックキャッシュを構成します。

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
))

セマンティックキャッシュをRAGチェーンでテストしてください。

セマンティックキャッシュはプロンプトを自動的にキャッシュします。次のサンプル クエリを実行すると、2 番目のクエリの応答時間が大幅に短縮されることがわかります。回答と応答時間は異なる場合があります。

%%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.'