从 MapReduceDocumentsChain 迁移

MapReduceDocumentsChain 对（可能很长的）文本实施 map-reduce 策略。策略如下：

将文本拆分为较小的文档;
将流程映射到较小的文档;
将流程结果缩小或合并为最终结果。

请注意，映射步骤通常在输入文档上并行化。

在此上下文中应用的常见过程是摘要，其中 map 步骤汇总各个文档，reduce 步骤生成摘要。

在 reduce 步骤中，MapReduceDocumentsChain支持摘要的递归“折叠”：将根据令牌限制对输入进行分区，并生成分区的摘要。将重复此步骤，直到摘要的总长度在所需的限制内，从而允许对任意长度的文本进行摘要。这对于上下文窗口较小的模型特别有用。

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 参考：文档

遗产

详

下面我们展示了一个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 实现，它使用与上面相同的 prompt 模板。该图包括一个用于生成摘要的节点，该节点映射到输入文档列表中。然后，此节点流向生成最终摘要的第二个节点。

详

我们需要安装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 模板 | 发送 | 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支持摘要的递归“折叠”：根据令牌限制对输入进行分区，并生成分区的摘要。重复此步骤，直到摘要的总长度在所需的限制内，从而允许对任意长度的文本进行摘要。

这个 “collapse” 步骤作为while循环内MapReduceDocumentsChain.我们可以在较长的文本中演示此步骤，这是 Lilian Weng 的 LLM Powered Autonomous Agents 博客文章（如 RAG 教程和其他文档中所述）。

首先，我们加载帖子并将其分块到更小的 “sub documents” 中：

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 个代币。这导致总共 17 次 LLM 调用：

14 次调用用于汇总我们的文本拆分器生成的 14 个子文档。
这生成了总计约 1,000 - 2,000 个代币的摘要。因为我们在token_max在 1,000 个摘要中，还有两个调用来汇总（或“折叠”）这些摘要。
最后一个调用是生成两个 “折叠” 摘要的最终摘要。

LangGraph

详

我们可以在 LangGraph 中扩展我们原来的 map-reduce 实现来实现相同的递归折叠步骤。我们进行了以下更改：

添加collapsed_summaries键来存储折叠的摘要;
更新最终摘要节点以汇总折叠的摘要;
添加collapse_summaries节点，该节点根据令牌长度（此处为 1,000 个令牌，如前所述）对文档列表进行分区，并生成每个分区的摘要并将结果存储在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 的摘要策略，请参阅本教程。