突破Token限制！LangChain三大文档处理策略实战指南

当使用大语言模型（如 GPT-3、GPT-4）时，模型的上下文窗口（Token 限制）往往成为瓶颈。LangChain 作为一个强大的语言模型编排框架，提供了三种核心策略——MapReduce、Refine和Stuff，以解决这一难题。

本文将深入探讨这三种方法的工作原理、适用场景、技术实现以及如何根据实际需求做出最佳选择。

注：因本文会介绍三种方法的详细代码，建议收藏后观看

问题背景：为什么需要文档处理策略？

大语言模型（LLM）虽然功能强大，但其上下文窗口通常有限（例如 GPT-3 的 4096 Token）。当面对以下场景时，直接输入所有文档会导致模型无法处理或效果下降：

1. 1. 长文本处理：如书籍、研究报告的总结生成；
2. 2. 多文档分析：如企业知识库的跨文档问答；
3. 3. 复杂逻辑任务：如需要多步推理的代码生成或数学解题。

LangChain 通过结构化链式处理（Chains）提供灵活的策略，将输入文档拆分、重组，最终生成符合要求的输出。下面我们将逐一解析三种主流方法。

对比分析与选型指南

选型建议：

1. 1. 优先选择 Stuff：当文档总长度 < 模型 Token 限制的 70% 且任务简单时；
2. 2. 使用 Refine：处理书籍、长报告等需要逻辑连贯性的任务；
3. 3. 启用 MapReduce：面对海量文档（如 >100 篇）或需要高吞吐的场景。

Stuff 策略：简单直接的“暴力填充”

核心原理

Stuff（直接填充）是最基础的方法，其逻辑简单粗暴：将所有文档内容拼接成一个字符串，直接塞进模型的上下文窗口中，并附加任务指令（如“请总结以下内容”）。

技术实现示例

代码为完整代码，可直接运行

# 导入必要库 import os from langchain_core.prompts import PromptTemplate from langchain_ollama import ChatOllama from langchain_community.document_loaders import WebBaseLoader from langchain_core.runnables import RunnablePassthrough from langchain.chains.llm import LLMChain from langchain.chains.combine_documents.stuff import StuffDocumentsChain import logging # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s') logger = logging.getLogger(__name__) # AI服务配置 AI_SERVER = "http://192.168.1.1:11434" AI_MODEL = "qwen2.5:7b" # 代理设置（如需使用） # PROXY_CONFIG = {'http': '127.0.0.1:7890', 'https': '127.0.0.1:7890'} # os.environ.update(PROXY_CONFIG) classDocumentSummarizer: """文档摘要生成器类""" def__init__(self, model_name=AI_MODEL, server_url=AI_SERVER): """初始化摘要生成器""" logger.info(f"初始化AI模型: {model_name}") self.llm = ChatOllama(model=model_name, base_url=server_url) defload_from_url(self, url): """从网址加载文档""" logger.info(f"正在从以下地址加载文档: {url}") web_loader = WebBaseLoader(url) return web_loader.load() defcreate_summary_chain(self): """创建摘要生成链""" # 定义摘要提示模板 summary_template = """针对下面的内容，写一个简洁的总结摘要: "{text}" 简洁的总结摘要:""" summary_prompt = PromptTemplate.from_template(summary_template) # 创建LLM链 process_chain = LLMChain(llm=self.llm, prompt=summary_prompt) # 创建文档处理链 return StuffDocumentsChain( llm_chain=process_chain, document_variable_name='text' ) defgenerate_summary(self, url): """生成文档摘要的主函数""" # 加载文档 documents = self.load_from_url(url) # 创建摘要链 summarizer = self.create_summary_chain() # 执行摘要生成 logger.info("开始生成摘要...") result = summarizer.invoke(documents) return result['output_text'] # 主程序 if __name__ == "__main__": # 创建摘要器实例 summarizer = DocumentSummarizer() # 目标文章URL target_url = 'https://datamining.blog.csdn.net/article/details/144689191' # 获取并打印摘要 summary = summarizer.generate_summary(target_url) print("\n摘要结果:\n" + "="*50) print(summary) print("="*50)

优点与局限

• • ✅ 优点：

  • • 实现简单，适合快速开发原型；
  • • 保留完整的上下文关联性（例如文档间的因果关系）。

• • ❌ 缺点：

  • • 文档总长度超过 Token 限制时会直接报错；

• • 模型可能因信息过载而忽略关键内容（“注意力稀释”现象）。

适用场景

• • 单篇短文本的摘要生成（如新闻文章）；
• • 小规模文档的关键词提取或实体识别；
• • 对上下文完整性要求高的简单任务。

Refine 策略：迭代优化的“渐进式加工”

核心原理

Refine（迭代优化）采用递进式处理，类似于人类阅读长文本时的“逐步理解”过程。其工作流程分为以下步骤：

1. 1. 初始化：将第一个文档输入模型，生成初始结果；
2. 2. 迭代优化：将前一步的结果与下一个文档拼接，再次输入模型，要求“基于新内容优化答案”；
3. 3. 循环：重复上述步骤直至处理完所有文档。

技术实现示例

""" 智能文档分析与摘要系统 基于LangChain框架和Ollama模型的中文文档自动摘要工具 """ import os from typing importList, Dict, Any # 网络环境配置 os.environ['http_proxy'] = '127.0.0.1:7890' os.environ['https_proxy'] = '127.0.0.1:7890' # LangChain组件导入 from langchain_ollama import ChatOllama from langchain_core.prompts import PromptTemplate from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import WebBaseLoader from langchain.chains.summarize import load_summarize_chain classDocumentSummarizer: """文档摘要生成器""" def__init__(self, model_endpoint: str = "http://192.168.1.1:11434"): """初始化摘要生成器""" self.model_endpoint = model_endpoint self.ai_model = self._initialize_model() self.summarization_chain = None def_initialize_model(self) -> ChatOllama: """初始化语言模型""" return ChatOllama( model='qwen2.5:7b', base_url=self.model_endpoint ) def_create_text_processor(self) -> RecursiveCharacterTextSplitter: """创建文本处理器""" return RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=150, separators=["\n\n", "\n", "。", "；", "，", "！", "？", " ", ""], keep_separator=True ) def_setup_prompts(self) -> tuple: """设置提示模板""" # 初始提示模板 initial_summary_template = """ 请为以下内容创建一个精炼的摘要: {text} 摘要:""" # 优化提示模板 refining_template = """ 我们正在创建一篇文章的综合摘要。 已有的摘要信息: {existing_answer} 需要考虑的新内容: {text} 请根据新内容完善已有摘要。如新内容无关紧要，请保持原摘要不变。 请用中文输出最终摘要。 """ return ( PromptTemplate.from_template(initial_summary_template), PromptTemplate.from_template(refining_template) ) def_build_chain(self): """构建摘要处理链""" initial_prompt, refining_prompt = self._setup_prompts() self.summarization_chain = load_summarize_chain( llm=self.ai_model, chain_type="refine", question_prompt=initial_prompt, refine_prompt=refining_prompt, return_intermediate_steps=False, input_key="input_documents", output_key="output_text" ) defprocess_url(self, target_url: str) -> str: """处理指定URL的文档并生成摘要""" # 1. 加载网页内容 print(f"正在获取网页内容: {target_url}") content_loader = WebBaseLoader(target_url) raw_documents = content_loader.load() # 2. 分割文档 print("正在处理文档内容...") text_processor = self._create_text_processor() document_segments = text_processor.split_documents(raw_documents) print(f"文档已分割为{len(document_segments)}个片段") # 3. 初始化摘要链 ifnotself.summarization_chain: self._build_chain() # 4. 生成摘要 print("正在生成摘要...") result = self.summarization_chain.invoke( {"input_documents": document_segments}, return_only_outputs=True ) return result["output_text"] if __name__ == "__main__": # 运行摘要生成器 summarizer = DocumentSummarizer() article_url = "https://datamining.blog.csdn.net/article/details/144689191" summary = summarizer.process_url(article_url) print("\n" + "="*50) print("文章摘要") print("="*50) print(summary) print("="*50 + "\n")

优点与局限

• • ✅ 优点：

  • • Token 消耗可控，适合超长文本；
  • • 保留部分历史上下文，适合需要连贯性的任务。

• • ❌ 缺点：

  • • 文档处理顺序可能影响最终结果（例如关键信息在末尾时容易被忽略）；

• • 多次调用模型导致延迟较高。

适用场景

• • 书籍或论文的章节式总结（如逐章分析后生成全书摘要）；
• • 多轮对话中的上下文继承（例如客服聊天机器人）；
• • 需要结合历史数据的时序性任务（如股票报告分析）。

四、MapReduce 策略：分治并行的“工业化流水线”

核心原理

MapReduce借鉴分布式计算思想，将任务拆分为Map（映射）和Reduce（归约）两个阶段：

1. 1. Map 阶段：将每个文档独立输入模型，生成局部结果（如单文档摘要）；
2. 2. Reduce 阶段：汇总所有局部结果，再次输入模型生成最终答案。

技术实现示例

代码为完整代码，可直接运行

""" 基于MapReduce范式的大型文档分析系统 MapReduce是一种编程模型，适用于大规模数据处理： - Map阶段：将输入数据分解为较小的子任务并并行处理 - Reduce阶段：合并所有子任务结果到最终输出 """ import os import logging import time from typing importList, Dict, Optional, Tuple from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_ollama import ChatOllama from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import WebBaseLoader from pydantic import BaseModel, Field # 日志配置 logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s') logger = logging.getLogger("mapreduce_analyzer") # 网络和模型配置 PROXY_CONFIG = {'http': '127.0.0.1:7890', 'https': '127.0.0.1:7890'} os.environ.update(PROXY_CONFIG) MODEL_URL = "https://192.168.1.1:11434" MODEL_NAME = "qwen2.5:7b" classDocumentInsight(BaseModel): """文档洞察数据模型""" key_points: List[str] = Field(description="文档段落的关键点") segment_summary: str = Field(description="段落摘要") segment_id: int = Field(description="段落ID") classMapReduceProcessor: """基于MapReduce范式的文档处理系统""" def__init__(self, llm_url: str = MODEL_URL, llm_name: str = MODEL_NAME): """初始化MapReduce处理器""" logger.info(f"MapReduce处理器初始化，模型: {llm_name}，端点: {llm_url}") self.llm = ChatOllama(model=llm_name, base_url=llm_url) # 配置文档分割器 - Map阶段的预处理步骤 self.splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=100, separators=["\n\n", "\n", "。", ".", " ", ""] ) # 初始化性能指标 self.metrics = {"load_time": 0, "map_time": 0, "reduce_time": 0} defload_document(self, url: str) -> List[Document]: """加载文档 - MapReduce的输入数据准备阶段""" start_time = time.time() logger.info(f"[数据加载] 从URL获取文档: {url}") try: loader = WebBaseLoader(url) documents = loader.load() logger.info(f"[数据加载] 完成，文档大小: {sum(len(doc.page_content) for doc in documents)} 字符") self.metrics["load_time"] = time.time() - start_time return documents except Exception as e: logger.error(f"[数据加载] 失败: {e}") raise RuntimeError(f"文档加载失败: {str(e)}") defpartition_data(self, documents: List[Document]) -> List[Document]: """数据分区 - 为Map阶段准备并行处理单元""" logger.info("[数据分区] 将文档分割为更小的数据块") chunks = self.splitter.split_documents(documents) logger.info(f"[数据分区] 完成，共 {len(chunks)} 个数据块") return chunks defmap_function(self, doc_chunks: List[Document]) -> List[str]: """Map阶段 - 对每个文档块并行应用转换函数""" start_time = time.time() logger.info(f"[Map阶段] 开始处理 {len(doc_chunks)} 个文档块") # Map阶段提示模板 map_prompt = """ # 文档块分析任务 分析以下文本块，提取核心信息: --- {chunk_content} --- 请提供这段文本的关键信息提取: """ map_template = ChatPromptTemplate.from_template(map_prompt) # 执行Map操作 map_results = [] for i, chunk inenumerate(doc_chunks): logger.info(f"[Map] 处理块 {i+1}/{len(doc_chunks)}") result = self.llm.invoke( map_template.format(chunk_content=chunk.page_content) ) map_results.append(result.content) self.metrics["map_time"] = time.time() - start_time logger.info(f"[Map阶段] 完成，生成了 {len(map_results)} 个中间结果") return map_results defreduce_function(self, map_results: List[str]) -> str: """Reduce阶段 - 合并所有Map结果到最终输出""" start_time = time.time() logger.info(f"[Reduce阶段] 开始合并 {len(map_results)} 个Map结果") # Reduce阶段提示模板 reduce_prompt = """ # 信息整合任务 以下是从文档不同部分提取的信息片段: {map_results} ## 任务 将以上所有信息整合成一个连贯、全面的文档摘要。你的摘要应该: 1. 包含所有主要信息点 2. 消除重复和冗余 3. 保持逻辑连贯性 4. 按主题组织内容 5. 保留重要细节和数据 ## 综合摘要: """ reduce_template = ChatPromptTemplate.from_template(reduce_prompt) # 格式化Map结果 formatted_results = "\n\n".join([f"[片段 {i+1}]\n{result}"for i, result inenumerate(map_results)]) # 执行Reduce操作 result = self.llm.invoke( reduce_template.format(map_results=formatted_results) ) self.metrics["reduce_time"] = time.time() - start_time logger.info("[Reduce阶段] 完成，生成了最终摘要") return result.content defexecute_mapreduce(self, url: str) -> Dict[str, any]: """执行完整的MapReduce流程""" total_start = time.time() logger.info(f"[MapReduce任务] 开始处理URL: {url}") # 1. 加载数据 - 预处理 raw_documents = self.load_document(url) # 2. 分区 - 准备Map阶段 document_chunks = self.partition_data(raw_documents) # 3. Map阶段 - 并行处理每个块 map_results = self.map_function(document_chunks) # 4. Reduce阶段 - 合并结果 final_result = self.reduce_function(map_results) # 5. 计算总时间并返回结果 total_time = time.time() - total_start self.metrics["total_time"] = total_time logger.info(f"[MapReduce任务] 完成，总耗时: {total_time:.2f}秒") return { "summary": final_result, "metrics": self.metrics, "chunks_processed": len(document_chunks) } if __name__ == "__main__": # 创建MapReduce处理器 processor = MapReduceProcessor() # 执行MapReduce任务 url = 'https://datamining.blog.csdn.net/article/details/144689191' result = processor.execute_mapreduce(url) # 输出结果和性能指标 print("\n" + "="*50) print("MapReduce 文档分析结果:") print("="*50) print(result["summary"]) print("\n" + "-"*30) print("性能指标:") print(f"- 数据加载: {result['metrics']['load_time']:.2f}秒") print(f"- Map阶段: {result['metrics']['map_time']:.2f}秒") print(f"- Reduce阶段: {result['metrics']['reduce_time']:.2f}秒") print(f"- 总处理时间: {result['metrics']['total_time']:.2f}秒") print(f"- 处理的数据块数: {result['chunks_processed']}") print("="*50)

优点与局限

• • ✅ 优点：

  • • 支持并行处理，大幅提升吞吐量；
  • • 突破单文档 Token 限制，适合海量数据。

• • ❌ 缺点：

  • • Map 阶段的独立处理可能丢失文档间关联（例如跨文档的指代关系）；

• • 需要设计合理的 Map 和 Reduce 指令以避免信息失真。

适用场景

• • 企业知识库的批量问答（如从 1000 份产品文档中提取参数）；
• • 学术论文集的趋势分析（如从多篇论文摘要中提炼领域热点）；
• • 需要分布式计算的云原生应用。

实战技巧与进阶优化

技巧 1：混合使用策略

• • 对于超长单文档，可先用MapReduce 分块处理，再用Refine 合并结果。
• • 示例：处理一篇 50 页的 PDF 研究报告时，先按章节拆分（Map），再逐章迭代优化（Refine）。

技巧 2：优化 Reduce 阶段的提示词

• • 在 MapReduce 的 Reduce 阶段，使用结构化指令明确聚合逻辑：python复制

  reduce_prompt = """请根据以下局部摘要列表，生成一个全局总结： 要求： 1. 按时间顺序排列关键事件； 2. 标注不同观点的冲突点； 3. 字数不超过 300 字。 局部摘要列表：{text}"""

技巧 3：规避 Refine 的顺序偏差

• • 在迭代前对文档按重要性排序（例如用 TF-IDF 筛选关键段落）；
• • 在每次迭代时附加元数据（如“当前已处理 3/10 个文档”），帮助模型理解进度。

总结

LangChain 的 MapReduce、Refine 和 Stuff 策略为不同规模的文本处理任务提供了灵活解决方案。理解它们的底层逻辑和适用边界，能够帮助开发者在成本、效率和质量之间找到最佳平衡。随着 LLM 技术的演进，这些策略或许会进一步融合，但在可预见的未来，分治与迭代仍将是处理超长文本的核心方法论。

想入门 AI 大模型却找不到清晰方向？备考大厂 AI 岗还在四处搜集零散资料？别再浪费时间啦！2026 年AI 大模型全套学习资料已整理完毕，从学习路线到面试真题，从工具教程到行业报告，一站式覆盖你的所有需求，现在全部免费分享！

👇👇扫码免费领取全部内容👇👇

​

一、学习必备：100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT，帮你看透 AI 趋势

想了解大模型的行业动态、商业落地案例？大模型电子书？这份资料帮你站在 “行业高度” 学 AI：

1. 100+本大模型方向电子书

2. 26 份行业研究报告：覆盖多领域实践与趋势

报告包含阿里、DeepSeek 等权威机构发布的核心内容，涵盖：

• 职业趋势：《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》；
• 商业落地：《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》；
• 领域细分：《AGI 在金融领域的应用报告》《AI GC 实践案例集》；
• 行业监测：《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。

3. 600+套技术大会 PPT：听行业大咖讲实战

PPT 整理自 2024-2025 年热门技术大会，包含百度、腾讯、字节等企业的一线实践：

• 安全方向：《端侧大模型的安全建设》《大模型驱动安全升级（腾讯代码安全实践）》；
• 产品与创新：《大模型产品如何创新与创收》《AI 时代的新范式：构建 AI 产品》；
• 多模态与 Agent：《Step-Video 开源模型（视频生成进展）》《Agentic RAG 的现在与未来》；
• 工程落地：《从原型到生产：AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。

二、求职必看：大厂 AI 岗面试 “弹药库”，300 + 真题 + 107 道面经直接抱走

想冲字节、腾讯、阿里、蔚来等大厂 AI 岗？这份面试资料帮你提前 “押题”，拒绝临场慌！

1. 107 道大厂面经：覆盖 Prompt、RAG、大模型应用工程师等热门岗位

面经整理自 2021-2025 年真实面试场景，包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题，每道题都附带思路解析：

2. 102 道 AI 大模型真题：直击大模型核心考点

针对大模型专属考题，从概念到实践全面覆盖，帮你理清底层逻辑：

3. 97 道 LLMs 真题：聚焦大型语言模型高频问题

专门拆解 LLMs 的核心痛点与解决方案，比如让很多人头疼的 “复读机问题”：

三、路线必明： AI 大模型学习路线图，1 张图理清核心内容

刚接触 AI 大模型，不知道该从哪学起？这份「AI大模型 学习路线图」直接帮你划重点，不用再盲目摸索！

路线图涵盖 5 大核心板块，从基础到进阶层层递进：一步步带你从入门到进阶，从理论到实战。

L1阶段:启航篇丨极速破界AI新时代

L1阶段：了解大模型的基础知识，以及大模型在各个行业的应用和分析，学习理解大模型的核心原理、关键技术以及大模型应用场景。

L2阶段：攻坚篇丨RAG开发实战工坊

L2阶段：AI大模型RAG应用开发工程，主要学习RAG检索增强生成：包括Naive RAG、Advanced-RAG以及RAG性能评估，还有GraphRAG在内的多个RAG热门项目的分析。

L3阶段：跃迁篇丨Agent智能体架构设计

L3阶段：大模型Agent应用架构进阶实现，主要学习LangChain、 LIamaIndex框架，也会学习到AutoGPT、 MetaGPT等多Agent系统，打造Agent智能体。

L4阶段：精进篇丨模型微调与私有化部署

L4阶段：大模型的微调和私有化部署，更加深入的探讨Transformer架构，学习大模型的微调技术，利用DeepSpeed、Lamam Factory等工具快速进行模型微调，并通过Ollama、vLLM等推理部署框架，实现模型的快速部署。

L5阶段：专题集丨特训篇 【录播课】

四、资料领取：全套内容免费抱走，学 AI 不用再找第二份

不管你是 0 基础想入门 AI 大模型，还是有基础想冲刺大厂、了解行业趋势，这份资料都能满足你！
现在只需按照提示操作，就能免费领取：

👇👇扫码免费领取全部内容👇👇

​

2026 年想抓住 AI 大模型的风口？别犹豫，这份免费资料就是你的 “起跑线”！