前四篇分别处理了模型、Prompt、Chain 和输出。RAG 把这些能力落到“基于文档回答问题”的真实场景：先把知识离线写入向量库，再在用户提问时检索相关片段，最后要求模型严格依据上下文作答。

阅读说明

本文基于本地 LangChain 学习代码与课程笔记整理。示例中的模型、文件路径和参数需要按实际环境调整；涉及 API Key 时只使用环境变量或占位符。

---

1 学习目标

• 理解RAG概念和作用

• 理解RAG工作原理

• 用 LangChain 构建简单RAG系统

2 RAG相关介绍

2.1 RAG概念

是什么：RAG是一种在大模型回答前先检索外部知识再生成答案的技术。

通⽤的基础⼤模型存在一些问题：

• 幻觉问题，LLM有时会⽣成看似合理但实际错误的信息

• LLM的知识不是实时的，模型训练好后不自动更新知识，导致部分信息滞后

• LLM领域知识是缺乏的，大模型的知识来源于训练数据，这些数据主要来自公开的互联网和开源数据集，无法覆盖特定领域或高度专业化的内部知识

为解决或缓解上述问题，2020年Facebook发表了一篇论文——《Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks》，提出了RAG（Retrieval Augmented Generation，检索增强生成）技术。

RAG是一种将大规模语言模型（LLM）与外部知识源的检索相结合，以改进问答能力的工程框架。 它使用来自私有或专有数据源的信息来辅助文本生成，从而弥补LLM的局限性，特别是在解决幻觉问题和提升时效性方面。

对比四种主流方案：

维度	Prompt	RAG	微调	Agent
核心原理	设计提示引导模型	检索知识 + 生成	数据训练改参数	LLM + 工具 + 多步决策
是否新增知识	❌	✅	✅	依赖外部工具
是否实时数据	❌	✅	❌	✅
是否改模型	❌	❌	✅	❌
开发成本	低	中	高	很高
推理成本	中	中	低	高
典型场景	文案/问答	知识库/客服	分类/垂直领域	自动化任务
核心优点	快速低成本	可更新知识	稳定可控	可执行复杂任务
当前定位	基础能力	🔥主流	辅助增强	🔥前沿

2.2 RAG作用

克服LLM“幻觉”问题：LLM在生成文本时有时会“一本正经地胡说八道”，即生成听起来合理但实际上不准确或捏造的信息，这被称为“幻觉”。RAG通过提供外部事实依据，显著减少了这种幻觉现象，让LLM的输出更具事实性和可靠性。

获取最新信息：LLM的训练数据通常是静态的，这意味着它们无法获取到训练截止日期之后发生的事件或更新的信息。RAG允许LLM连接到实时或定期更新的外部数据源（如新闻、数据库、内部文档等），从而提供最新、最及时的答案。

领域特定知识增强：对于特定行业或企业内部的知识，LLM的通用训练数据往往不足。RAG能够将LLM与企业内部的知识库文档或特定领域的数据连接起来，使LLM能够回答高度专业化的问题，并提供更符合上下文的答案。

成本低于模型微调：传统上，为了让LLM适应特定任务或数据，需要进行昂贵的微调（Fine-tuning）。RAG提供了一种更经济高效的替代方案，它无需修改LLM的底层参数，只需更新外部知识库即可，大大降低了维护和更新模型的成本。

提高答案的可解释性和溯源性：RAG可以引用其获取信息的来源，这意味着用户可以查看LLM答案所依据的原始文档或数据，增强了答案的透明度和用户信任度。

RAG通过将检索和生成相结合，既保留了传统检索问答的可靠性，又获得了LLM的灵活性和自然表达能力。它能让AI始终基于最新的、可信的知识来回答问题，同时保持对话的流畅自然。

传统检索式问答 (Retrieval QA)：

• ✅ 可靠性高：答案直接来自知识库，有明确的来源

• ✅ 知识可更新：添加新文档即可更新知识

• ❌ 灵活性差：只能返回知识库中已有的内容

• ❌ 表达生硬：难以用自然语言组织答案

纯LLM问答：

• ✅ 表达自然：能用流畅的语言组织答案

• ✅ 灵活理解：可以理解各种表达方式的问题

• ❌ 知识固化：知识仅限于训练数据，无法及时更新

• ❌ 可靠性差：容易产生幻觉，难以验证答案准确性

RAG方案：

• ✅ 可靠且可溯源：答案基于检索到的具体文档

• ✅ 知识可更新：可以持续添加新的知识

• ✅ 表达自然：利用LLM的语言能力组织答案

• ✅ 灵活理解：能理解各种形式的问题

• ✅ 成本可控：主要消耗在必要的API调用上

RAG的典型应用场景：

• 企业知识库问答：帮助企业构建对内员工知识库或对外客户问答系统。

• 法律法规、论文等参考场景：需要给出权威来源或证据的回答。

• 任何需要"带有引用信息"的回答场景。

2.3 RAG 的工作原理

1 工作流程图解

2 RAG标准流程

RAG 标准流程由索引（Indexing）、检索（Retriever）和生成（Generation）三个核心阶段组成。

• 索引阶段，通过处理多种来源多种格式的文档提取其中文本，将其切分为标准长度的文本块（chunk），并进行嵌入向量化（embedding），向量存储在向量数据库（vector database）中。

  • 加载文件

  • 内容提取

  • 文本分割 ，形成chunk

  • 文本向量化

  • 将向量存储到向量数据库

    常见的向量数据库如下：

    

• 检索阶段，用户输入的查询（query）被转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的文本块。

  • query向量化

  • 在文本向量中匹配出与问句向量相似的top_k个

• 生成阶段，检索到的相关文本与原始查询共同构成提示词（Prompt），输入大语言模型（LLM），生成精确且具备上下文关联的回答。

  • 匹配出的文本作为上下文和问题一起添加到prompt中

  • 提交给LLM生成答案

3 RAG问答机器人

3.1 项目背景

物流行业的客服场景。客户在查询物流信息时，常常会提出关于运输状态、配送时效、包裹轨迹和派送范围等问题。随着业务量的增长，传统人工客服难以做到实时、统一地答复。为此，XX公司需要一个RAG问答机器人，实现：基于物流信息文档与运单数据构建知识库，并利用RAG技术搭建智能物流查询系统，以自动解答客户常见问题，减轻客服压力，提升物流信息服务的准确性和响应速度。

3.2 项目思路

• 离线部分

  • 本地知识文件加载，读取

  • 文本切分

  • 向量化

  • 存入向量库

• 在线部分

  • query 向量化

  • 在文本向量中匹配出与问句向量相似的top_k个

  • 匹配出的文本作为上下文和问题一起添加到prompt中

  • 提交给LLM生成答案

3.3 项目代码

项目结构：

1 构建向量数据库

• 目的：读取文本，切分，向量化，存入向量数据库，构建检索器

"""
演示
    RAG系统的离线部分：由知识库生成向量数据库。
工作流程：
    知识库 -> 文档加载 -> 文本分割 -> 嵌入 -> 向量数据库
"""
from langchain_community.document_loaders import PyMuPDFLoader
# from langchain_unstructured import UnstructuredLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import FAISS

# 1.定义函数，生成向量数据库
def get_vector_db():
    # 1.加载文档
    # loader = UnstructuredLoader("物流信息.txt")
    loader = PyMuPDFLoader("物流信息.pdf")
    data = loader.load()
    print(f'data->{data}')
    # 2.文本分割
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=128,
        chunk_overlap=4)
    chunks = text_splitter.split_documents(data)
    # 3.创建Embedding模型
    embed_model = OllamaEmbeddings(model="mxbai-embed-large")

    # 4.生成向量数据库
    vector_db = FAISS.from_documents(chunks, embed_model)
    vector_db.save_local("./faiss/logistics")
    print(f'向量数据库保存成功')
    return chunks

# 主程序
if __name__ == '__main__':
    get_vector_db()

e:\conda_envs\langchain_project\Lib\site-packages\tqdm\auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html
  from .autonotebook import tqdm as notebook_tqdm


data->[Document(metadata={'producer': 'macOS 版本10.16（版号21E230） Quartz PDFContext', 'creator': 'typora', 'creationdate': "D:20240604062031Z00'00'", 'source': '物流信息.pdf', 'file_path': '物流信息.pdf', 'total_pages': 1, 'format': 'PDF 1.3', 'title': '物流信息', 'author': '', 'subject': '', 'keywords': '', 'moddate': "D:20240604062031Z00'00'", 'trapped': '', 'modDate': "D:20240604062031Z00'00'", 'creationDate': "D:20240604062031Z00'00'", 'page': 0}, page_content='物流公司：速达物流 公司总部：北京市 业务范围：国际快递、仓储管理\n货物追踪：\n货物编号：ABC123456\n发货⽇期：2023-01-15\n当前位置：上海分拨中⼼\n预计到达⽇期：2023-01-20\n运输⽅式：\n运输公司：快运通\n运输⽅式：陆运\n出发地：⼴州\n⽬的地：重庆\n预计运输时间：3天\n仓储信息：\n仓库名称：东⽅仓储中⼼\n仓库位置：深圳市\n存储货物类型：电⼦产品\n存储条件：常温仓储\n当前库存量：1000件')]
向量数据库保存成功

---

2 构建RAG主逻辑

• 目的：定义prompt ，并构建rag检索问答逻辑

"""
演示
    一个简单的RAG问答系统-物流信息查询助手。
工作流程：
    1.离线部分：由知识库生成向量数据库。
    2.在线部分：用户输入问题，系统从向量数据库中检索相关文档，构造提示词，并结合大模型进行回答。
"""
import time
from langchain_core.prompts import PromptTemplate
from langchain_ollama import OllamaLLM
from langchain_ollama import OllamaEmbeddings
from langchain_community.vectorstores import FAISS

# 1.加载向量数据库
embed_model = OllamaEmbeddings(model="mxbai-embed-large")
vector_db = FAISS.load_local(
    "./faiss/logistics",
    embed_model,
    allow_dangerous_deserialization=True
)
start_time = time.time()

# 2.定义函数，将文档转为文本
def get_related_docs(related_docs):
    """
将相关文档列表转换为文本字符串，供提示词使用。
    """
    related_content = []
    for doc in related_docs:
        related_content.append(doc.page_content.replace("\n\n", "\n"))
    return "\n".join(related_content)

# 3.定义函数，构造提示词
def define_prompt():
    """
构造提示词，包含用户问题和相关文档内容。
    """
    question = '我的快递出发地是哪？预计几天的时间到达？'
    # 1.从向量数据库中检索相关文档
    docs = vector_db.similarity_search(question, k=2)
    # 2.将相关文档转换为文本
    related_content = get_related_docs(docs)
    # 3.构造提示词
    prompt_template = """
    基于已知信息，简洁专业的回答用户问题，不允许在答案中添加编造成分。
    已知信息为：
    {related_content}。
    用户问题为：
    {question}。
    """
    prompt = PromptTemplate(
        input_variables=["related_content", "question"],
        template=prompt_template,
    )
    prompt = prompt.format(related_content=related_content, question=question)
    return prompt

# 4.定义函数，调用大模型进行回答
def get_answer():
    model = OllamaLLM(model="qwen3.5:4b")
    prompt = define_prompt()
    print(f"构造的提示词: {prompt}")
    result = model.invoke(prompt)
    return result

# 主程序
if __name__ == '__main__':
    result = get_answer()
    print(f"答案: {result}")
    end_time = time.time()
    print(f"总耗时: {end_time - start_time:.4f}s")

构造的提示词: 
    基于已知信息，简洁专业的回答用户问题，不允许在答案中添加编造成分。
    已知信息为：
    出发地：⼴州
⽬的地：重庆
预计运输时间：3天
仓储信息：
仓库名称：东⽅仓储中⼼
仓库位置：深圳市
存储货物类型：电⼦产品
存储条件：常温仓储
当前库存量：1000件
物流公司：速达物流 公司总部：北京市 业务范围：国际快递、仓储管理
货物追踪：
货物编号：ABC123456
发货⽇期：2023-01-15
当前位置：上海分拨中⼼
预计到达⽇期：2023-01-20
运输⽅式：
运输公司：快运通
运输⽅式：陆运。
    用户问题为：
    我的快递出发地是哪？预计几天的时间到达？。
答案: 出发地是广州，预计3天到达。
总耗时: 33.8501s

3 构建RAG前后端交互

• 目的：通过Gradio来实现前后端交互。

# Logistics_Assistant_Web.py
import gradio as gr
import time
# from get_vector_ollama import *
from langchain_core.prompts import PromptTemplate
from langchain_ollama import OllamaLLM
import os

# 全局变量存储向量数据库
current_db = None


def create_vector_db_from_file(file_path):
    """
    从上传的文件创建向量数据库
    支持 txt, pdf, docx 格式
    """
    try:
        if file_path.endswith('.txt'):
            from langchain_community.document_loaders import TextLoader
            loader = TextLoader(file_path, encoding='utf-8')
        elif file_path.endswith('.pdf'):
            from langchain_community.document_loaders import PyPDFLoader
            loader = PyPDFLoader(file_path)
        elif file_path.endswith('.docx'):
            from langchain_community.document_loaders import Docx2txtLoader
            loader = Docx2txtLoader(file_path)
        else:
            raise ValueError(f"不支持的文件格式: {file_path}")

        documents = loader.load()
        # 文本分割
        from langchain_text_splitters import RecursiveCharacterTextSplitter
        text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
        texts = text_splitter.split_documents(documents)

        # 创建向量数据库
        embeddings = OllamaEmbeddings(model="mxbai-embed-large", temperature=0)
        db = FAISS.from_documents(texts, embeddings)
        return db
    except Exception as e:
        raise e


def load_default_db():
    """
    加载默认的向量数据库
    """
    global current_db
    try:
        embeddings = OllamaEmbeddings(model="mxbai-embed-large", temperature=0)
        current_db = FAISS.load_local("faiss/wuliu", embeddings, allow_dangerous_deserialization=True)
    except Exception as e:
        print(f"加载默认向量数据库失败: {e}")
        current_db = None


def handle_file_upload(file_obj):
    """
    处理文件上传并创建向量数据库
    """
    global current_db
    try:
        if file_obj is None:
            return "未选择文件", None

        file_path = file_obj.name
        current_db = create_vector_db_from_file(file_path)
        return f"成功加载文件: {os.path.basename(file_path)}", current_db is not None
    except Exception as e:
        return f"处理文件时出错: {str(e)}", None


def get_related_content(related_docs):
    """
    提取相关文档内容
    """
    related_content = []
    for doc in related_docs:
        related_content.append(doc.page_content.replace("\n\n", "\n"))
    return "\n".join(related_content)


def answer_question(question):
    """
    回答用户提出的问题
    """
    global current_db
    start_time = time.time()

    try:
        # 如果没有加载向量数据库，则尝试加载默认数据库
        if current_db is None:
            embeddings = OllamaEmbeddings(model="mxbai-embed-large", temperature=0)
            current_db = FAISS.load_local("faiss/wuliu", embeddings, allow_dangerous_deserialization=True)

        # 搜索相关文档
        docs = current_db.similarity_search(question, k=2)
        related_content = get_related_content(docs)

        # 构建提示词模板
        PROMPT_TEMPLATE = """
        基于以下已知信息，简洁和专业的来回答用户的问题。不允许在答案中添加编造成分。
        已知内容:
        {context}
        问题:
        {question}"""

        prompt = PromptTemplate(
            input_variables=["context", "question"],
            template=PROMPT_TEMPLATE,
        )

        formatted_prompt = prompt.format(context=related_content, question=question)

        # 使用模型生成答案
        model = OllamaLLM(model="qwen3.5:4b")
        result = model.invoke(formatted_prompt)
        print(f'>>>答案：{result}')

        end_time = time.time()
        retrieved_content_output = related_content[:500] + "..." if len(related_content) > 500 else related_content
        # 返回答案和相关信息
        return result, f"{end_time - start_time:.4f}s", retrieved_content_output
    except Exception as e:
        end_time = time.time()
        return f"处理过程中出现错误: {str(e)}", f"{end_time - start_time:.4f}s", ""


def create_gradio_interface():
    """
    创建Gradio界面
    """
    with gr.Blocks(title="智能物流助手") as demo:
        gr.Markdown("# 🚚 智能物流助手")
        gr.Markdown("基于向量数据库的物流信息查询系统")

        with gr.Row():
            with gr.Column():
                # 文件上传组件
                file_input = gr.File(
                    label="上传知识库文件",
                    file_types=['.txt', '.pdf', '.docx'],
                    file_count="single"
                )
                upload_btn = gr.Button("加载文件到知识库")
                upload_status = gr.Textbox(label="上传状态", interactive=False)

                question_input = gr.Textbox(
                    label="请输入您的物流问题",
                    placeholder="例如：我的快递出发地是哪？预计几天的时间到达？",
                    lines=3
                )

                submit_btn = gr.Button("获取答案", variant="primary")

                # 示例问题
                gr.Examples(
                    examples=[
                        "我的快递出发地是哪？",
                        "预计几天的时间到达？",
                        "我的订单状态如何？",
                        "物流信息是什么？"
                    ],
                    inputs=question_input
                )

            with gr.Column():
                processing_time_output = gr.Textbox(
                    label="处理时间",
                    interactive=False
                )
                retrieved_content_output = gr.Textbox(
                    label="检索到的相关内容",
                    lines=10,
                    interactive=False
                )
                answer_output = gr.Textbox(
                    label="AI回答",
                    lines=8,
                    interactive=False
                )

        # 事件绑定
        upload_btn.click(
            fn=handle_file_upload,
            inputs=file_input,
            outputs=[upload_status, gr.State()]
        )

        submit_btn.click(
            fn=answer_question,
            inputs=question_input,
            outputs=[answer_output, processing_time_output, retrieved_content_output]
        )

    return demo


if __name__ == "__main__":
    # 初始化时加载默认数据库
    load_default_db()
    interface = create_gradio_interface()
    interface.launch(
        server_name="127.0.0.1",
        server_port=7860,
        share=False,  # 设为True可创建公共链接
        debug=True
    )

---

本篇小结

RAG 的重点不是单纯接上向量数据库，而是检索质量、上下文约束和可验证性。后续可以继续扩展引用溯源、Rerank、评测集、权限控制和增量索引。