从模型输出到知识库检索LangChain 数据管道实战大模型能够生成自然、连贯的文本但真正把它接入业务系统时仅仅“能回答问题”远远不够。程序需要稳定的数据结构知识库需要统一的文档对象检索系统还要把长文档切成大小合适、语义完整的片段。这三件事分别对应 LangChain 中的三类基础组件输出解析器、文档加载器和文本分割器。它们共同组成了一条重要的数据管道原始文件 - Document - 文档块 - 检索上下文 - 模型回答 - 结构化结果本文从工程视角拆解这条管道重点说明每个组件解决什么问题、如何选择以及怎样把它们组合成可维护的 RAG 应用。为什么模型输出必须经过解析大模型的原始输出通常是一段自然语言。人可以轻松读懂但程序很难可靠处理。例如业务接口可能期待如下数据{answer:退款通常会在 3 个工作日内到账。,sources:[售后服务说明],confidence:0.92}如果模型有时返回 JSON有时添加解释文字有时改变字段名下游代码就会变得非常脆弱。输出解析器的价值就是把模型的非结构化回答转换为字符串、字典、列表或 Pydantic 对象让模型调用从“文本对话”变成可验证的数据接口。只需要文本时使用 StrOutputParser聊天模型通常返回AIMessage文本位于content字段中。StrOutputParser可以直接取出文本内容并且实现了 Runnable 接口可以自然地接入 LCEL 管道。fromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini,temperature0)chainmodel|StrOutputParser()resultchain.invoke(用一句话解释 RAG。)print(result)流式调用时解析器也可以继续向下游传递文本片段forchunkinchain.stream(写一段不超过 80 字的产品简介。):print(chunk,end,flushTrue)如果最终只需要展示回答StrOutputParser通常已经足够。需要强类型对象时使用 PydanticOutputParser当模型结果要进入数据库、工作流或后端接口时更适合先定义一个明确的数据模型再让解析器负责格式说明和结果校验。fromtypingimportOptionalfrompydanticimportBaseModel,Fieldfromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportPydanticOutputParserfromlangchain_core.promptsimportPromptTemplateclassJoke(BaseModel):setup:strField(description笑话的开头)punchline:strField(description笑话的包袱)rating:Optional[int]Field(defaultNone,ge1,le10,description1 到 10 分的评分,)modelChatOpenAI(modelgpt-4o-mini,temperature0)parserPydanticOutputParser(pydantic_objectJoke)promptPromptTemplate(template(请根据用户主题创作一个简短笑话。\n{format_instructions}\n主题{topic}),input_variables[topic],partial_variables{format_instructions:parser.get_format_instructions()},)chainprompt|model|parser resultchain.invoke({topic:程序员写单元测试})print(result.setup)print(result.punchline)print(result.rating)这里最关键的方法是get_format_instructions()。它根据 Pydantic 模型生成格式约束并把约束注入提示词。模型负责按格式生成解析器负责把结果转换成对象并校验字段。需要通用字典时使用 JsonOutputParser如果下游系统只要求 JSON 或 Python 字典不需要强类型对象可以使用JsonOutputParserfromlangchain_core.output_parsersimportJsonOutputParserfromlangchain_core.promptsimportPromptTemplate parserJsonOutputParser()promptPromptTemplate(template(提取文本中的姓名和城市。\n{format_instructions}\n文本{text}),input_variables[text],partial_variables{format_instructions:parser.get_format_instructions()},)chainprompt|model|parser resultchain.invoke({text:李华毕业后前往杭州工作。})print(result)除了字符串、Pydantic 和 JSONLangChain 还提供 XML、YAML、CSV、枚举、日期等解析器。选择原则很简单下游需要什么类型就让解析器输出什么类型。输出解析器与 with_structured_output 如何选择二者都可以获得结构化结果但控制位置不同PydanticOutputParser是独立组件格式说明、提示词和解析过程都显式可见适合需要精细控制提示词或统一解析逻辑的场景。with_structured_output()是聊天模型的方法通常会优先利用模型提供方的工具调用或结构化输出能力代码更短也更贴近模型原生能力。structured_modelmodel.with_structured_output(Joke)resultstructured_model.invoke(讲一个关于数据库索引的笑话)如果模型提供方原生支持稳定的结构化输出可以优先考虑with_structured_output()如果需要把解析器作为独立 Runnable 复用或需要明确控制格式提示则使用输出解析器更合适。无论采用哪一种方式都不应假设模型永远会返回合法数据。生产环境还需要捕获解析异常、记录原始响应并根据业务风险决定重试、降级还是拒绝结果。用 Document 统一不同数据源RAG 系统可能要处理 PDF、Markdown、网页、数据库记录甚至代码文件。如果每种来源都保留自己的数据结构后续的切分、向量化和检索会非常复杂。LangChain 使用Document作为统一抽象。一个Document主要包含三类信息page_content正文文本。metadata来源、页码、标题层级、文件名等元数据。id可选的唯一标识。fromlangchain_core.documentsimportDocument documents[Document(page_content退款申请提交后通常会在 3 个工作日内完成审核。,metadata{source:售后服务说明.md,category:退款规则,},),Document(page_content审核通过后款项会原路退回。,metadata{source:售后服务说明.md,category:退款规则,},),]正文用于向量化和模型上下文元数据则用于追踪来源、过滤检索结果和生成引用。很多 RAG 系统能检索到正确内容却无法告诉用户答案来自哪里根本原因往往不是模型能力不足而是数据加载时丢失了元数据。加载 PDF 文档PyPDFLoader可以把 PDF 转换为Document列表。默认情况下每一页通常对应一个对象页码和文件来源会保存在元数据中。fromlangchain_community.document_loadersimportPyPDFLoader loaderPyPDFLoader(./docs/product-manual.pdf)docsloader.load()print(f文档页数{len(docs)})print(docs[0].page_content[:200])print(docs[0].metadata)按页加载有两个明显优势一是可以保留页码方便答案引用二是某一页解析失败时更容易定位问题。但 PDF 并不等于纯文本。遇到扫描件、多栏排版、复杂表格或图表时普通文本抽取可能会出现顺序错乱或内容缺失。此时应根据文件特征增加 OCR、版面分析或多模态模型而不是只调整切分参数。加载 Markdown 文档Markdown 天然带有标题、列表、表格等结构适合构建高质量知识库。可以使用UnstructuredMarkdownLoader读取pipinstallunstructured[md]nltksingle模式把整个文件作为一个Document返回fromlangchain_community.document_loadersimportUnstructuredMarkdownLoader loaderUnstructuredMarkdownLoader(./docs/faq.md,modesingle,)docsloader.load()print(len(docs))print(docs[0].metadata)elements模式会根据标题、叙述文本、列表项、表格、图片等元素拆分文档loaderUnstructuredMarkdownLoader(./docs/faq.md,modeelements,)elementsloader.load()forelementinelements[:3]:print(element.page_content)print(element.metadata.get(category))元素模式通常会产生更丰富的元数据例如category元素类型如Title、ListItem、Table、NarrativeText。element_id当前元素的唯一标识。parent_id父元素标识可用于恢复标题与正文之间的层级关系。如果知识库非常依赖文档结构例如需要按标题过滤、保留问答层级或单独处理表格元素模式更有价值如果后续会统一使用自己的分割策略单文档模式通常更简单。RAG 的离线处理与在线检索文档加载器和文本分割器主要工作在离线阶段输出解析器则更多出现在在线回答阶段。一条完整的 RAG 链路可以分成两个过程。离线数据处理加载文档 - 清洗内容 - 切分文档 - 生成向量 - 写入向量数据库在线检索接收问题 - 检索相关文档块 - 拼装提示词 - 调用模型 - 解析结果离线处理决定“知识是否进入系统”在线检索决定“正确知识能否在正确时刻被取出”。如果加载阶段漏掉了正文或者切分阶段破坏了语义后面的模型再强也无法凭空恢复信息。文本切分为什么会影响检索质量长文档不适合直接送入向量模型和聊天模型。一方面大块文本难以与具体问题精确匹配另一方面模型上下文窗口有限过大的片段会挤占输入空间并增加成本。但切得越小也不一定越好。片段过小会丢失上下文例如只保留“审核通过后”而没有保留前文中的“退款申请”检索结果即使命中也可能无法独立表达完整含义。文本切分本质上是在三个目标之间取平衡检索粒度块越小定位通常越精确。语义完整性块越大信息通常越完整。上下文与成本块越大占用的 Token 越多。基于字符长度切分CharacterTextSplitter会围绕指定分隔符组织文本块并用字符数衡量长度。fromlangchain_text_splittersimportCharacterTextSplitter text_splitterCharacterTextSplitter(separator\n\n,chunk_size500,chunk_overlap80,length_functionlen,is_separator_regexFalse,)chunkstext_splitter.split_documents(docs)几个核心参数需要一起理解separator优先在哪些位置断开例如段落之间的双换行。chunk_size目标块大小。chunk_overlap相邻块重复保留的内容用于维持上下文连续性。length_function如何计算长度len代表字符数。chunk_size是目标值不一定是绝对上限。如果某个完整段落本身已经超过目标大小而当前分隔策略无法继续拆分分割器可能保留整个段落并给出超长提示。这通常是在语义完整性和长度限制之间选择前者。处理这类提示时应先查看超长块的分布如果大部分块都超长说明目标值可能太小如果只有少量异常块则应检查文档中是否存在超长段落、URL、表格或缺失换行。基于 Token 长度切分模型按 Token 而不是按字符计费和限制上下文。中英文、数字和符号的 Token 比例并不相同因此对接特定模型时Token 长度通常比字符长度更准确。fromlangchain_text_splittersimportCharacterTextSplitter text_splitterCharacterTextSplitter.from_tiktoken_encoder(encoding_namecl100k_base,chunk_size300,chunk_overlap50,)chunkstext_splitter.split_documents(docs)这里chunk_size和chunk_overlap使用 Token 数衡量。编码方式需要与目标模型尽量匹配否则估算会产生偏差。需要注意改变长度计算方式并不会自动解决语义边界问题。即使按 Token 计数如果分隔器仍然不愿破坏一个完整的大段落也可能生成超过目标值的块。需要硬性上限时使用递归切分如果任何文本块都不能超过指定大小可以使用RecursiveCharacterTextSplitter。它会按分隔符优先级逐层尝试先按段落再按行、标点、空格最后才退化为更细的字符边界。fromlangchain_text_splittersimportRecursiveCharacterTextSplitter text_splitterRecursiveCharacterTextSplitter.from_tiktoken_encoder(encoding_namecl100k_base,chunk_size300,chunk_overlap50,separators[\n\n,\n,。,,,,,.,,, ,,],)chunkstext_splitter.split_documents(docs)中文文本应显式加入中文句号、逗号、问号和分号等分隔符。如果只使用空格和换行中文词组更容易在不自然的位置被切开。递归切分可以更严格地控制大小但代价是极端情况下会牺牲语义完整性。因此硬性限制不应该只是为了让日志更干净而应来自明确的模型上下文、向量模型限制或接口约束。特殊文档优先使用结构感知切分纯字符切分不了解文档语法。对于结构明确的数据按结构切分通常更合理Markdown按#、##、###等标题层级切分。HTML按标题或其他标签切分。JSON按对象或数组元素切分。代码按类、函数或逻辑块切分。例如Python 代码可以使用专门的分割器fromlangchain_text_splittersimportPythonCodeTextSplitter python_code class PriceService: def calculate(self, amount, discount): return amount * (1 - discount) def format_price(value): return f¥{value:.2f} splitterPythonCodeTextSplitter(chunk_size120,chunk_overlap20,)chunkssplitter.create_documents([python_code])结构感知切分能够尽量让一个函数、一个标题下的内容或一个 JSON 对象保持在同一块中通常比固定字符截断更适合检索。把加载、切分和输出解析串起来下面用一个简化示例展示三类组件如何协作。向量检索部分可以替换为 Redis、Chroma、Pinecone 或其他向量存储关键是保持输入输出边界清晰。frompathlibimportPathfromtypingimportListfrompydanticimportBaseModel,Fieldfromlangchain_community.document_loadersimport(PyPDFLoader,UnstructuredMarkdownLoader,)fromlangchain_core.documentsimportDocumentfromlangchain_core.output_parsersimportPydanticOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAIfromlangchain_text_splittersimportRecursiveCharacterTextSplitterdefload_documents(file_path:str)-List[Document]:pathPath(file_path)suffixpath.suffix.lower()ifsuffix.pdf:returnPyPDFLoader(str(path)).load()ifsuffixin{.md,.markdown}:returnUnstructuredMarkdownLoader(str(path),modesingle).load()raiseValueError(f不支持的文件类型{suffix})defsplit_documents(docs:List[Document])-List[Document]:splitterRecursiveCharacterTextSplitter.from_tiktoken_encoder(encoding_namecl100k_base,chunk_size400,chunk_overlap60,separators[\n\n,\n,。,,,,, ,],)returnsplitter.split_documents(docs)classRagAnswer(BaseModel):answer:strField(description只依据上下文生成的答案)sources:List[str]Field(description答案使用的来源)found:boolField(description上下文中是否存在足够信息)defformat_context(docs:List[Document])-str:parts[]forindex,docinenumerate(docs,start1):sourcedoc.metadata.get(source,unknown)pagedoc.metadata.get(page)locationf{source}#page{page1}ifpageisnotNoneelsesource parts.append(f[文档{index}]\n来源{location}\n内容{doc.page_content})return\n\n.join(parts)parserPydanticOutputParser(pydantic_objectRagAnswer)promptChatPromptTemplate.from_template( 你是一个知识库问答助手只能根据给定上下文回答。 如果上下文不足请明确说明并将 found 设置为 false。 {format_instructions} 问题{question} 上下文 {context} ).partial(format_instructionsparser.get_format_instructions())modelChatOpenAI(modelgpt-4o-mini,temperature0)answer_chainprompt|model|parser# 离线阶段加载、切分并将 chunks 写入向量数据库。docsload_documents(./docs/product-manual.pdf)chunkssplit_documents(docs)# 在线阶段这里用检索器返回的结果替换示例变量。retrieved_docschunks[:3]resultanswer_chain.invoke({question:退款审核通常需要多长时间,context:format_context(retrieved_docs),})print(result.model_dump())这个示例中每个组件都只承担一种职责加载器适配数据源分割器控制检索粒度解析器保证结果结构。它们之间通过Document、字符串和 Pydantic 对象建立清晰边界后续替换模型、向量数据库或文件格式时不需要重写整条链路。生产环境中的调优重点真正上线前建议重点检查以下问题解析成功率统计结构化结果的解析失败率并保留失败时的原始模型响应。元数据完整性至少保留来源、页码或标题路径保证答案可以追溯。空内容与乱码加载后先检查空页面、异常编码和重复页眉页脚。块大小分布不要只看平均值还要观察最大值、分位数和超长块数量。重叠比例重叠过小可能丢上下文过大则会增加向量存储、检索重复和 Token 成本。中文分隔符根据语料加入中文标点避免在词组中间切断。结构化文档Markdown、HTML、JSON 和代码优先采用结构感知策略。检索评估准备一组真实问题验证正确片段能否进入 Top K而不是只观察最终回答是否流畅。输出解析、文档加载和文本切分看起来都是基础组件却直接决定了 LLM 应用的数据质量。可靠的系统并不是把模型接上接口就结束而是让输入有来源、过程有边界、输出有结构。把这条数据管道打磨好RAG 应用才会真正具备可检索、可验证和可维护的工程能力。
