How to use runnable map in current version of langchain

[deepak-kumar-98]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

def map_reduce_chain_function():
    # Map Template with input variable
    map_template = """Write a concise summary of the following content:

    {content}

    Summary:
    """
    map_prompt = PromptTemplate(
        input_variables=["content"], 
        template=map_template
    )
    
    # LLM - ChatOpenAI
    llm = ChatOpenAI()

    # Map Chain (Runnable)
    map_chain = map_prompt | llm | StrOutputParser()

    # Reduce Template with input variable
    reduce_template = """The following is a set of summaries:

    {doc_summaries}

    Summarize the above summaries with all the key details.
    Summary:
    """
    reduce_prompt = PromptTemplate(
        input_variables=["doc_summaries"], 
        template=reduce_template
    )

    # Reduce Chain (Runnable)
    reduce_chain = reduce_prompt | llm | StrOutputParser()

    # Combine Map and Reduce Chains
    map_reduce_chain = RunnableMap(
        {
            "map": map_chain,
            "reduce": reduce_chain
        }
    )

    return map_reduce_chain

Description

Is the above implementation is correct for latest version of langchain, also mention all the latest import like this below
from langchain.runnables import RunnableMap, RunnableLambda
also the code is for summarizer, is it correct?

System Info

langchain==0.3.0
langchain-community==0.3.0
langchain-core==0.3.5
langchain-openai==0.2.0
langchain-text-splitters==0.3.0
langchainhub==0.1.21
ubuntu

[dosubot[bot]]

Hey @aa31089907! 👋 I'm here to help you with any bugs, questions, or contributions you have. I'm a bot designed to assist you while waiting for a human maintainer. Let's tackle this together!

To use RunnableMap in the current version of LangChain, you should use RunnableParallel as RunnableMap is an alias for RunnableParallel. The provided summarizer implementation is correct with the latest imports if it uses RunnableParallel in place of RunnableMap.

Here is the updated implementation with the correct imports:

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableParallel

def map_reduce_chain_function():
    # Map Template with input variable
    map_template = """Write a concise summary of the following content:

    {content}

    Summary:
    """
    map_prompt = PromptTemplate(
        input_variables=["content"], 
        template=map_template
    )
    
    # LLM - ChatOpenAI
    llm = ChatOpenAI()

    # Map Chain (Runnable)
    map_chain = map_prompt | llm | StrOutputParser()

    # Reduce Template with input variable
    reduce_template = """The following is a set of summaries:

    {doc_summaries}

    Summarize the above summaries with all the key details.
    Summary:
    """
    reduce_prompt = PromptTemplate(
        input_variables=["doc_summaries"], 
        template=reduce_template
    )

    # Reduce Chain (Runnable)
    reduce_chain = reduce_prompt | llm | StrOutputParser()

    # Combine Map and Reduce Chains
    map_reduce_chain = RunnableParallel(
        {
            "map": map_chain,
            "reduce": reduce_chain
        }
    )

    return map_reduce_chain

This code snippet demonstrates how to use RunnableParallel (formerly RunnableMap) in the current version of LangChain. The imports and usage are up-to-date with the latest version of the library ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

To fetch the chunks from vector_db for summarization instead of directly passing the text, you can use the MultiVectorRetriever to retrieve the relevant chunks and then pass them to the summarization chain. Here is a step-by-step guide:

1. Build the Retriever: Use the MultiVectorRetriever to fetch the relevant chunks from the vector database.
2. Summarize the Chunks: Apply the summarization chain to the retrieved chunks.

Here is the code to achieve this:

from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
import uuid

# Define the summarization chain
prompt_text = """You are an assistant tasked with summarizing tables and text. 
Give a concise summary of the table or text. Table or text chunk: {element} """
prompt = ChatPromptTemplate.from_template(prompt_text)
model = ChatOpenAI(temperature=0, model="gpt-4")
summarize_chain = {"element": lambda x: x} | prompt | model | StrOutputParser()

# Function to build the retriever
def build_retriever(text_elements, tables, table_summaries):
    vectorstore = Chroma(
        collection_name="summaries", embedding_function=OpenAIEmbeddings()
    )
    store = InMemoryStore()
    id_key = "doc_id"
    retriever = MultiVectorRetriever(
        vectorstore=vectorstore,
        docstore=store,
        id_key=id_key,
    )
    texts = [i.text for i in text_elements]
    doc_ids = [str(uuid.uuid4()) for _ in texts]
    retriever.docstore.mset(list(zip(doc_ids, texts)))
    table_ids = [str(uuid.uuid4()) for _ in tables]
    summary_tables = [
        Document(page_content=s, metadata={id_key: table_ids[i]})
        for i, s in enumerate(table_summaries)
    ]
    retriever.vectorstore.add_documents(summary_tables)
    retriever.docstore.mset(list(zip(table_ids, tables)))
    return retriever

# Function to build the RAG chain
def build_chain(retriever, model):
    system_prompt = SystemMessagePromptTemplate.from_template(
        "You are a helpful assistant that answers questions based on provided context. Your provided context can include text or tables, "
        "and may also contain semantic XML markup. Pay attention the semantic XML markup to understand more about the context semantics as "
        "well as structure (e.g. lists and tabular layouts expressed with HTML-like tags)"
    )
    human_prompt = HumanMessagePromptTemplate.from_template(
        """Context:
        {context}
        Question: {question}"""
    )
    prompt = ChatPromptTemplate.from_messages([system_prompt, human_prompt])
    model = ChatOpenAI(temperature=0, model="gpt-4")
    chain = (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | model
        | StrOutputParser()
    )
    return chain

# Example usage
retriever = build_retriever(text_elements, tables, table_summaries)
chain = build_chain(retriever, model)
result = chain.invoke("Name all the people authorized to receive confidential information, and their roles")
print(result)

This code sets up a retriever to fetch chunks from the vector database and then uses a summarization chain to summarize the retrieved chunks. The MultiVectorRetriever allows for the retrieval of smaller chunks of documents, which can then be summarized to provide concise information ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[deepak-kumar-98]

@dosu

import os
from langchain_core.prompts import PromptTemplate
from langchain_openai.chat_models.azure import AzureChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableParallel
from langchain_community.document_loaders.pdf import PDFPlumberLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import AzureOpenAIEmbeddings
from langchain_postgres import PGVector
from langchain_core.documents import Document

def map_reduce_chain_function():
    # Map Template with input variable
    map_template = """Write a concise summary of the following content:

    {content}

    Summary:
    """
    map_prompt = PromptTemplate(
        input_variables=["content"], 
        template=map_template
    )
    
    # LLM - AzureChatOpenAI
    llm = AzureChatOpenAI(
        azure_deployment="gpt-35-turbo",  # or your deployment
        api_version="2023-06-01-preview",  # or your api version
        temperature=0,
        max_tokens=None,
        timeout=None,
        max_retries=2,
    )

    # Map Chain (Runnable)
    map_chain = map_prompt | llm | StrOutputParser()

    # Reduce Template with input variable
    reduce_template = """The following is a set of summaries:

    {doc_summaries}

    Summarize the above summaries with all the key details.
    Summary:
    """
    reduce_prompt = PromptTemplate(
        input_variables=["doc_summaries"], 
        template=reduce_template
    )

    # Reduce Chain (Runnable)
    reduce_chain = reduce_prompt | llm | StrOutputParser()

    # Combine Map and Reduce Chains
    map_reduce_chain = RunnableParallel(
        {
            "map": map_chain,
            "reduce": reduce_chain
        }
    )

    return map_reduce_chain

def generate_embeddings(file_path=None, chunk_size=200, chunk_overlap=10, persist_directory=None):
    if file_path:
        _, file_extension = os.path.splitext(file_path)
        file_extension = file_extension.lower()
        if file_path.lower().endswith(".pdf"):
            loader = PDFPlumberLoader(file_path)
            document = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=document)
        elif file_path.lower().endswith(".docx") or file_path.lower().endswith(".doc"):
            loader = Docx2txtLoader(file_path)
            document = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=document)
        elif file_path.lower().endswith(".xlsx"):
            loader = UnstructuredExcelLoader(file_path)
            docs = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=docs)
        else:
            pass
        
        embedding = AzureOpenAIEmbeddings(
            openai_api_version="2024-04-01-preview",
            openai_api_type='azure',
            openai_api_key=os.getenv("AZURE_OPENAI_API_KEY"), 
            azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
            azure_deployment="ada",
            model="text-embedding-ada-002",
            validate_base_url=True
        )
        
        CONNECTION_STRING = "your_connection_string_here"
        try:
            try:
                PGVector.from_texts(
                    embedding=embedding,
                    texts=texts,
                    collection_name=persist_directory,
                    connection_string=CONNECTION_STRING,
                )
            except:
                PGVector.from_documents(
                    embedding=embedding,
                    documents=texts,
                    collection_name=persist_directory,
                    connection_string=CONNECTION_STRING,
                )
        except Exception as e:
            print(e)
        file_crawl_status = True
    else:
        file_crawl_status = False

    return file_crawl_status

# Example usage
pdf_path = "path/to/your/pdf_file.pdf"
map_reduce_chain = map_reduce_chain_function()
pdf_text = generate_embeddings(file_path=pdf_path)
documents = [Document(page_content=pdf_text)]

result = map_reduce_chain.invoke(documents)
print(result["output_text"])

follow this flow only with retreival from vector_db as preprocessing each documents on each call is bit time taking

[dosubot[bot]]

To modify your code to retrieve chunks from a vector database for summarization instead of preprocessing each document on each call, you can use the MultiVectorRetriever. Here is the updated implementation:

1. Build the Retriever: Use the MultiVectorRetriever to fetch the relevant chunks from the vector database.
2. Summarize the Chunks: Apply the summarization chain to the retrieved chunks.

Here is the updated code:

import os
import uuid
from langchain_core.prompts import PromptTemplate
from langchain_openai.chat_models.azure import AzureChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableParallel
from langchain_community.document_loaders.pdf import PDFPlumberLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import AzureOpenAIEmbeddings
from langchain_postgres import PGVector
from langchain_core.documents import Document
from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_core.runnables import RunnablePassthrough

def map_reduce_chain_function():
    # Map Template with input variable
    map_template = """Write a concise summary of the following content:

    {content}

    Summary:
    """
    map_prompt = PromptTemplate(
        input_variables=["content"], 
        template=map_template
    )
    
    # LLM - AzureChatOpenAI
    llm = AzureChatOpenAI(
        azure_deployment="gpt-35-turbo",  # or your deployment
        api_version="2023-06-01-preview",  # or your api version
        temperature=0,
        max_tokens=None,
        timeout=None,
        max_retries=2,
    )

    # Map Chain (Runnable)
    map_chain = map_prompt | llm | StrOutputParser()

    # Reduce Template with input variable
    reduce_template = """The following is a set of summaries:

    {doc_summaries}

    Summarize the above summaries with all the key details.
    Summary:
    """
    reduce_prompt = PromptTemplate(
        input_variables=["doc_summaries"], 
        template=reduce_template
    )

    # Reduce Chain (Runnable)
    reduce_chain = reduce_prompt | llm | StrOutputParser()

    # Combine Map and Reduce Chains
    map_reduce_chain = RunnableParallel(
        {
            "map": map_chain,
            "reduce": reduce_chain
        }
    )

    return map_reduce_chain

def build_retriever(text_elements, tables, table_summaries):
    vectorstore = Chroma(
        collection_name="summaries", embedding_function=AzureOpenAIEmbeddings()
    )
    store = InMemoryStore()
    id_key = "doc_id"
    retriever = MultiVectorRetriever(
        vectorstore=vectorstore,
        docstore=store,
        id_key=id_key,
    )
    texts = [i.text for i in text_elements]
    doc_ids = [str(uuid.uuid4()) for _ in texts]
    retriever.docstore.mset(list(zip(doc_ids, texts)))
    table_ids = [str(uuid.uuid4()) for _ in tables]
    summary_tables = [
        Document(page_content=s, metadata={id_key: table_ids[i]})
        for i, s in enumerate(table_summaries)
    ]
    retriever.vectorstore.add_documents(summary_tables)
    retriever.docstore.mset(list(zip(table_ids, tables)))
    return retriever

def build_chain(retriever, model):
    system_prompt = SystemMessagePromptTemplate.from_template(
        "You are a helpful assistant that answers questions based on provided context. Your provided context can include text or tables, "
        "and may also contain semantic XML markup. Pay attention the semantic XML markup to understand more about the context semantics as "
        "well as structure (e.g. lists and tabular layouts expressed with HTML-like tags)"
    )
    human_prompt = HumanMessagePromptTemplate.from_template(
        """Context:
        {context}
        Question: {question}"""
    )
    prompt = ChatPromptTemplate.from_messages([system_prompt, human_prompt])
    model = AzureChatOpenAI(temperature=0, model="gpt-4")
    chain = (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | model
        | StrOutputParser()
    )
    return chain

def generate_embeddings(file_path=None, chunk_size=200, chunk_overlap=10, persist_directory=None):
    if file_path:
        _, file_extension = os.path.splitext(file_path)
        file_extension = file_extension.lower()
        if file_path.lower().endswith(".pdf"):
            loader = PDFPlumberLoader(file_path)
            document = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=document)
        elif file_path.lower().endswith(".docx") or file_path.lower().endswith(".doc"):
            loader = Docx2txtLoader(file_path)
            document = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=document)
        elif file_path.lower().endswith(".xlsx"):
            loader = UnstructuredExcelLoader(file_path)
            docs = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=docs)
        else:
            pass
        
        embedding = AzureOpenAIEmbeddings(
            openai_api_version="2024-04-01-preview",
            openai_api_type='azure',
            openai_api_key=os.getenv("AZURE_OPENAI_API_KEY"), 
            azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
            azure_deployment="ada",
            model="text-embedding-ada-002",
            validate_base_url=True
        )
        
        CONNECTION_STRING = "your_connection_string_here"
        try:
            try:
                PGVector.from_texts(
                    embedding=embedding,
                    texts=texts,
                    collection_name=persist_directory,
                    connection_string=CONNECTION_STRING,
                )
            except:
                PGVector.from_documents(
                    embedding=embedding,
                    documents=texts,
                    collection_name=persist_directory,
                    connection_string=CONNECTION_STRING,
                )
        except Exception as e:
            print(e)
        file_crawl_status = True
    else:
        file_crawl_status = False

    return file_crawl_status

# Example usage
pdf_path = "path/to/your/pdf_file.pdf"
map_reduce_chain = map_reduce_chain_function()
pdf_text = generate_embeddings(file_path=pdf_path)
documents = [Document(page_content=pdf_text)]

# Build retriever and chain
retriever = build_retriever(text_elements, tables, table_summaries)
chain = build_chain(retriever, model)

# Invoke the chain
result = chain.invoke("Name all the people authorized to receive confidential information, and their roles")
print(result)

This code sets up a retriever to fetch chunks from the vector database and then uses a summarization chain to summarize the retrieved chunks. The MultiVectorRetriever allows for the retrieval of smaller chunks of documents, which can then be summarized to provide concise information ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[deepak-kumar-98]

@dosu you are not using map_reduce_chain any where above , cant we add retreival chain in map_reduce_chain = RunnableParallel

[dosubot[bot]]

To add a retrieval chain in map_reduce_chain using RunnableParallel, you can integrate the retrieval process into the map-reduce workflow. Here is how you can do it:

1. Set up the retrieval chain: Use MultiVectorRetriever to fetch relevant chunks from the vector database.
2. Integrate the retrieval chain into the map-reduce chain: Combine the retrieval chain with the map and reduce chains using RunnableParallel.

Here is the updated code:

import os
import uuid
from langchain_core.prompts import PromptTemplate
from langchain_openai.chat_models.azure import AzureChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableParallel, RunnablePassthrough
from langchain_community.document_loaders.pdf import PDFPlumberLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import AzureOpenAIEmbeddings
from langchain_postgres import PGVector
from langchain_core.documents import Document
from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma

def map_reduce_chain_function(retriever):
    # Map Template with input variable
    map_template = """Write a concise summary of the following content:

    {content}

    Summary:
    """
    map_prompt = PromptTemplate(
        input_variables=["content"], 
        template=map_template
    )
    
    # LLM - AzureChatOpenAI
    llm = AzureChatOpenAI(
        azure_deployment="gpt-35-turbo",  # or your deployment
        api_version="2023-06-01-preview",  # or your api version
        temperature=0,
        max_tokens=None,
        timeout=None,
        max_retries=2,
    )

    # Map Chain (Runnable)
    map_chain = map_prompt | llm | StrOutputParser()

    # Reduce Template with input variable
    reduce_template = """The following is a set of summaries:

    {doc_summaries}

    Summarize the above summaries with all the key details.
    Summary:
    """
    reduce_prompt = PromptTemplate(
        input_variables=["doc_summaries"], 
        template=reduce_template
    )

    # Reduce Chain (Runnable)
    reduce_chain = reduce_prompt | llm | StrOutputParser()

    # Combine Retrieval, Map, and Reduce Chains
    map_reduce_chain = RunnableParallel(
        {
            "retrieval": retriever,
            "map": map_chain,
            "reduce": reduce_chain
        }
    )

    return map_reduce_chain

def build_retriever():
    vectorstore = Chroma(
        collection_name="summaries", embedding_function=AzureOpenAIEmbeddings()
    )
    store = InMemoryStore()
    id_key = "doc_id"
    retriever = MultiVectorRetriever(
        vectorstore=vectorstore,
        docstore=store,
        id_key=id_key,
    )
    return retriever

def generate_embeddings(file_path=None, chunk_size=200, chunk_overlap=10, persist_directory=None):
    if file_path:
        _, file_extension = os.path.splitext(file_path)
        file_extension = file_extension.lower()
        if file_path.lower().endswith(".pdf"):
            loader = PDFPlumberLoader(file_path)
            document = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=document)
        elif file_path.lower().endswith(".docx") or file_path.lower().endswith(".doc"):
            loader = Docx2txtLoader(file_path)
            document = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=document)
        elif file_path.lower().endswith(".xlsx"):
            loader = UnstructuredExcelLoader(file_path)
            docs = loader.load()
            text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
            texts = text_splitter.split_documents(documents=docs)
        else:
            pass
        
        embedding = AzureOpenAIEmbeddings(
            openai_api_version="2024-04-01-preview",
            openai_api_type='azure',
            openai_api_key=os.getenv("AZURE_OPENAI_API_KEY"), 
            azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
            azure_deployment="ada",
            model="text-embedding-ada-002",
            validate_base_url=True
        )
        
        CONNECTION_STRING = "your_connection_string_here"
        try:
            try:
                PGVector.from_texts(
                    embedding=embedding,
                    texts=texts,
                    collection_name=persist_directory,
                    connection_string=CONNECTION_STRING,
                )
            except:
                PGVector.from_documents(
                    embedding=embedding,
                    documents=texts,
                    collection_name=persist_directory,
                    connection_string=CONNECTION_STRING,
                )
        except Exception as e:
            print(e)
        file_crawl_status = True
    else:
        file_crawl_status = False

    return file_crawl_status

# Example usage
pdf_path = "path/to/your/pdf_file.pdf"
generate_embeddings(file_path=pdf_path)

# Build retriever and map-reduce chain
retriever = build_retriever()
map_reduce_chain = map_reduce_chain_function(retriever)

# Invoke the chain
result = map_reduce_chain.invoke({"input": "Name all the people authorized to receive confidential information, and their roles"})
print(result)

This code integrates the retrieval chain into the map-reduce chain using RunnableParallel. The retriever fetches relevant chunks from the vector database, which are then processed by the map and reduce chains ^[1][2].

To continue talking to Dosu, mention @dosu.