前言：
很多人学 LangChain，学到最后代码写得一团糟：Prompt、模型、解析器各调各的，处理文档自己写循环读文件，加个功能要改好几个地方，逻辑散得像一锅粥。
说白了，你不是在用框架，你是在用零散的 API。

真正工程化的 LangChain 开发，核心就两件事：
一是用 LCEL 把所有组件串成标准链路，可复用可扩展；
二是把非结构化的文档转换成模型能用的知识块，为 RAG 打好基础。

这篇是 LangChain 系列第三篇，我把 Runnable 体系、LCEL 链式语法、三种调用方式、并行/透传高阶用法，以及文档加载、文本切分、知识库预处理全流程一次性讲透。
全是可直接落地的代码，跟着敲完，你就能写出真正工程化的大模型应用，为后面的 RAG 知识库铺平道路。

---

一、LCEL：告别零散调用，用管道串起所有组件

先问你一个问题：之前写的代码，是不是都是这样？

prompt = prompt_template.invoke(data)
response = model.invoke(prompt)
result = parser.invoke(response)

三步分开写，短的时候没问题，等组件多了、链路复杂了，维护起来就是灾难。
加个预处理函数要插在最前面，加个并行分支要自己写多线程，改个组件要翻半天代码。

LCEL 就是解决这个问题的。

1. 先搞懂两个核心概念

Runnable：所有组件的统一接口

LangChain 里几乎所有组件都是 Runnable：

• Prompt 模板
• 大模型
• Output Parser
• 检索器
• 你自己写的函数（包装后）

所有 Runnable 都有统一的调用方法：

方法	作用	适用场景
invoke	单条输入处理	单次问答、单条数据处理
batch	批量输入处理	批量生成文案、批量分类
stream	流式逐块输出	聊天助手、长文本生成

LCEL：用管道把 Runnable 串起来

LCEL 全称 LangChain Expression Language，核心就是一个 | 管道符。
只要是 Runnable，就能用 | 串成一条处理链，数据从左往右依次流过每个组件。

最经典的三段式：

chain = prompt_template | model | parser
result = chain.invoke({"topic": "LangChain"})

和分步调用效果完全一样，但代码更干净，组件可插拔，而且自动继承 invoke/batch/stream 三种能力。

一句话总结：
Runnable 是标准零件，LCEL 是把零件拼起来的管道。
以后写 LangChain 代码，优先想「怎么串成链」，而不是「怎么一步步调」。

2. 基础实战：技术概念解释链

先写第一条完整的链，感受一下 LCEL 的简洁：

from langchain.chat_models import init_chat_model
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from utils.model_factory import get_deepseek_model

model = get_deepseek_model(0.5)

template = ChatPromptTemplate.from_messages([
    ("system", "你是一名编程讲师，擅长用简单语言解释技术概念。"),
    ("human", """
        请解释下面的技术概念：
        概念：{topic}
        学习者水平：{level}
        要求：
        1. 先给一句话定义
        2. 再给一个简单例子
        3. 不超过 200 字
    """)
])

parser = StrOutputParser()

# 串成链
chain = template | model | parser

# 单条调用
result = chain.invoke({
    "topic": "什么是RAG?",
    "level": "编程小白"
})
print(result)

3. 三种调用方式，按需使用

① invoke：单条处理

最常用，一次处理一条输入。

result = chain.invoke({"topic": "Embedding", "level": "初学者"})

② batch：批量处理

一次提交多条数据，自动批量执行，适合离线处理大量数据。

# 批量处理三个概念
results = chain.batch([
    {"topic": "Embedding", "level": "初学者"},
    {"topic": "向量数据库", "level": "初学者"},
    {"topic": "RAG", "level": "初学者"},
])

for res in results:
    print(res)
    print("-" * 50)

避坑提醒：
batch 不是无限并发的，太多请求会触发模型限流。
可以通过 config={"max_concurrency": 2} 限制最大并发数，避免 429 报错。

③ stream：流式输出

边生成边返回，适合聊天、写长文等需要实时反馈的场景。

for chunk in chain.stream({
    "topic": "RAG",
    "level": "编程小白"
}):
    print(chunk, end="", flush=True)

注意一定要加 flush=True，否则内容会攒在缓冲区，看不到逐字输出的效果。

4. 高阶用法一：RunnableLambda 把普通函数塞进链里

不是所有逻辑都要交给大模型，比如输入清洗、格式转换、数据预处理，用普通 Python 函数就行。
RunnableLambda 就是把普通函数包装成 Runnable，让它能进管道。

实战：带输入清洗的问答链

from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from utils.model_factory import get_deepseek_model
from utils.prompt_template import getPromptTemplate

model = get_deepseek_model()
template = getPromptTemplate(
    "你是一名 Python 讲师，请简洁回答学生问题。",
    "学生的问题是:{question}"
)
parser = StrOutputParser()

# 普通预处理函数：去空格、字段转换
def clean_input(qdict):
    question = qdict.get("wenti").strip()
    return {"question": question}

# 包装成 Runnable
cleaner = RunnableLambda(clean_input)

# 插入到链路最前面
chain = cleaner | template | model | parser

result = chain.invoke({
    "wenti": "    什么是闭包？    "
})
print(result)

业务逻辑和模型逻辑彻底分开，预处理函数改起来也方便。

5. 高阶用法二：RunnableParallel 并行执行多任务

同一条输入，需要同时做多个处理怎么办？比如一条评论，既要情感分析，又要提取关键词，还要生成回复。
总不能写三条链依次跑吧，太慢了。
RunnableParallel 就是干这个的：同一个输入同时分给多条子链并行执行，最后汇总成字典返回。

完整实战：评论分析流水线

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel
from utils.model_factory import get_deepseek_model
from utils.prompt_template import getPromptTemplate

model = get_deepseek_model()
parser = StrOutputParser()

# 三条子链
sentiment_chain = getPromptTemplate(
    "你是评论情感分析助手，只回答：正面、中性或负面。",
    "用户的评价是:{review}"
) | model | parser

keyword_chain = getPromptTemplate(
    "你是关键词提取助手，请提取 3 个关键词，用顿号分隔。",
    "用户的评价是:{review}"
) | model | parser

reply_chain = getPromptTemplate(
    "你是电商客服，请根据评论生成一段不超过 80 字的礼貌回复。",
    "用户的评价是:{review}"
) | model | parser

# 并行组装
analysis_chain = RunnableParallel(
    sentiment=sentiment_chain,
    keywords=keyword_chain,
    reply=reply_chain
)

result = analysis_chain.invoke({
    "review": "鼠标手感不错，也很安静，但是滚轮用了两周就有异响。"
})

print("情感：", result["sentiment"])
print("关键词：", result["keywords"])
print("回复：", result["reply"])

划重点：
并行执行总耗时 ≈ 最慢那条分支的时间，不是三条加起来。
业务里有多任务处理的场景，优先用 RunnableParallel，性能提升非常明显。

6. 高阶用法三：RunnablePassthrough 透传与新增字段

RAG 场景里最常用的组件，核心两个作用：

1. 原样透传上游输入数据
2. 保留原有字段的同时，新增计算出来的新字段

最基础的透传用法：

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from utils.model_factory import get_deepseek_model

model = get_deepseek_model()
template = PromptTemplate.from_template("""
    基于下面上下文回答用户问题：
    上下文："RunnablePassthrough 用于透传上游数据，可搭配 assign 新增字段。"
    用户问题：{question}
""")

chain = {"question": RunnablePassthrough()} | template | model | StrOutputParser()
result = chain.invoke("什么是RunnablePassthrough?")
print(result)

小技巧：管道里写 {key: value} 字典，LangChain 会自动包装成 RunnableParallel，不用手动写。

后面讲 RAG 的时候，我们会大量用到 RunnablePassthrough.assign() 来拼接检索到的上下文，这里先有个印象就行。

---

二、文档预处理：RAG 应用的第一步，决定了上限

前面我们处理的数据都是写死在代码里的字符串。
真实业务里，企业知识全在各种文件里：员工手册、产品说明书、售后规则、技术文档……
要让大模型用上这些知识，第一步就是把文件读出来、切合适大小，这就是文档预处理。

1. 完整知识库处理流程

先建立整体认知，RAG 的数据预处理全流程：

TXT/MD/PDF 文件
    ↓
Document Loader 加载
    ↓
Document 对象列表
    ↓
Text Splitter 切分
    ↓
小块 Document 列表
    ↓
向量化 → 存入向量数据库

这一章我们讲前半段：加载和切分。向量化和检索下一篇讲。

2. Document：LangChain 的文档标准对象

不管什么格式的文件，加载后都会统一变成 Document 对象，两个核心属性：

属性	作用
page_content	文档正文内容
metadata	元数据，记录来源、页码、分类等附加信息

from langchain_core.documents import Document

doc = Document(
    page_content="公司所有正式员工每年享有 5 天带薪年假。",
    metadata={
        "source": "员工手册.md",
        "category": "考勤制度",
    }
)

经验之谈：
元数据非常重要，千万别嫌麻烦不写。
后面检索出答案，要靠元数据告诉用户「这条答案来自哪份文件第几页」，这是企业知识库的刚需。

3. 三类常见文件加载

① TXT / Markdown 文件

最基础的文本文件，用 TextLoader 直接加载。

from langchain_community.document_loaders import TextLoader

loader = TextLoader(
    file_path="data/employee_handbook.txt",
    encoding="utf-8"
)

documents = loader.load()
print(f"文档数量：{len(documents)}")
print(f"内容：{documents[0].page_content}")
print(f"元数据：{documents[0].metadata}")

Markdown 文件也可以直接用 TextLoader 读取，纯文本内容不会丢失格式。

② PDF 文件

企业里最常见的格式，用 PyPDFLoader 加载，每页 PDF 会生成一个 Document 对象，元数据自动带上页码。

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(file_path="data/product_manual.pdf")
documents = loader.load()

print(f"总页数：{len(documents)}")
for doc in documents:
    print(f"第{doc.metadata['page']}页")
    print(doc.page_content[:50])

避坑提醒：

1. 只能加载可复制文字的文本型 PDF，扫描版 PDF 是图片，需要 OCR 才能识别
2. 页码从 0 开始计数，展示给用户的时候记得 +1
3. 排版复杂、表格多的 PDF，普通加载器效果不好，需要专门的解析工具

4. 为什么一定要切分文档？

一份员工手册几百页，总不能全塞给模型吧？
不切分会有几个致命问题：

• 超过模型上下文窗口，直接报错
• 每次请求带整份文档，token 成本爆炸
• 用户只问一个小问题，无关内容太多，模型容易答偏
• 向量检索无法精确定位，搜出来全是不相关的内容

所以必须把长文档切成几百字的小块，用户提问时只检索相关的几块，既省钱又准确。

5. 切分首选：RecursiveCharacterTextSplitter

LangChain 提供了很多切分器，普通中英文文档，直接用这个就够了。
它的原理是递归切分：优先按段落切，段落太长就按换行切，还长就按句号、逗号切，尽量保证语义完整，不会把一句话劈成两半。

两个核心参数：

参数	作用	建议值
chunk_size	每个文档块最大字符数	普通知识文档 300-500
chunk_overlap	相邻块重叠字符数	设为 chunk_size 的 10%-20%

重叠的作用是什么？
避免一句话刚好被切断，导致上下文丢失。比如前一块结尾和后一块开头保留一部分重复内容，保证每块的语义都是完整的。
但也不能太大，否则重复内容太多，浪费存储，检索结果也会重复。

6. 实战：切分文档并保留元数据

注意：切分用 split_documents() 方法，返回的还是 Document 列表，原有的元数据会自动继承。

from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1. 加载PDF
loader = PyPDFLoader(file_path="data/员工守则.pdf")
documents = loader.load()

# 2. 初始化切分器
splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=30,
    add_start_index=True, # 元数据里加上块在原文档中的起始位置
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
)

# 3. 切分
chunks = splitter.split_documents(documents)

print(f"切分前：{len(documents)} 页")
print(f"切分后：{len(chunks)} 个文档块")

# 每个块都保留了完整元数据
for i, chunk in enumerate(chunks[:3], start=1):
    print(f"\n第{i}块：")
    print(chunk.page_content[:50])
    print("元数据：", chunk.metadata)

7. 企业级实战：知识库批量预处理

最后写一个完整的工具函数：扫描整个知识库目录，自动识别文件类型加载，统一切分，补充元数据。

from pathlib import Path
from langchain_community.document_loaders import TextLoader, PyPDFLoader
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter

def load_knowledge_base(dir_path: str) -> list[Document]:
    """加载指定目录下所有 TXT/MD/PDF 文件"""
    kb_dir = Path(dir_path)
    all_docs = []

    for file_path in kb_dir.rglob("*"):
        if file_path.is_dir():
            continue
        
        suffix = file_path.suffix.lower()
        rel_path = str(file_path.relative_to(kb_dir.parent))

        if suffix in [".txt", ".md"]:
            loader = TextLoader(file_path=str(file_path), encoding="utf-8")
            docs = loader.load()
        elif suffix == ".pdf":
            loader = PyPDFLoader(file_path=str(file_path))
            docs = loader.load()
        else:
            continue # 跳过不支持的格式

        # 补充通用元数据
        for doc in docs:
            doc.metadata["file_name"] = file_path.name
            doc.metadata["file_type"] = suffix
            doc.metadata["file_path"] = rel_path
        
        all_docs.extend(docs)
    
    return all_docs

def split_documents(documents: list[Document]) -> list[Document]:
    """统一切分文档并添加块ID"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=300,
        chunk_overlap=50,
        add_start_index=True,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
    )
    chunks = splitter.split_documents(documents)
    
    # 给每个块加唯一ID
    for idx, chunk in enumerate(chunks, start=1):
        chunk.metadata["chunk_id"] = idx
    
    return chunks

# 使用
if __name__ == "__main__":
    raw_docs = load_knowledge_base("knowledge_base")
    print(f"加载完成，共 {len(raw_docs)} 份原始文档")
    
    chunked_docs = split_documents(raw_docs)
    print(f"切分完成，共 {len(chunked_docs)} 个知识块")
    
    # 预览前3块
    for chunk in chunked_docs[:3]:
        print("\n" + "="*50)
        print("来源：", chunk.metadata["file_name"])
        print("内容：", chunk.page_content[:80])

这就是企业知识库预处理的标准雏形，后面直接把切好的文档块扔给向量化模块就行。

---

三、给新手的 5 条实战心法

1. 能写成链的，就不要分步调用

不要图省事写三步 invoke，越早养成 LCEL 的习惯，后面代码越清爽。
加组件、改顺序，只需要动管道里的一行，扩展性天差地别。

2. 普通逻辑别麻烦大模型，用 RunnableLambda

输入清洗、格式转换、字段计算，Python 能搞定的就自己写函数，包装一下塞进链里。
又快又稳还省钱，别什么都丢给模型处理。

3. 文档预处理的质量，决定了 RAG 的上限

很多人做 RAG 效果差，上来就怪模型不行、向量数据库不行。
其实 80% 的问题出在预处理：文档切得稀碎，语义不完整，元数据全空。
把切分参数调好，把元数据补全，效果立马提升一大截。

4. chunk_size 没有标准答案，要结合业务调

不是所有文档都用 500 字一块。
FAQ 类文档可以小一点，技术文档可以大一点；问答场景小一点，总结场景大一点。
多测几组参数，看检索效果，不要机械抄别人的数值。

5. 元数据从一开始就要重视

不要等上线了要溯源，才发现所有文档都没存来源。
加载的时候就把文件名、页码、分类、部门这些信息补全，后面做过滤、溯源、权限控制都用得上。

---

最后

到这里，LangChain 的核心基础就全部打通了：模型调用、Prompt 模板、结构化输出、链式编排、文档预处理。
这些东西吃透，你已经具备了开发完整 RAG 应用的能力。

如果文章对你有帮助，欢迎点赞收藏，有问题评论区交流。