流程总览

Source(多源文件) → Load(文档加载) → Transform(文本分割清洗) → Embed(向量化) → Store(向量入库) → Retrieve(查询召回+LLM应答)

前 5 步：离线建库；

第 6 步：在线推理问答

环境安装

bash

运行

pip install langchain langchain-openai langchain-chroma python-dotenv

python

运行

from dotenv import load_dotenv
load_dotenv()
# 通用依赖导入
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import PromptTemplate
import os

# 初始化基础模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
embedding = OpenAIEmbeddings()
# 预设知识库文本（模拟多源Source数据）
source_text = """
公司差旅报销制度：
1. 一线城市（北上广深）住宿费标准350元/晚；二线城市280元/晚；三线及以下200元/晚。
2. 试用期员工不享受出差交通补贴。
3. 餐饮补贴统一80元/天，不分城市等级。
"""
# 落地为本地txt，模拟Source源文件
with open("./company_rule.txt","w",encoding="utf-8") as f:
    f.write(source_text)

---

步骤 1 Source：原始数据源接入

说明：PDF/Word/TXT/ 网页 / 数据库 / Excel 等异构数据，LangChain 用各类 Loader 对接 示例数据源：本地company_rule.txt文本文件

步骤 2 Load：文档加载

读取源文件，统一转为 LangChain 标准Document对象

python

运行

# 加载文件
loader = TextLoader(file_path="./company_rule.txt", encoding="utf-8")
docs = loader.load() # 返回 List[Document]
print(f"加载文档总数：{len(docs)}")

步骤 3 Transform：文本预处理 + 切片

• 清洗无效字符、换行；
• 大文档切块（Chunk），避免超长文本 Embedding 丢失语义，是 RAG 关键优化点

python

运行

# 递归文本分割器（工业最常用）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=120,    # 单块字符长度
    chunk_overlap=20   # 块之间重叠字符，防止上下文断裂
)
split_docs = text_splitter.split_documents(docs)
print(f"切割后文本块数量：{len(split_docs)}")

步骤 4 Embed：文本向量化

调用 Embedding 模型，自然语言文本 → 多维浮点向量

python

运行

# 内部由OpenAIEmbedding自动完成文本→向量转换，无需手动计算
# 向量格式：[0.12,0.34,-0.56 ...]

步骤 5 Store：向量 + 原文存入向量库

向量 + 文档元数据持久化存入 Chroma 向量数据库（本地文件型向量库）

python

运行

# 向量落地存储
vector_db = Chroma.from_documents(
    documents=split_docs,
    embedding=embedding,
    persist_directory="./chroma_company_db" # 持久化文件夹
)
# 生成Retriever实例（后续检索核心对象）
retriever = vector_db.as_retriever(search_kwargs={"k":2}) # 每次召回Top2相似文档

前 5 步完成离线建库，后续无需重复执行，服务重启直接加载向量库

步骤 6 Retrieve：在线检索 + RAG 问答（两种实现：固定 RAG / Agent-RAG）

方式 A：LCEL 固定 RAG（L3，强制每次必检索，企业标准知识库）

python

运行

# 文档拼接函数
def format_docs(docs):
    return "\n\n".join([d.page_content for d in docs])

# 构建RAG链路
rag_prompt = PromptTemplate.from_template("""
请严格依据下面参考资料回答用户问题，禁止编造内容：
【参考资料】
{context}
用户问题：{question}
""")

fixed_rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
)

# 测试提问
res = fixed_rag_chain.invoke("北京出差住宿报销多少钱？")
print("固定RAG应答：",res.content)

方式 B：Agent-RAG（L4，LLM 自主按需检索，常识不查库）

python

运行

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools.retriever import create_retriever_tool

# Retriever包装为知识库工具
kb_tool = create_retriever_tool(retriever, name="企业知识库", description="查询公司差旅报销制度")
tools = [kb_tool]

agent_prompt = PromptTemplate.from_template("你是企业行政助手 {input}")
agent = create_tool_calling_agent(llm, tools, agent_prompt)
agent_exe = AgentExecutor(agent=agent, tools=tools)

# 测试：常识不检索，业务问题自动查RAG
print(agent_exe.invoke({"input":"1+1等于几？"})["output"])
print(agent_exe.invoke({"input":"试用期员工有没有出差补贴？"})["output"])

---

拓展：后续加载已建好的向量库（不用重复前 5 步）

python

运行

# 项目重启直接读取存量向量库
db_load = Chroma(persist_directory="./chroma_company_db", embedding_function=embedding)
retriever_new = db_load.as_retriever(k=2)

全链路优化补充💡

1. Transform 优化：ParentDocumentRetriever 大小双切片；
2. Retrieve 优化：MultiQueryRetriever（问题扩写）、EnsembleRetriever（向量 + BM25 混合召回）、ContextualCompressionRetriever（结果过滤精简）；
3. 部署架构：向量库独立 MCP-Server，Agent 远程调用 RAG（分布式知识库）