從 MapReduceDocumentsChain 遷移

MapReduceDocumentsChain 實作了針對（可能很長的）文字的 map-reduce 策略。該策略如下：

將文字分割成較小的文件；
將一個程序映射到較小的文件上；
將程序的結果縮減或整合為最終結果。

請注意，map 步驟通常在輸入文件上平行化處理。

在此上下文中應用的常見程序是摘要，其中 map 步驟摘要個別文件，而 reduce 步驟生成摘要的摘要。

在 reduce 步驟中，MapReduceDocumentsChain 支援摘要的遞迴「折疊」：輸入將根據 Token 限制進行分割，並將生成分割區的摘要。此步驟將重複執行，直到摘要的總長度在所需的限制範圍內，從而允許摘要任意長度的文字。這對於上下文視窗較小的模型特別有用。

LangGraph 支援 map-reduce 工作流程，並為此問題帶來許多優勢：

LangGraph 允許串流個別步驟（例如連續摘要），從而可以更好地控制執行；
LangGraph 的檢查點支援錯誤恢復、擴展人工迴路工作流程，以及更輕鬆地整合到對話式應用程式中。
LangGraph 實作更容易擴展，我們將在下面看到。

下面我們將詳細介紹 MapReduceDocumentsChain 和對應的 LangGraph 實作，首先是一個簡單的範例用於說明目的，其次是一個較長的範例文字，以示範遞迴 reduce 步驟。

讓我們首先載入一個聊天模型

選擇聊天模型

pip install -qU "langchain[openai]"

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain.chat_models import init_chat_model

llm = init_chat_model("gpt-4o-mini", model_provider="openai")

基本範例（短文件）

讓我們使用以下 3 個文件進行說明。

from langchain_core.documents import Document

documents = [
    Document(page_content="Apples are red", metadata={"title": "apple_book"}),
    Document(page_content="Blueberries are blue", metadata={"title": "blueberry_book"}),
    Document(page_content="Bananas are yelow", metadata={"title": "banana_book"}),
]

API 參考：Document

傳統版

詳細資訊

下面我們展示了使用 MapReduceDocumentsChain 的實作。我們為 map 和 reduce 步驟定義提示範本，實例化這些步驟的單獨鏈，最後實例化 MapReduceDocumentsChain

from langchain.chains import MapReduceDocumentsChain, ReduceDocumentsChain
from langchain.chains.combine_documents.stuff import StuffDocumentsChain
from langchain.chains.llm import LLMChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import CharacterTextSplitter

# Map
map_template = "Write a concise summary of the following: {docs}."
map_prompt = ChatPromptTemplate([("human", map_template)])
map_chain = LLMChain(llm=llm, prompt=map_prompt)


# Reduce
reduce_template = """
The following is a set of summaries:
{docs}
Take these and distill it into a final, consolidated summary
of the main themes.
"""
reduce_prompt = ChatPromptTemplate([("human", reduce_template)])
reduce_chain = LLMChain(llm=llm, prompt=reduce_prompt)


# Takes a list of documents, combines them into a single string, and passes this to an LLMChain
combine_documents_chain = StuffDocumentsChain(
    llm_chain=reduce_chain, document_variable_name="docs"
)

# Combines and iteratively reduces the mapped documents
reduce_documents_chain = ReduceDocumentsChain(
    # This is final chain that is called.
    combine_documents_chain=combine_documents_chain,
    # If documents exceed context for `StuffDocumentsChain`
    collapse_documents_chain=combine_documents_chain,
    # The maximum number of tokens to group documents into.
    token_max=1000,
)

# Combining documents by mapping a chain over them, then combining results
map_reduce_chain = MapReduceDocumentsChain(
    # Map chain
    llm_chain=map_chain,
    # Reduce chain
    reduce_documents_chain=reduce_documents_chain,
    # The variable name in the llm_chain to put the documents in
    document_variable_name="docs",
    # Return the results of the map steps in the output
    return_intermediate_steps=False,
)

result = map_reduce_chain.invoke(documents)

print(result["output_text"])

Fruits come in a variety of colors, with apples being red, blueberries being blue, and bananas being yellow.

在 LangSmith 追蹤中，我們觀察到四個 LLM 調用：一個摘要三個輸入文件中的每一個，另一個摘要摘要。

LangGraph

下面我們展示了 LangGraph 實作，使用與上述相同的提示範本。該圖形包含一個用於生成摘要的節點，該節點映射到輸入文件列表。然後，此節點流向第二個節點，該節點生成最終摘要。

詳細資訊

我們需要安裝 langgraph

%pip install -qU langgraph

import operator
from typing import Annotated, List, TypedDict

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langgraph.constants import Send
from langgraph.graph import END, START, StateGraph

map_template = "Write a concise summary of the following: {context}."

reduce_template = """
The following is a set of summaries:
{docs}
Take these and distill it into a final, consolidated summary
of the main themes.
"""

map_prompt = ChatPromptTemplate([("human", map_template)])
reduce_prompt = ChatPromptTemplate([("human", reduce_template)])

map_chain = map_prompt | llm | StrOutputParser()
reduce_chain = reduce_prompt | llm | StrOutputParser()

# Graph components: define the components that will make up the graph


# This will be the overall state of the main graph.
# It will contain the input document contents, corresponding
# summaries, and a final summary.
class OverallState(TypedDict):
    # Notice here we use the operator.add
    # This is because we want combine all the summaries we generate
    # from individual nodes back into one list - this is essentially
    # the "reduce" part
    contents: List[str]
    summaries: Annotated[list, operator.add]
    final_summary: str


# This will be the state of the node that we will "map" all
# documents to in order to generate summaries
class SummaryState(TypedDict):
    content: str


# Here we generate a summary, given a document
async def generate_summary(state: SummaryState):
    response = await map_chain.ainvoke(state["content"])
    return {"summaries": [response]}


# Here we define the logic to map out over the documents
# We will use this an edge in the graph
def map_summaries(state: OverallState):
    # We will return a list of `Send` objects
    # Each `Send` object consists of the name of a node in the graph
    # as well as the state to send to that node
    return [
        Send("generate_summary", {"content": content}) for content in state["contents"]
    ]


# Here we will generate the final summary
async def generate_final_summary(state: OverallState):
    response = await reduce_chain.ainvoke(state["summaries"])
    return {"final_summary": response}


# Construct the graph: here we put everything together to construct our graph
graph = StateGraph(OverallState)
graph.add_node("generate_summary", generate_summary)
graph.add_node("generate_final_summary", generate_final_summary)
graph.add_conditional_edges(START, map_summaries, ["generate_summary"])
graph.add_edge("generate_summary", "generate_final_summary")
graph.add_edge("generate_final_summary", END)
app = graph.compile()

API 參考：StrOutputParser | ChatPromptTemplate | Send | StateGraph

from IPython.display import Image

Image(app.get_graph().draw_mermaid_png())

請注意，以串流模式調用圖形允許我們監控步驟，並可能在執行期間對它們採取行動。

# Call the graph:
async for step in app.astream({"contents": [doc.page_content for doc in documents]}):
    print(step)

{'generate_summary': {'summaries': ['Apples are typically red in color.']}}
{'generate_summary': {'summaries': ['Bananas are yellow in color.']}}
{'generate_summary': {'summaries': ['Blueberries are a type of fruit that are blue in color.']}}
{'generate_final_summary': {'final_summary': 'The main themes are the colors of different fruits: apples are red, blueberries are blue, and bananas are yellow.'}}

在 LangSmith 追蹤中，我們恢復了與之前相同的四個 LLM 調用。

摘要長文件

當文字相對於 LLM 的上下文視窗較長時，Map-reduce 流程特別有用。MapReduceDocumentsChain 支援摘要的遞迴「折疊」：輸入根據 Token 限制進行分割，並生成分割區的摘要。此步驟會重複執行，直到摘要的總長度在所需的限制範圍內，從而允許摘要任意長度的文字。

此「折疊」步驟實作為 MapReduceDocumentsChain 中的 while 迴圈。我們可以在較長的文字上示範此步驟，即 Lilian Weng 的一篇 LLM Powered Autonomous Agents 部落格文章（如 RAG 教學和其他文件中所述）。

首先，我們載入文章並將其分塊為較小的「子文件」

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import CharacterTextSplitter

loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
documents = loader.load()

text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=1000, chunk_overlap=0
)
split_docs = text_splitter.split_documents(documents)
print(f"Generated {len(split_docs)} documents.")

API 參考：WebBaseLoader | CharacterTextSplitter

USER_AGENT environment variable not set, consider setting it to identify your requests.
Created a chunk of size 1003, which is longer than the specified 1000
``````output
Generated 14 documents.

傳統版

詳細資訊

我們可以像以前一樣調用 MapReduceDocumentsChain

result = map_reduce_chain.invoke(split_docs)

print(result["output_text"])

The article discusses the use of Large Language Models (LLMs) to power autonomous agents in various tasks, showcasing their capabilities in problem-solving beyond generating written content. Key components such as planning, memory optimization, and tool use are explored, with proof-of-concept demos like AutoGPT and GPT-Engineer demonstrating the potential of LLM-powered agents. Challenges include limitations in historical information retention and natural language interface reliability, while the potential of LLMs in enhancing reasoning, problem-solving, and planning proficiency for autonomous agents is highlighted. Overall, the article emphasizes the versatility and power of LLMs in creating intelligent agents for tasks like scientific discovery and experiment design.

請參考上述調用的 LangSmith 追蹤。在實例化我們的 ReduceDocumentsChain 時，我們將 token_max 設定為 1,000 個 Token。這總共產生了 17 個 LLM 調用

14 個調用用於摘要我們的文字分割器生成的 14 個子文件。
這產生了總計約 1,000 - 2,000 個 Token 的摘要。由於我們設定了 1,000 的 token_max，因此還有兩個調用用於摘要（或「折疊」）這些摘要。
最後一個調用用於生成兩個「折疊」摘要的最終摘要。

LangGraph

詳細資訊

我們可以擴展 LangGraph 中的原始 map-reduce 實作，以實作相同的遞迴折疊步驟。我們進行以下變更：

將 collapsed_summaries 鍵新增到狀態以儲存折疊的摘要；
更新最終摘要節點以摘要折疊的摘要；
新增 collapse_summaries 節點，該節點根據 Token 長度（此處為 1,000 個 Token，與之前相同）分割文件列表，並生成每個分割區的摘要，並將結果儲存在 collapsed_summaries 中。

我們從 collapse_summaries 新增一個條件邊緣到自身以形成一個迴圈：如果折疊的摘要總計超過 token_max，我們將重新執行該節點。

from typing import Literal

from langchain.chains.combine_documents.reduce import (
    acollapse_docs,
    split_list_of_docs,
)


def length_function(documents: List[Document]) -> int:
    """Get number of tokens for input contents."""
    return sum(llm.get_num_tokens(doc.page_content) for doc in documents)


token_max = 1000


class OverallState(TypedDict):
    contents: List[str]
    summaries: Annotated[list, operator.add]
    collapsed_summaries: List[Document]  # add key for collapsed summaries
    final_summary: str


# Add node to store summaries for collapsing
def collect_summaries(state: OverallState):
    return {
        "collapsed_summaries": [Document(summary) for summary in state["summaries"]]
    }


# Modify final summary to read off collapsed summaries
async def generate_final_summary(state: OverallState):
    response = await reduce_chain.ainvoke(state["collapsed_summaries"])
    return {"final_summary": response}


graph = StateGraph(OverallState)
graph.add_node("generate_summary", generate_summary)  # same as before
graph.add_node("collect_summaries", collect_summaries)
graph.add_node("generate_final_summary", generate_final_summary)


# Add node to collapse summaries
async def collapse_summaries(state: OverallState):
    doc_lists = split_list_of_docs(
        state["collapsed_summaries"], length_function, token_max
    )
    results = []
    for doc_list in doc_lists:
        results.append(await acollapse_docs(doc_list, reduce_chain.ainvoke))

    return {"collapsed_summaries": results}


graph.add_node("collapse_summaries", collapse_summaries)


def should_collapse(
    state: OverallState,
) -> Literal["collapse_summaries", "generate_final_summary"]:
    num_tokens = length_function(state["collapsed_summaries"])
    if num_tokens > token_max:
        return "collapse_summaries"
    else:
        return "generate_final_summary"


graph.add_conditional_edges(START, map_summaries, ["generate_summary"])
graph.add_edge("generate_summary", "collect_summaries")
graph.add_conditional_edges("collect_summaries", should_collapse)
graph.add_conditional_edges("collapse_summaries", should_collapse)
graph.add_edge("generate_final_summary", END)
app = graph.compile()

API 參考：acollapse_docs | split_list_of_docs

LangGraph 允許繪製圖形結構以幫助視覺化其功能

from IPython.display import Image

Image(app.get_graph().draw_mermaid_png())

與之前一樣，我們可以串流圖形以觀察其步驟順序。下面，我們將只列印步驟的名稱。

請注意，由於圖形中有迴圈，因此在其執行中指定 recursion_limit 可能很有用。這類似於 ReduceDocumentsChain.token_max，當超出指定的限制時，將引發特定錯誤。

async for step in app.astream(
    {"contents": [doc.page_content for doc in split_docs]},
    {"recursion_limit": 10},
):
    print(list(step.keys()))

['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['generate_summary']
['collect_summaries']
['collapse_summaries']
['generate_final_summary']

print(step)

{'generate_final_summary': {'final_summary': 'The summaries discuss the use of Large Language Models (LLMs) to power autonomous agents in various tasks such as problem-solving, planning, and tool use. Key components like planning, memory, and task decomposition are highlighted, along with challenges such as inefficient planning and hallucination. Techniques like Algorithm Distillation and Maximum Inner Product Search are explored for optimization, while frameworks like ReAct and Reflexion show improvements in knowledge-intensive tasks. The importance of accurate interpretation of user input and well-structured code for functional autonomy is emphasized, along with the potential of LLMs in prompting, reasoning, and emergent social behavior in simulation environments. Challenges in real-world scenarios and the use of LLMs with expert-designed tools for tasks like organic synthesis and drug discovery are also discussed.'}}

在相應的 LangSmith 追蹤中，我們可以看到與之前相同的 17 個 LLM 調用，這次它們按各自的節點分組。

後續步驟

查看 LangGraph 文件，以了解有關使用 LangGraph 建立的詳細資訊，包括關於 LangGraph 中 map-reduce 詳細資訊的本指南。

請參閱本教學，了解更多基於 LLM 的摘要策略。

基本範例（短文件）​

傳統版​

LangGraph​

摘要長文件​

傳統版​

LangGraph​

後續步驟​

此頁面是否有幫助？

基本範例（短文件）

傳統版

LangGraph

摘要長文件

傳統版

LangGraph

後續步驟