升级LangChain到1.0之后我的老项目第一次运行就红了一大片。最扎眼的是from langchain.document_transformers import DoctranQATransformer这一行。项目里那套把PDF转成问答对再做检索增强的流程全靠doctran在顶着换到新版本后import直接报错网上搜了一圈答案基本都是“装回0.2.x”或者“别用LangChain了”。我不甘心为了一个文档转换器把整个LangChain生态锁死在旧版所以花了一晚上把它拆开重写了一遍。这篇文章就是那次改造的完整记录包括替代方案的设计思路、可以直接抄走的代码以及跑生产环境前必须处理的几个坑。内容不挑模型厂商OpenAI、Ollama、Azure都能接适合正在从旧版迁移或者想在LangChain 1.0里自建文档问答转换能力的同学。1. doctran为什么在LangChain 1.0里沦为了“遗留组件”1.1 doctran当年做的事先说清楚doctran是什么。它并不是LangChain官方维护的库而是一个第三方文档转换工具但因为功能贴合LLM应用场景在早期LangChain文档里拿到了“一等公民”的位置。它的核心能力是把原始文档改写成更适合模型使用的形态。常用转换器包括DoctranQATransformer把一个纯文本块变成若干个“问题答案”的组合。DoctranTextCleaner压缩语气、删除冗余、规整表达。DoctranDateExtractor从文本中识别并标准化日期方便做时间过滤。DoctranPropertyExtractor抽取出文档里的关键属性比如公司名、产品名。其中DoctranQATransformer最常用。它在底层做的事情其实并不复杂拿到一段文本调用一次LLM告诉模型“请从这段文本里提取3到5个问答对”再把模型返回的结果拆成新的Document对象。这些问答对可以直接丢进向量库做RAG也可以作为模型微调的种子数据。这个思路本身是很漂亮的。因为用户提问通常是“XX的价格是多少”“XX流程的第几步是什么”而文档原文可能是大段描述。把原文预先切成问答对相当于给检索器做了一层语义对齐命中率会比直接拿长段落去embedding高不少。1.2 升级之后到底断在哪里LangChain 1.0发布后老代码报错的链条是这样的langchain.document_transformers这个路径被清理第三方转换器不再从主包导出DoctranQATransformer直接import不到。doctran底层调用的是旧版OpenAI客户端和langchain-openai现在的模型抽象不一致即使你把import路径改对内部构造参数也对不上。LangChain 1.0核心全面转向pydantic v2doctran没有同步跟进导致类型校验在运行时各种报错。本质上不是“不能兼容”而是“没人维护这种兼容”。doctran的核心逻辑和LangChain耦合很深LangChain一升级它就要跟着改而它本身又不是官方项目社区贡献者没有足够的动力持续跟进。doctran旧功能LangChain 1.0现状替代方式问答转换官方无直接替代自己实现核心代码不过几十行文本清理官方只有切分器正则自定义清理函数日期提取官方无直接替代dateutil.parser正则属性提取官方推荐结构化输出with_structured_output这里最坑的是第一个问答转换看起来简单但网上几乎没有“1.0可直接运行”的完整示例。老教程清一色是from langchain.document_transformers import DoctranQATransformer代码抄下来直接就是红的。1.3 不降级的底线我的结论很明确不要为了一个文档转换器把版本钉死在0.2.x。为什么LangChain 1.0带来的不只是版本号变化而是一整套接口重构。Runnable协议、LangGraph的Agent状态管理、with_structured_output、新的callback系统这些都是在旧版本上很难平滑使用的。你为doctran锁一次版本之后每一次想用新特性都要重新考虑兼容性成本会越来越高。而且自己实现一个QA转换器并不复杂。核心就三步切块、让LLM生成问答对、把结果装回Document。doctran替你封装的是“调用姿势”不是“算法黑科技”。只要理解了这一点替代它就是一次很普通的工程改造。2. 不降级的替代架构三个可替换部件2.1 文档加载与清洗第一步还是要把PDF、Word、HTML等原始格式变成纯文本。langchain_community里提供了PyPDFLoader用起来很简单from langchain_community.document_loaders import PyPDFLoader loader PyPDFLoader(产品白皮书.pdf) docs loader.load()它会按页返回每个Document的page_content是一页文字metadata里带source和page。这个page字段后面非常有价值它能让问答对回到原始文档的物理位置。但这里有一个很常见的问题PDF提取出来的文本经常有断行。一页文本可能是每行一句甚至每行半句直接拿去切分语义是碎的。所以我一般会在切分之前加一道清洗函数import re def normalize_paragraph(text): lines [line.strip() for line in text.splitlines()] paragraphs [] buf [] for line in lines: if not line: if buf: paragraphs.append(.join(buf)) buf [] continue # 跳过疑似页码的独立数字 if re.fullmatch(r\d, line): continue # 跳过疑似页眉页脚的短句 if len(line) 30 and line.isupper(): continue # 如果上一行末尾不是句号、冒号、逗号就拼接否则换成句号拼接 if buf and not re.search(r[。.!?]$, buf[-1]): buf.append(line) elif buf: buf.append(。 line) else: buf.append(line) if buf: paragraphs.append(.join(buf)) return \n\n.join(paragraphs)这个函数不追求完美但能把绝大多数PDF常见的断行问题解决掉。清洗后再把文本放回Documentcleaned_docs [] for doc in docs: text normalize_paragraph(doc.page_content) if text: cleaned_docs.append(Document(page_contenttext, metadatadoc.metadata))这一步不做好的话后面LLM生成的问答质量会直接打折。2.2 切分策略决定问答质量上限清洗完成之后要切分。切分是整个流程的质量天花板比Prompt设计更重要。为什么因为LLM只能看到你给它的那一段文本。如果一段文本里混杂了三个互不相关的主题模型生成的问题就会失焦回答也会带上无关信息。RecursiveCharacterTextSplitter是默认选择它按[\n\n, \n, , ]的优先级逐层找切点对普通技术文档和产品白皮书都够用from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size1200, chunk_overlap200, ) chunks splitter.split_documents(cleaned_docs) for idx, chunk in enumerate(chunks): chunk.metadata[chunk_id] idx参数怎么调chunk_size1200是我在大多数场景下的起点。太小比如300到500单个chunk信息量不足LLM生成的问题会很琐碎太大比如2000以上模型容易漏掉细节而且输出延迟明显上升。chunk_overlap200是为了防止一个完整知识点被截断成两半。但overlap不是越大越好它会让相邻chunk重复内容后面LLM调用次数和成本都会增加。如果你的内容语义结构很强比如会议纪要、技术文档可以试试SemanticChunker。它通过embedding相似度来判断语义边界切出来的块更“自然”。但代价是要额外调embedding模型处理速度也慢。我的建议是第一版先用RecursiveCharacterTextSplitter等评估出来确实有语义割裂问题再升级。2.3 QA生成层切分完成后每个chunk会独立进入LLM生成问答对。这一层要解决两件事输出格式怎么约定问题质量怎么控制。输出格式我强烈建议用LangChain 1.0的with_structured_output绑定一个Pydantic模型。它会走模型原生的tool calling / function calling能力比在Prompt里写“请返回JSON”可靠得多。这个问题后面专门讲。问题质量要从Prompt层面约束。最核心的一条规则是“答案必须能在给定文本中找到依据不得补充外部知识。”否则模型会在你的知识库里一本正经地胡编。其次要约束问题类型让模型优先提取“指标、定义、步骤、结论”这类有独立检索价值的信息而不是生成“这段话讲了什么”这种低质量问题。3. 完整实现在LangChain 1.0里复刻DoctranQA转换器3.1 依赖与准备先装依赖pip install langchain1.0.0 langchain-openai langchain-text-splitters langchain-community pypdf pydantic各包职责langchain-coreDocument对象、BaseDocumentTransformer协议。langchain-openaiChatOpenAI是大模型调用入口。langchain-text-splittersRecursiveCharacterTextSplitter。langchain-communityPyPDFLoader等文档加载器。pypdfPDF解析底层库。如果你用的是Ollama这类本地模型把langchain-openai替换成langchain-ollama就行后面的ChatOllama调用方式和ChatOpenAI几乎一样。3.2 Pydantic Schema与Prompt设计先定义输出结构from pydantic import BaseModel, Field from typing import List class QAPair(BaseModel): question: str Field(..., description能独立回答的清晰问题) answer: str Field(..., description严格来自原文内容的答案) class QAPairList(BaseModel): qa_pairs: List[QAPair] Field(..., description从当前文档块中提取的问答列表)Field里的description很重要它会被拼进模型的tool schema里等于你在用元信息告诉模型每个字段该填什么。再写Prompt模板QA_PROMPT_TEMPLATE 你是一个文档问答转换器。 下面是一段文档内容请从中提取 {max_questions} 个问答对。 要求 1. 问题要有明确的信息量比如“某某指标是多少”“流程的第几步是什么”。 2. 所有答案必须能在给定文档中找到依据不得补充文档外的知识。 3. 如果文档内容不足以凑满 {max_questions} 个问题生成多少算多少。 4. 回答保留数字、百分比、人名、版本号等关键信息。 文档内容 {content} 注意第3条。很多模型在收到“生成5个问题”的指令后即使文本信息不够也会硬凑。硬凑出来的问答不仅没有检索价值还会污染向量库。这条约束能把硬凑概率压下来。3.3 实现 QATransformer核心类如下import logging from langchain_core.documents import Document from langchain_openai import ChatOpenAI logger logging.getLogger(__name__) class QATransformer: 在LangChain 1.0中替代DoctranQATransformer的最小实现。 def __init__(self, llm, max_questions3): self.llm llm.with_structured_output(QAPairList) self.max_questions max_questions def transform_documents(self, documents): new_docs [] for doc in documents: content doc.page_content.strip() if not content: continue prompt QA_PROMPT_TEMPLATE.format( contentcontent, max_questionsself.max_questions ) try: result self.llm.invoke(prompt) except Exception: logger.warning(QA生成失败跳过 chunk%s, doc.metadata) continue for idx, qa in enumerate(result.qa_pairs): page f问题{qa.question}\n回答{qa.answer} metadata { **doc.metadata, question: qa.question, answer: qa.answer, qa_index: idx, } new_docs.append(Document(page_contentpage, metadatametadata)) return new_docs这里有一个关键设计把question和answer同时放进metadata。这样后面做评估、去重、过滤时不需要重新解析page_content直接用doc.metadata[question]就行。另外metadata里保留了原chunk的chunk_id和page出问题的时候能一路回溯到原始PDF的物理位置。有一点要说明这个类没有继承BaseDocumentTransformer因为LangChain 1.0里不同子包的导出路径略有差异。实际上LangChain的文档转换器在使用时是鸭子类型只要你有transform_documents方法就能被TextSplitter之后的流程当转换器用。如果你更想要正式一点可以试试from langchain_core.documents.transformers import BaseDocumentTransformer class QATransformer(BaseDocumentTransformer): ...如果你的版本能import到就继承不能就用上面那个裸类不影响使用。3.4 从PDF开始跑通全流程把清洗、切分、转换串起来from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader from langchain_openai import ChatOpenAI # 1. 加载PDF loader PyPDFLoader(产品白皮书.pdf) raw_docs loader.load() # 2. 清洗断行 cleaned_docs [] for doc in raw_docs: text normalize_paragraph(doc.page_content) if text: cleaned_docs.append(Document(page_contenttext, metadatadoc.metadata)) # 3. 切分 splitter RecursiveCharacterTextSplitter(chunk_size1200, chunk_overlap200) chunks splitter.split_documents(cleaned_docs) for idx, chunk in enumerate(chunks): chunk.metadata[chunk_id] idx # 4. 初始化模型和转换器 llm ChatOpenAI(modelgpt-4o-mini, temperature0) transformer QATransformer(llm, max_questions3) # 5. 转换 qa_docs transformer.transform_documents(chunks) print(f原始文档块: {len(chunks)} 个生成问答对: {len(qa_docs)} 个) for doc in qa_docs[:3]: print(---) print(doc.page_content)跑出来的效果大概是这样的原始文档块: 12 个生成问答对: 31 个 --- 问题产品A的定价策略包含哪三档 回答产品A的定价策略包含标准版、专业版和企业版三档标准版按年订阅企业版支持私有化部署。这里temperature0是必须的。问答转换要的是忠实提取不是创意发挥。温度调高之后模型会在你没注意的地方“润色”原文这对RAG场景是灾难。4. 生产环境里真正值得注意的坑4.1 分块大小与模型上下文窗口的匹配很多人在本地模型上跑这套代码发现结果很差第一反应是模型能力不行。但很多时候问题出在分块大小和模型上下文窗口不匹配。假设你用的是7B参数本地模型上下文窗口是8192 tokens。你的chunk有1200 tokens加上生成的3个问答对、tool schema、系统提示词总共可能到2500到3000 tokens看上去没问题。但是如果把chunk_size调到6000整个请求已经逼近8000模型还要分配精力去生成结构化输出结果就是前言不搭后语。一个粗略的估算公式content_tokens max_questions * 80 500 context_limit * 0.7500是给tool schema和系统提示词留的余量。如果你的chunk token数超出这个范围优先调小chunk_size而不是换更强模型。另外建议把chunk大小从500到1500每隔200扫一遍对同一篇文档各跑一次对比生成问答的“忠实度”和“信息覆盖度”。这道工序花不了多少时间但能省掉后面很多排查功夫。4.2 结构化输出失败与兜底重试with_structured_output虽然比“Prompt返回JSON”可靠但也不是100%成功。模型可能返回空字段可能JSON schema校验失败也可能因为网络超时根本没返回。我实际跑下来gpt-4o-mini在tool calling模式下的失败率很低但本地7B模型失败率能达到5%到10%。建议做一个带重试的生成方法def generate_qa_pairs(transformer, content, max_questions, retries2): prompt QA_PROMPT_TEMPLATE.format( contentcontent, max_questionsmax_questions ) for attempt in range(retries): try: return transformer.llm.invoke(prompt) except Exception: logger.warning(第 %s 次生成失败降低问题数重试, attempt 1) if attempt 0: prompt QA_PROMPT_TEMPLATE.format( contentcontent, max_questions1 ) return QAPairList(qa_pairs[])重试时的关键技巧是第二次不要沿用原来的max_questions要把问题数降到1。因为很多失败是模型输出长度不够、生成到一半被截断问题数减少后输出token数量也减少成功率会明显提高。4.3 成本、限流与缓存成本很容易被低估。一个10页的PDF按每页800字算切分成1200字符的chunk大概有6到8个chunk每个chunk生成3个问答对一次PDF要做6到8次LLM调用。如果文档库有1000个PDF就是6000到8000次调用。用gpt-4o-mini还能接受换成claude或大尺寸模型账单会非常难看。我的建议是加缓存。最简单的做法是用chunk内容做键import hashlib import json class CachedQATransformer(QATransformer): def __init__(self, llm, max_questions3, cacheNone): super().__init__(llm, max_questions) self.cache cache if cache is not None else {} def transform_documents(self, documents): new_docs [] for doc in documents: content doc.page_content.strip() if not content: continue key hashlib.sha256(content.encode()).hexdigest() if key in self.cache: qa_pairs self.cache[key].qa_pairs else: result self.generate_with_retry(content) qa_pairs result.qa_pairs self.cache[key] result ...不要小看这个缓存。文档库里的很多PDF会有重复章节不同版本的合同、白皮书里大段文字完全相同。缓存能让你第二次全量重建向量库时几乎不花LLM费用。并发方面ChatOpenAI自身有max_retries参数处理限流但真正大批量处理时建议加信号量。简单起见可以先同步跑等数据量真的上来了再改成asyncio.gatherSemaphore(5)把并发压到5左右既能提速又不会触发太多限流。4.4 不同模型提供商的适配如果你用的是Ollamafrom langchain_ollama import ChatOllama llm ChatOllama(modelqwen2.5:7b, temperature0)但这里有个前提with_structured_output依赖模型支持tool calling / function calling。Ollama很多模型是支持的但支持程度参差不齐。如果发现结构化输出频繁失败可以退回传统方案用PydanticOutputParser把format instructions写进Promptfrom langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import PromptTemplate parser PydanticOutputParser(pydantic_objectQAPairList) prompt PromptTemplate( templateQA_PROMPT_TEMPLATE \n{format_instructions}, input_variables[content, max_questions], partial_variables{format_instructions: parser.get_format_instructions()}, ) chain prompt | llm | parser result chain.invoke({content: content, max_questions: 3})这种方式兼容性最好代价是让模型自己生成JSON解析失败率和平均延迟都会略高一点。对有tool calling但不稳定的本地模型这是性价比更高的选择。5. 接入RAG与验证转换器到底有没有用5.1 把问答对写入向量库转换器跑完之后qa_docs就是一批“问题回答”样式的Document。接下来直接建向量库from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS embedding OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore FAISS.from_documents(qa_docs, embedding) retriever vectorstore.as_retriever(search_kwargs{k: 4})为什么不直接索引原始chunks而是索引问答对因为语义对齐。用户的问题通常是问句问句和问句的嵌入向量距离更近而问句和长段落的嵌入距离可能很远。检索器在找“产品A的定价策略包含哪三档”时更容易命中“问题产品A的定价策略包含哪三档”这条QA文档。但也要注意它的代价QA转换是信息有损的。模型只会生成3到5个问答对如果原文某个关键信息没被转换成问题用户怎么搜都搜不到。所以我推荐混合索引原始chunks和QA文档都写入向量库用metadata[doc_type]区分检索时同时召回两者。5.2 简单问答链路转换后的结果可以直接接一个最小的RAG链路from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template( 基于以下检索到的问答片段回答用户问题。 片段 {context} 用户问题{question} 如果片段中没有相关信息请直接说“未知”不要编造。 ) def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) def ask(question): docs retriever.invoke(question) context format_docs(docs) response llm.invoke(prompt.format(contextcontext, questionquestion)) return response.content这个链路看起来简单但足够验证转换器的效果。如果ask的回答经常是“未知”说明QA转换后信息覆盖不够需要调大max_questions或者调整切分策略。5.3 验证思路验证不能靠感觉。我建议按以下三步走格式成功率。统计全部chunk中LLM输出能成功解析成QAPairList的比例。这个指标反映管线是否可靠正常应该在95%以上。信息覆盖抽检。人工从原始PDF里挑10个关键事实逐一检查转换出的问答对是否覆盖到了。检索命中率。把关键事实改写成测试问题跑retriever.invoke(test_question)看top-4结果里是否包含能回答该问题的QA文档。如果你做了“原始chunk”和“QA文档”两种索引可以对比它们的命中率。QA文档在query是问句时通常表现更好但原始chunk在召回细节事实时更有优势。两边的数据放在一起才能判断你的转换器到底值不值得用。我自己在这套改造里最深的一个体会是文档转换器不需要绑死在某个库上核心逻辑本身就几行代码难的永远是边界处理。不要为了省事锁死LangChain版本也不要因为doctran不能用了就把整个文档处理逻辑推倒重来。先跑通最小链路再根据评估数据去调切分、调Prompt、调缓存。最后分享一个小技巧QA文档的metadata里一定要保留原始页码和chunk_id。否则以后线上回答出了问题你连这个答案是哪个PDF里的哪一页来的都查不到那才是真的麻烦。 
长河君 · 2026/10/5 2:49:29 · 阅读 1.2k LangChain 1.0升级后DoctranQA失效？手写文档问答转换器替代方案
升级LangChain到1.0之后我的老项目第一次运行就红了一大片。最扎眼的是from langchain.document_transformers import DoctranQATransformer这一行。项目里那套把PDF转成问答对再做检索增强的流程全靠doctran在顶着换到新版本后import直接报错网上搜了一圈答案基本都是“装回0.2.x”或者“别用LangChain了”。我不甘心为了一个文档转换器把整个LangChain生态锁死在旧版所以花了一晚上把它拆开重写了一遍。这篇文章就是那次改造的完整记录包括替代方案的设计思路、可以直接抄走的代码以及跑生产环境前必须处理的几个坑。内容不挑模型厂商OpenAI、Ollama、Azure都能接适合正在从旧版迁移或者想在LangChain 1.0里自建文档问答转换能力的同学。1. doctran为什么在LangChain 1.0里沦为了“遗留组件”1.1 doctran当年做的事先说清楚doctran是什么。它并不是LangChain官方维护的库而是一个第三方文档转换工具但因为功能贴合LLM应用场景在早期LangChain文档里拿到了“一等公民”的位置。它的核心能力是把原始文档改写成更适合模型使用的形态。常用转换器包括DoctranQATransformer把一个纯文本块变成若干个“问题答案”的组合。DoctranTextCleaner压缩语气、删除冗余、规整表达。DoctranDateExtractor从文本中识别并标准化日期方便做时间过滤。DoctranPropertyExtractor抽取出文档里的关键属性比如公司名、产品名。其中DoctranQATransformer最常用。它在底层做的事情其实并不复杂拿到一段文本调用一次LLM告诉模型“请从这段文本里提取3到5个问答对”再把模型返回的结果拆成新的Document对象。这些问答对可以直接丢进向量库做RAG也可以作为模型微调的种子数据。这个思路本身是很漂亮的。因为用户提问通常是“XX的价格是多少”“XX流程的第几步是什么”而文档原文可能是大段描述。把原文预先切成问答对相当于给检索器做了一层语义对齐命中率会比直接拿长段落去embedding高不少。1.2 升级之后到底断在哪里LangChain 1.0发布后老代码报错的链条是这样的langchain.document_transformers这个路径被清理第三方转换器不再从主包导出DoctranQATransformer直接import不到。doctran底层调用的是旧版OpenAI客户端和langchain-openai现在的模型抽象不一致即使你把import路径改对内部构造参数也对不上。LangChain 1.0核心全面转向pydantic v2doctran没有同步跟进导致类型校验在运行时各种报错。本质上不是“不能兼容”而是“没人维护这种兼容”。doctran的核心逻辑和LangChain耦合很深LangChain一升级它就要跟着改而它本身又不是官方项目社区贡献者没有足够的动力持续跟进。doctran旧功能LangChain 1.0现状替代方式问答转换官方无直接替代自己实现核心代码不过几十行文本清理官方只有切分器正则自定义清理函数日期提取官方无直接替代dateutil.parser正则属性提取官方推荐结构化输出with_structured_output这里最坑的是第一个问答转换看起来简单但网上几乎没有“1.0可直接运行”的完整示例。老教程清一色是from langchain.document_transformers import DoctranQATransformer代码抄下来直接就是红的。1.3 不降级的底线我的结论很明确不要为了一个文档转换器把版本钉死在0.2.x。为什么LangChain 1.0带来的不只是版本号变化而是一整套接口重构。Runnable协议、LangGraph的Agent状态管理、with_structured_output、新的callback系统这些都是在旧版本上很难平滑使用的。你为doctran锁一次版本之后每一次想用新特性都要重新考虑兼容性成本会越来越高。而且自己实现一个QA转换器并不复杂。核心就三步切块、让LLM生成问答对、把结果装回Document。doctran替你封装的是“调用姿势”不是“算法黑科技”。只要理解了这一点替代它就是一次很普通的工程改造。2. 不降级的替代架构三个可替换部件2.1 文档加载与清洗第一步还是要把PDF、Word、HTML等原始格式变成纯文本。langchain_community里提供了PyPDFLoader用起来很简单from langchain_community.document_loaders import PyPDFLoader loader PyPDFLoader(产品白皮书.pdf) docs loader.load()它会按页返回每个Document的page_content是一页文字metadata里带source和page。这个page字段后面非常有价值它能让问答对回到原始文档的物理位置。但这里有一个很常见的问题PDF提取出来的文本经常有断行。一页文本可能是每行一句甚至每行半句直接拿去切分语义是碎的。所以我一般会在切分之前加一道清洗函数import re def normalize_paragraph(text): lines [line.strip() for line in text.splitlines()] paragraphs [] buf [] for line in lines: if not line: if buf: paragraphs.append(.join(buf)) buf [] continue # 跳过疑似页码的独立数字 if re.fullmatch(r\d, line): continue # 跳过疑似页眉页脚的短句 if len(line) 30 and line.isupper(): continue # 如果上一行末尾不是句号、冒号、逗号就拼接否则换成句号拼接 if buf and not re.search(r[。.!?]$, buf[-1]): buf.append(line) elif buf: buf.append(。 line) else: buf.append(line) if buf: paragraphs.append(.join(buf)) return \n\n.join(paragraphs)这个函数不追求完美但能把绝大多数PDF常见的断行问题解决掉。清洗后再把文本放回Documentcleaned_docs [] for doc in docs: text normalize_paragraph(doc.page_content) if text: cleaned_docs.append(Document(page_contenttext, metadatadoc.metadata))这一步不做好的话后面LLM生成的问答质量会直接打折。2.2 切分策略决定问答质量上限清洗完成之后要切分。切分是整个流程的质量天花板比Prompt设计更重要。为什么因为LLM只能看到你给它的那一段文本。如果一段文本里混杂了三个互不相关的主题模型生成的问题就会失焦回答也会带上无关信息。RecursiveCharacterTextSplitter是默认选择它按[\n\n, \n, , ]的优先级逐层找切点对普通技术文档和产品白皮书都够用from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size1200, chunk_overlap200, ) chunks splitter.split_documents(cleaned_docs) for idx, chunk in enumerate(chunks): chunk.metadata[chunk_id] idx参数怎么调chunk_size1200是我在大多数场景下的起点。太小比如300到500单个chunk信息量不足LLM生成的问题会很琐碎太大比如2000以上模型容易漏掉细节而且输出延迟明显上升。chunk_overlap200是为了防止一个完整知识点被截断成两半。但overlap不是越大越好它会让相邻chunk重复内容后面LLM调用次数和成本都会增加。如果你的内容语义结构很强比如会议纪要、技术文档可以试试SemanticChunker。它通过embedding相似度来判断语义边界切出来的块更“自然”。但代价是要额外调embedding模型处理速度也慢。我的建议是第一版先用RecursiveCharacterTextSplitter等评估出来确实有语义割裂问题再升级。2.3 QA生成层切分完成后每个chunk会独立进入LLM生成问答对。这一层要解决两件事输出格式怎么约定问题质量怎么控制。输出格式我强烈建议用LangChain 1.0的with_structured_output绑定一个Pydantic模型。它会走模型原生的tool calling / function calling能力比在Prompt里写“请返回JSON”可靠得多。这个问题后面专门讲。问题质量要从Prompt层面约束。最核心的一条规则是“答案必须能在给定文本中找到依据不得补充外部知识。”否则模型会在你的知识库里一本正经地胡编。其次要约束问题类型让模型优先提取“指标、定义、步骤、结论”这类有独立检索价值的信息而不是生成“这段话讲了什么”这种低质量问题。3. 完整实现在LangChain 1.0里复刻DoctranQA转换器3.1 依赖与准备先装依赖pip install langchain1.0.0 langchain-openai langchain-text-splitters langchain-community pypdf pydantic各包职责langchain-coreDocument对象、BaseDocumentTransformer协议。langchain-openaiChatOpenAI是大模型调用入口。langchain-text-splittersRecursiveCharacterTextSplitter。langchain-communityPyPDFLoader等文档加载器。pypdfPDF解析底层库。如果你用的是Ollama这类本地模型把langchain-openai替换成langchain-ollama就行后面的ChatOllama调用方式和ChatOpenAI几乎一样。3.2 Pydantic Schema与Prompt设计先定义输出结构from pydantic import BaseModel, Field from typing import List class QAPair(BaseModel): question: str Field(..., description能独立回答的清晰问题) answer: str Field(..., description严格来自原文内容的答案) class QAPairList(BaseModel): qa_pairs: List[QAPair] Field(..., description从当前文档块中提取的问答列表)Field里的description很重要它会被拼进模型的tool schema里等于你在用元信息告诉模型每个字段该填什么。再写Prompt模板QA_PROMPT_TEMPLATE 你是一个文档问答转换器。 下面是一段文档内容请从中提取 {max_questions} 个问答对。 要求 1. 问题要有明确的信息量比如“某某指标是多少”“流程的第几步是什么”。 2. 所有答案必须能在给定文档中找到依据不得补充文档外的知识。 3. 如果文档内容不足以凑满 {max_questions} 个问题生成多少算多少。 4. 回答保留数字、百分比、人名、版本号等关键信息。 文档内容 {content} 注意第3条。很多模型在收到“生成5个问题”的指令后即使文本信息不够也会硬凑。硬凑出来的问答不仅没有检索价值还会污染向量库。这条约束能把硬凑概率压下来。3.3 实现 QATransformer核心类如下import logging from langchain_core.documents import Document from langchain_openai import ChatOpenAI logger logging.getLogger(__name__) class QATransformer: 在LangChain 1.0中替代DoctranQATransformer的最小实现。 def __init__(self, llm, max_questions3): self.llm llm.with_structured_output(QAPairList) self.max_questions max_questions def transform_documents(self, documents): new_docs [] for doc in documents: content doc.page_content.strip() if not content: continue prompt QA_PROMPT_TEMPLATE.format( contentcontent, max_questionsself.max_questions ) try: result self.llm.invoke(prompt) except Exception: logger.warning(QA生成失败跳过 chunk%s, doc.metadata) continue for idx, qa in enumerate(result.qa_pairs): page f问题{qa.question}\n回答{qa.answer} metadata { **doc.metadata, question: qa.question, answer: qa.answer, qa_index: idx, } new_docs.append(Document(page_contentpage, metadatametadata)) return new_docs这里有一个关键设计把question和answer同时放进metadata。这样后面做评估、去重、过滤时不需要重新解析page_content直接用doc.metadata[question]就行。另外metadata里保留了原chunk的chunk_id和page出问题的时候能一路回溯到原始PDF的物理位置。有一点要说明这个类没有继承BaseDocumentTransformer因为LangChain 1.0里不同子包的导出路径略有差异。实际上LangChain的文档转换器在使用时是鸭子类型只要你有transform_documents方法就能被TextSplitter之后的流程当转换器用。如果你更想要正式一点可以试试from langchain_core.documents.transformers import BaseDocumentTransformer class QATransformer(BaseDocumentTransformer): ...如果你的版本能import到就继承不能就用上面那个裸类不影响使用。3.4 从PDF开始跑通全流程把清洗、切分、转换串起来from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import PyPDFLoader from langchain_openai import ChatOpenAI # 1. 加载PDF loader PyPDFLoader(产品白皮书.pdf) raw_docs loader.load() # 2. 清洗断行 cleaned_docs [] for doc in raw_docs: text normalize_paragraph(doc.page_content) if text: cleaned_docs.append(Document(page_contenttext, metadatadoc.metadata)) # 3. 切分 splitter RecursiveCharacterTextSplitter(chunk_size1200, chunk_overlap200) chunks splitter.split_documents(cleaned_docs) for idx, chunk in enumerate(chunks): chunk.metadata[chunk_id] idx # 4. 初始化模型和转换器 llm ChatOpenAI(modelgpt-4o-mini, temperature0) transformer QATransformer(llm, max_questions3) # 5. 转换 qa_docs transformer.transform_documents(chunks) print(f原始文档块: {len(chunks)} 个生成问答对: {len(qa_docs)} 个) for doc in qa_docs[:3]: print(---) print(doc.page_content)跑出来的效果大概是这样的原始文档块: 12 个生成问答对: 31 个 --- 问题产品A的定价策略包含哪三档 回答产品A的定价策略包含标准版、专业版和企业版三档标准版按年订阅企业版支持私有化部署。这里temperature0是必须的。问答转换要的是忠实提取不是创意发挥。温度调高之后模型会在你没注意的地方“润色”原文这对RAG场景是灾难。4. 生产环境里真正值得注意的坑4.1 分块大小与模型上下文窗口的匹配很多人在本地模型上跑这套代码发现结果很差第一反应是模型能力不行。但很多时候问题出在分块大小和模型上下文窗口不匹配。假设你用的是7B参数本地模型上下文窗口是8192 tokens。你的chunk有1200 tokens加上生成的3个问答对、tool schema、系统提示词总共可能到2500到3000 tokens看上去没问题。但是如果把chunk_size调到6000整个请求已经逼近8000模型还要分配精力去生成结构化输出结果就是前言不搭后语。一个粗略的估算公式content_tokens max_questions * 80 500 context_limit * 0.7500是给tool schema和系统提示词留的余量。如果你的chunk token数超出这个范围优先调小chunk_size而不是换更强模型。另外建议把chunk大小从500到1500每隔200扫一遍对同一篇文档各跑一次对比生成问答的“忠实度”和“信息覆盖度”。这道工序花不了多少时间但能省掉后面很多排查功夫。4.2 结构化输出失败与兜底重试with_structured_output虽然比“Prompt返回JSON”可靠但也不是100%成功。模型可能返回空字段可能JSON schema校验失败也可能因为网络超时根本没返回。我实际跑下来gpt-4o-mini在tool calling模式下的失败率很低但本地7B模型失败率能达到5%到10%。建议做一个带重试的生成方法def generate_qa_pairs(transformer, content, max_questions, retries2): prompt QA_PROMPT_TEMPLATE.format( contentcontent, max_questionsmax_questions ) for attempt in range(retries): try: return transformer.llm.invoke(prompt) except Exception: logger.warning(第 %s 次生成失败降低问题数重试, attempt 1) if attempt 0: prompt QA_PROMPT_TEMPLATE.format( contentcontent, max_questions1 ) return QAPairList(qa_pairs[])重试时的关键技巧是第二次不要沿用原来的max_questions要把问题数降到1。因为很多失败是模型输出长度不够、生成到一半被截断问题数减少后输出token数量也减少成功率会明显提高。4.3 成本、限流与缓存成本很容易被低估。一个10页的PDF按每页800字算切分成1200字符的chunk大概有6到8个chunk每个chunk生成3个问答对一次PDF要做6到8次LLM调用。如果文档库有1000个PDF就是6000到8000次调用。用gpt-4o-mini还能接受换成claude或大尺寸模型账单会非常难看。我的建议是加缓存。最简单的做法是用chunk内容做键import hashlib import json class CachedQATransformer(QATransformer): def __init__(self, llm, max_questions3, cacheNone): super().__init__(llm, max_questions) self.cache cache if cache is not None else {} def transform_documents(self, documents): new_docs [] for doc in documents: content doc.page_content.strip() if not content: continue key hashlib.sha256(content.encode()).hexdigest() if key in self.cache: qa_pairs self.cache[key].qa_pairs else: result self.generate_with_retry(content) qa_pairs result.qa_pairs self.cache[key] result ...不要小看这个缓存。文档库里的很多PDF会有重复章节不同版本的合同、白皮书里大段文字完全相同。缓存能让你第二次全量重建向量库时几乎不花LLM费用。并发方面ChatOpenAI自身有max_retries参数处理限流但真正大批量处理时建议加信号量。简单起见可以先同步跑等数据量真的上来了再改成asyncio.gatherSemaphore(5)把并发压到5左右既能提速又不会触发太多限流。4.4 不同模型提供商的适配如果你用的是Ollamafrom langchain_ollama import ChatOllama llm ChatOllama(modelqwen2.5:7b, temperature0)但这里有个前提with_structured_output依赖模型支持tool calling / function calling。Ollama很多模型是支持的但支持程度参差不齐。如果发现结构化输出频繁失败可以退回传统方案用PydanticOutputParser把format instructions写进Promptfrom langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import PromptTemplate parser PydanticOutputParser(pydantic_objectQAPairList) prompt PromptTemplate( templateQA_PROMPT_TEMPLATE \n{format_instructions}, input_variables[content, max_questions], partial_variables{format_instructions: parser.get_format_instructions()}, ) chain prompt | llm | parser result chain.invoke({content: content, max_questions: 3})这种方式兼容性最好代价是让模型自己生成JSON解析失败率和平均延迟都会略高一点。对有tool calling但不稳定的本地模型这是性价比更高的选择。5. 接入RAG与验证转换器到底有没有用5.1 把问答对写入向量库转换器跑完之后qa_docs就是一批“问题回答”样式的Document。接下来直接建向量库from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS embedding OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore FAISS.from_documents(qa_docs, embedding) retriever vectorstore.as_retriever(search_kwargs{k: 4})为什么不直接索引原始chunks而是索引问答对因为语义对齐。用户的问题通常是问句问句和问句的嵌入向量距离更近而问句和长段落的嵌入距离可能很远。检索器在找“产品A的定价策略包含哪三档”时更容易命中“问题产品A的定价策略包含哪三档”这条QA文档。但也要注意它的代价QA转换是信息有损的。模型只会生成3到5个问答对如果原文某个关键信息没被转换成问题用户怎么搜都搜不到。所以我推荐混合索引原始chunks和QA文档都写入向量库用metadata[doc_type]区分检索时同时召回两者。5.2 简单问答链路转换后的结果可以直接接一个最小的RAG链路from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template( 基于以下检索到的问答片段回答用户问题。 片段 {context} 用户问题{question} 如果片段中没有相关信息请直接说“未知”不要编造。 ) def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) def ask(question): docs retriever.invoke(question) context format_docs(docs) response llm.invoke(prompt.format(contextcontext, questionquestion)) return response.content这个链路看起来简单但足够验证转换器的效果。如果ask的回答经常是“未知”说明QA转换后信息覆盖不够需要调大max_questions或者调整切分策略。5.3 验证思路验证不能靠感觉。我建议按以下三步走格式成功率。统计全部chunk中LLM输出能成功解析成QAPairList的比例。这个指标反映管线是否可靠正常应该在95%以上。信息覆盖抽检。人工从原始PDF里挑10个关键事实逐一检查转换出的问答对是否覆盖到了。检索命中率。把关键事实改写成测试问题跑retriever.invoke(test_question)看top-4结果里是否包含能回答该问题的QA文档。如果你做了“原始chunk”和“QA文档”两种索引可以对比它们的命中率。QA文档在query是问句时通常表现更好但原始chunk在召回细节事实时更有优势。两边的数据放在一起才能判断你的转换器到底值不值得用。我自己在这套改造里最深的一个体会是文档转换器不需要绑死在某个库上核心逻辑本身就几行代码难的永远是边界处理。不要为了省事锁死LangChain版本也不要因为doctran不能用了就把整个文档处理逻辑推倒重来。先跑通最小链路再根据评估数据去调切分、调Prompt、调缓存。最后分享一个小技巧QA文档的metadata里一定要保留原始页码和chunk_id。否则以后线上回答出了问题你连这个答案是哪个PDF里的哪一页来的都查不到那才是真的麻烦。