系列导读：本文是「智答」RAG 智能问答系统开发系列的第一篇。本系列面向有 Python 基础、希望系统入门 RAG 工程开发的程序员，共 6 篇文章，配套一个完整的实践项目——「技术文档智能问答助手」。跟完整个系列，你将得到一个支持多格式文档上传、多轮对话、混合检索的生产级 RAG 系统。
在讨论 RAG 之前，我们先来认真审视一下它所要解决的问题——这不是套话，理解痛点是理解技术价值最有效的路径。
几乎每一个有一定规模的产品，都会有一个 FAQ（Frequently Asked Questions）页面或者对话机器人。这套系统的运作逻辑很朴素：人工整理高频问题，配上标准答案，用户来了就匹配规则。
这套方案在业务早期有效。但随着时间推移，问题就来了：
据一些公司的客服运营负责人反映，一个中等规模的 SaaS 产品，维护一份高质量的 FAQ 库，每月需要投入 2-3 名专职运营人员的精力。而且这还是"维护存量"，业务增长带来的新问题不断涌入，人永远在追着变化跑。
用户的问法是千变万化的。同一个问题，可以有几十种不同的表达：
"怎么导出数据？" "数据能下载吗？" "我想把报表存成 Excel" "export 功能在哪里" "能不能把这个表格发到我邮箱"
基于关键词匹配或者简单意图分类的传统系统，覆盖这些变体需要大量的同义词配置和人工标注。现实情况是，70% 的用户问题集中在 20% 的高频场景，但另外那 80% 的长尾问题，每一个都可能是用户流失的起点。
更麻烦的是，这些长尾问题往往是最需要帮助的用户在问的——他们卡住了，找不到答案，然后悄悄离开。
当自动化问答系统覆盖不到的时候，请求就会落到人工客服头上。人工客服有几个根本性的限制：
这三个困境叠加在一起，构成了传统问答系统的根本性矛盾：知识是静态的，但用户的问题是动态的。
当大语言模型（LLM）出现之后，很多人的第一反应是：直接用 ChatGPT 不就解决了吗？
这个想法方向是对的，但实际落地会遇到几个硬问题：
问题一：LLM 不知道你公司的私有知识
GPT-4 或者 Qwen 训练的是公开互联网上的数据。你公司的内部文档、产品手册、业务规则，它根本没见过。你不能指望它回答"我们 Q3 的促销活动有哪些优惠"这类问题。
问题二：知识截止日期
LLM 的训练数据有截止日期。昨天刚更新的政策，它不知道；上周发布的新版本功能，它也不知道。
问题三：幻觉（Hallucination）
这是 LLM 最被诟病的问题。当模型不知道答案的时候，它不会说"我不知道"，而是会非常流畅地编造一个听起来合理的答案。在问答场景里，这是灾难性的——用户可能据此做出错误决策。
问题四：成本
把所有知识都塞进 prompt（上下文）是一种方案，但这意味着每次对话都要传输大量文本，token 费用和延迟都是问题。
RAG 是 Retrieval-Augmented Generation 的缩写，中文一般译为"检索增强生成"。
它的核心思想用一句话概括：先检索，再生成。
用一个类比来理解：你是一名大学生，现在要回答一道开卷考试题。你有两种策略：
策略 B 的优势显而易见：答案有据可查、不依赖记忆的准确性、可以处理你从没学过的内容。
这就是 RAG 的本质。它不是让 LLM"记住"所有知识，而是在需要回答时，先从知识库里检索出最相关的片段，再把这些片段作为上下文交给 LLM，让 LLM 基于这些真实资料生成回答。
一个完整的 RAG 系统由两条链路组成：离线链路（知识入库）和在线链路（问答服务）。
╔══════════════════════════════════════════════════════════════╗ ║ 离线链路（知识入库） ║ ╠══════════════════════════════════════════════════════════════╣ ║ ║ ║ 原始文档 ║ ║ (PDF/MD/TXT/PPT) ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 文档加载 │ ← 各类 Loader（PDF/Markdown/TXT...） ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 文本清洗 │ ← 去噪、去重、实体消歧 ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 文档切分 │ ← 父子块切分策略 ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 向量化 │ ← BGE-M3（稠密 + 稀疏双向量） ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 向量存储 │ ← Milvus 向量数据库 ║ ║ └─────────────┘ ║ ║ ║ ╚══════════════════════════════════════════════════════════════╝ ╔══════════════════════════════════════════════════════════════╗ ║ 在线链路（问答服务） ║ ╠══════════════════════════════════════════════════════════════╣ ║ ║ ║ 用户输入问题 ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 语义缓存 │ ← Redis Semantic Cache（热点直接返回） ║ ║ └──────┬──────┘ ║ ║ │ （未命中缓存） ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 意图识别 │ ← BERT 分类器 ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ┌────┴────┐ ║ ║ ▼ ▼ ║ ║ RAG模式 直聊模式 ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 混合检索 │ ← 稠密向量 + 稀疏向量 Hybrid Search ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ Reranker │ ← BGE-Reranker 精排 ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ Prompt组装 │ ← 检索结果 + 对话历史 → 动态 Prompt ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ LLM 生成 │ ← Qwen2.5（流式 SSE 输出） ║ ║ └──────┬──────┘ ║ ║ │ ║ ║ ▼ ║ ║ ┌─────────────┐ ║ ║ │ 结果返回 │ ← FastAPI SSE 接口 ║ ║ └─────────────┘ ║ ║ ║ ╚══════════════════════════════════════════════════════════════╝
离线链路是系统的"地基"，负责把原始的非结构化文档，加工成可以高效检索的向量索引。这条链路的质量直接决定了系统的知识储备上限——你装进去的是什么品质的知识，系统能检索出来的就是什么品质的知识，这个原则在工程界被称为"GIGO"（Garbage In, Garbage Out）。
离线链路通常不是实时运行的，而是批量处理的：当有新文档需要入库时，触发一次离线处理流程。
在线链路是用户实际交互的部分，它的核心目标是两个字：又快又准。
一个容易混淆的点是：检索用的向量和生成用的 LLM，它们消费的东西是不同的。
这个分工意味着你可以独立替换任何一个模块——换一个更好的 Embedding 模型，或者换一个更强的生成模型，而不需要推倒重来。
在开始写代码之前，有必要把每个技术选型背后的理由说清楚。这不是凑字数，是为了让你在后面的开发中遇到问题时，知道"为什么是这个，不是那个"。
LangChain 是目前 RAG 和 LLM 应用开发最主流的框架。1.0 版本（也就是基于 langchain-core 的 LCEL 新架构）相比早期版本有几个重要改进：
langchain-core
LCEL（LangChain Expression Language）
LCEL 是 LangChain 1.0 的核心设计。它用管道符 | 把各个组件串联起来，使链路的定义变得极其简洁：
|
chain = prompt | llm | output_parser
这种写法不只是语法糖，它背后有完整的流式、批处理、异步支持，不需要额外配置。
Runnable 接口统一
所有组件（Prompt、LLM、Retriever、OutputParser）都实现了同一个 Runnable 接口，意味着它们可以任意组合，行为一致。
Runnable
更好的流式支持
LCEL 链路原生支持 astream()，配合 FastAPI 的 StreamingResponse 可以很方便地实现 SSE 流式输出。
astream()
StreamingResponse
向量数据库是 RAG 系统的"检索引擎"，市面上有不少选择：FAISS、Chroma、Weaviate、Qdrant、Milvus 等。
我们选择 Milvus 的理由：
对于本系列的实践项目，我们会用 Milvus Lite（嵌入式版本，无需独立部署），降低入门门槛。生产环境可以无缝切换到完整版 Milvus。
BGE-M3 是智源研究院（BAAI）开源的多功能 Embedding 模型，它最大的特点是一个模型同时输出三种向量：
混合检索就是同时使用稠密 + 稀疏向量，兼顾语义理解和关键词匹配，在绝大多数场景下效果优于单一向量。
Reranker 是一种交叉编码器（Cross-Encoder）模型。它和 Embedding 模型（双塔编码器）的区别在于：
这就是"粗排 + 精排"两阶段策略的意义：先用快速的向量检索海选，再用精准的 Reranker 细选。
Redis 在这里承担的角色不是普通的 Key-Value 缓存，而是语义缓存（Semantic Cache）。
普通缓存：完全相同的问题才能命中缓存。 语义缓存：语义相近的问题就能命中缓存。
例如"如何导出数据"和"数据怎么下载"在语义上是相同的问题，语义缓存可以直接返回之前的答案，不需要重新检索和生成。LangChain 提供了开箱即用的 RedisSemanticCache。
RedisSemanticCache
FastAPI 是目前 Python 生态中最适合构建 AI 服务的 Web 框架：异步原生、性能优秀、自动生成 API 文档。
Qwen2.5 是阿里通义团队开源的大语言模型，中文能力强，有多个尺寸可选（0.5B 到 72B），可以本地部署也可以调用 API，适合国内开发者。
本系列配套的实践项目叫做「TechBot：技术文档智能问答助手」。
跟完本系列后，项目的最终目录结构如下：
techbot/ ├── README.md ├── requirements.txt ├── .env # 环境变量（API Keys 等） │ ├── config/ │ └── settings.py # 全局配置 │ ├── data/ │ ├── raw/ # 原始文档存放目录 │ └── processed/ # 处理后的中间文件 │ ├── core/ │ ├── __init__.py │ ├── loader.py # 第2篇：文档加载与清洗 │ ├── splitter.py # 第2篇：父子文档切分 │ ├── embedder.py # 第3篇：BGE-M3 向量化 │ ├── vectorstore.py # 第3篇：Milvus 存储 │ ├── retriever.py # 第4篇：混合检索 + 重排序 │ ├── router.py # 第4篇：意图识别与路由 │ ├── chain.py # 第5篇：对话链路组装 │ └── cache.py # 第5篇：Redis 语义缓存 │ ├── api/ │ ├── __init__.py │ ├── main.py # 第5篇：FastAPI 入口 │ ├── routes/ │ │ ├── ingest.py # 文档入库接口 │ │ └── chat.py # 对话接口（SSE） │ └── schemas.py # 请求/响应数据模型 │ ├── evaluation/ │ └── ragas_eval.py # 第6篇：RAGAS 评估 │ └── scripts/ └── ingest_docs.py # 批量文档入库脚本
本文（第一篇）的实践目标是：
后续每篇文章会逐步替换 Demo 中的简化实现，换成生产级的方案。
venv
conda
# 创建项目目录 mkdir techbot && cd techbot # 创建虚拟环境（使用 venv） python -m venv .venv # 激活虚拟环境 # Linux/macOS: source .venv/bin/activate # Windows: .venv\Scripts\activate # 验证 Python 版本 python --version # 应该显示 3.10 或以上
创建 requirements.txt：
requirements.txt
# LangChain 核心 langchain>=0.3.0 langchain-community>=0.3.0 langchain-core>=0.3.0 # LLM（使用 OpenAI 兼容接口，Qwen/DeepSeek 均支持） langchain-openai>=0.2.0 openai>=1.0.0 # 向量数据库 pymilvus>=2.4.0 # Embedding 模型 FlagEmbedding>=1.2.0 # 文档加载 pypdf>=4.0.0 unstructured>=0.14.0 # Web 框架 fastapi>=0.111.0 uvicorn>=0.30.0 python-multipart>=0.0.9 # 缓存 redis>=5.0.0 # 数据库 sqlalchemy>=2.0.0 # 工具 python-dotenv>=1.0.0 pydantic>=2.0.0 tiktoken>=0.7.0 # 评估（第6篇使用） ragas>=0.1.0
安装依赖：
pip install -r requirements.txt
注意：unstructured 依赖较多，如果安装报错可以先跳过，本篇 Demo 不需要它。后续第 2 篇会详细处理各种文档格式的加载问题。
unstructured
在项目根目录创建 .env 文件：
.env
# .env # ======= LLM 配置 ======= # 方案一：使用 DashScope（阿里云，Qwen 官方接口） DASHSCOPE_API_KEY=your_dashscope_api_key_here LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 LLM_MODEL=qwen-plus # 方案二：使用 DeepSeek（价格便宜，效果好） # DEEPSEEK_API_KEY=your_deepseek_api_key_here # LLM_BASE_URL=https://api.deepseek.com/v1 # LLM_MODEL=deepseek-chat # 方案三：本地 Ollama（完全免费，需要本地 GPU） # LLM_BASE_URL=http://localhost:11434/v1 # LLM_MODEL=qwen2.5:7b # ======= Milvus 配置 ======= # Lite 模式使用本地文件，无需额外部署 MILVUS_URI=./milvus_lite.db # ======= Redis 配置（第5篇使用）======= REDIS_URL=redis://localhost:6379 # ======= 项目配置 ======= COLLECTION_NAME=techbot_docs
创建 config/settings.py：
config/settings.py
# config/settings.py import os from dotenv import load_dotenv load_dotenv() class Settings: # LLM LLM_BASE_URL: str = os.getenv("LLM_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1") LLM_API_KEY: str = os.getenv("DASHSCOPE_API_KEY", os.getenv("DEEPSEEK_API_KEY", "")) LLM_MODEL: str = os.getenv("LLM_MODEL", "qwen-plus") # Milvus MILVUS_URI: str = os.getenv("MILVUS_URI", "./milvus_lite.db") COLLECTION_NAME: str = os.getenv("COLLECTION_NAME", "techbot_docs") # 检索参数 TOP_K_RETRIEVAL: int = 20 # 初步召回数量 TOP_K_RERANK: int = 5 # 精排后保留数量 CHUNK_SIZE: int = 512 # 子文档切片大小 CHUNK_OVERLAP: int = 50 # 切片重叠大小 PARENT_CHUNK_SIZE: int = 2048 # 父文档块大小 settings = Settings()
运行下面的脚本，验证核心依赖是否正常安装：
# scripts/check_env.py """ 环境检查脚本 运行方式：python scripts/check_env.py """ def check_import(module_name: str, package_name: str = None): """尝试导入模块，返回是否成功""" try: __import__(module_name) print(f" ✅ {package_name or module_name}") return True except ImportError as e: print(f" ❌ {package_name or module_name}: {e}") return False print("=" * 50) print("TechBot 环境检查") print("=" * 50) print("\n📦 核心依赖：") check_import("langchain", "langchain") check_import("langchain_core", "langchain-core") check_import("langchain_community", "langchain-community") check_import("langchain_openai", "langchain-openai") print("\n📦 向量存储：") check_import("pymilvus", "pymilvus") print("\n📦 文档处理：") check_import("pypdf", "pypdf") print("\n📦 Web 框架：") check_import("fastapi", "fastapi") check_import("uvicorn", "uvicorn") print("\n📦 工具库：") check_import("dotenv", "python-dotenv") check_import("pydantic", "pydantic") print("\n🔑 环境变量：") import os from dotenv import load_dotenv load_dotenv() api_key = os.getenv("DASHSCOPE_API_KEY") or os.getenv("DEEPSEEK_API_KEY") if api_key: print(f" ✅ API Key 已配置（{api_key[:8]}...）") else: print(" ⚠️ 未检测到 API Key，请检查 .env 文件") milvus_uri = os.getenv("MILVUS_URI", "./milvus_lite.db") print(f" ✅ Milvus URI: {milvus_uri}") print("\n" + "=" * 50) print("检查完成！如有 ❌ 项目，请运行 pip install -r requirements.txt") print("=" * 50)
运行：
python scripts/check_env.py
期望输出：
================================================== TechBot 环境检查 ================================================== 📦 核心依赖： ✅ langchain ✅ langchain-core ✅ langchain-community ✅ langchain-openai 📦 向量存储： ✅ pymilvus 📦 文档处理： ✅ pypdf 📦 Web 框架： ✅ fastapi ✅ uvicorn 📦 工具库： ✅ python-dotenv ✅ pydantic 🔑 环境变量： ✅ API Key 已配置（sk-xxxxxx...） ✅ Milvus URI: ./milvus_lite.db ================================================== 检查完成！如有 ❌ 项目，请运行 pip install -r requirements.txt ==================================================
在正式进入工程细节之前，先用最少的代码跑通 RAG 的完整链路，建立直观感受。这个 Demo 使用了大量简化假设，后续每一篇文章会逐步替换成生产级实现。
在项目根目录创建 data/raw/ 目录，然后创建一个测试文档：
data/raw/
mkdir -p data/raw
创建 data/raw/python_faq.md，内容如下：
data/raw/python_faq.md
# Python 常见问题解答 ## 1. 如何安装第三方包？ 使用 pip 命令安装： ```bash pip install 包名
例如安装 requests：pip install requests 建议在虚拟环境中安装，避免包版本冲突。
pip install requests
虚拟环境是 Python 的独立运行环境，每个项目可以有自己独立的依赖包版本。 创建虚拟环境：python -m venv myenv 激活虚拟环境（Linux/Mac）：source myenv/bin/activate 激活虚拟环境（Windows）：myenv\Scripts\activate
python -m venv myenv
source myenv/bin/activate
myenv\Scripts\activate
列表（list）是可变的，可以增删改元素：[1, 2, 3] 元组（tuple）是不可变的，创建后不能修改：(1, 2, 3) 元组通常用于表示固定数据，比列表略快，可以作为字典的键。
[1, 2, 3]
(1, 2, 3)
读取文件推荐使用 with 语句：
with open('file.txt', 'r', encoding='utf-8') as f: content = f.read()
写入文件：
with open('file.txt', 'w', encoding='utf-8') as f: f.write('内容')
GIL（全局解释器锁）是 CPython 的一个机制，同一时刻只允许一个线程执行 Python 字节码。 这意味着多线程在 CPU 密集型任务中无法真正并行。 解决方案：CPU 密集型任务使用多进程（multiprocessing），IO 密集型任务使用多线程或异步（asyncio）。
### 7.2 最简 RAG Demo 创建 `scripts/simple_rag_demo.py`： ```python """ 最简 RAG Demo 演示 RAG 的完整链路：文档加载 → 向量化 → 存储 → 检索 → 生成 运行方式：python scripts/simple_rag_demo.py 注意：这是高度简化的演示版本，使用了： - 简单的文本切分（后续第2篇替换为父子切分） - Milvus Lite 本地存储（生产环境使用完整版 Milvus） - 基础的相似度检索（后续第4篇替换为混合检索 + Reranker） - 简单的 Prompt 模板（后续第5篇替换为动态模板 + 对话历史） """ import os from dotenv import load_dotenv # 加载环境变量 load_dotenv() # ===================================================== # Step 1: 文档加载与切分（简化版） # ===================================================== print("\n📂 Step 1: 加载文档...") from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 加载 Markdown 文件（TextLoader 可以读取纯文本和 Markdown） loader = TextLoader("data/raw/python_faq.md", encoding="utf-8") documents = loader.load() print(f" 加载完成：{len(documents)} 个文档对象") print(f" 文档总字符数：{sum(len(d.page_content) for d in documents)}") # 切分文档 text_splitter = RecursiveCharacterTextSplitter( chunk_size=300, # 每个切片最多 300 字符 chunk_overlap=50, # 相邻切片重叠 50 字符（防止语义断裂） separators=["\n\n", "\n", "。", "！", "？", " ", ""], ) chunks = text_splitter.split_documents(documents) print(f" 切分完成：{len(chunks)} 个文本块") for i, chunk in enumerate(chunks[:3]): # 预览前3个块 print(f" --- 块 {i+1} (前80字符) ---") print(f" {chunk.page_content[:80].replace(chr(10), ' ')}...") # ===================================================== # Step 2: 向量化与存储（使用 Milvus Lite） # ===================================================== print("\n🗄️ Step 2: 向量化并存入 Milvus...") from langchain_community.vectorstores import Milvus from langchain_openai import OpenAIEmbeddings # 使用 OpenAI 兼容接口的 Embedding # 注意：这里使用 text-embedding-v3 是 DashScope 的模型名 # 如果使用 DeepSeek，DeepSeek 目前不提供 Embedding，建议用 DashScope 或本地模型 embeddings = OpenAIEmbeddings( model="text-embedding-v3", # DashScope 的 embedding 模型 openai_api_key=os.getenv("DASHSCOPE_API_KEY"), openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", dimensions=1024, ) # 存入 Milvus Lite（本地文件，无需启动服务） vectorstore = Milvus.from_documents( documents=chunks, embedding=embeddings, connection_args={"uri": "./demo_milvus.db"}, # Milvus Lite 使用本地文件 collection_name="python_faq_demo", drop_old=True, # 每次运行重新创建（演示用） ) print(f" 向量化完成，已存入 {len(chunks)} 个文档块") # ===================================================== # Step 3: 检索（简化版相似度搜索） # ===================================================== print("\n🔍 Step 3: 检索测试...") test_query = "Python 虚拟环境怎么用？" print(f" 查询问题：{test_query}") # 相似度搜索，返回 Top-3 最相关的文档块 retrieved_docs = vectorstore.similarity_search(test_query, k=3) print(f" 检索到 {len(retrieved_docs)} 个相关片段：") for i, doc in enumerate(retrieved_docs): print(f"\n --- 片段 {i+1} ---") print(f" {doc.page_content[:150].replace(chr(10), ' ')}...") # ===================================================== # Step 4: 生成回答（简化版 Prompt） # ===================================================== print("\n🤖 Step 4: 生成回答...") from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 初始化 LLM llm = ChatOpenAI( model=os.getenv("LLM_MODEL", "qwen-plus"), api_key=os.getenv("DASHSCOPE_API_KEY"), base_url=os.getenv("LLM_BASE_URL"), temperature=0.1, # 低温度，让回答更确定、更忠实于原文 ) # 定义 Prompt 模板 # 这里的关键是：明确告诉 LLM 只能基于提供的上下文回答 prompt = ChatPromptTemplate.from_template(""" 你是一个技术文档问答助手。请严格根据以下提供的上下文内容来回答用户的问题。 如果上下文中没有相关信息，请直接说"抱歉，我在文档中没有找到相关信息"，不要编造答案。 【参考上下文】 {context} 【用户问题】 {question} 【回答】 """) # 构建 LCEL 链（这就是 LangChain 1.0 的核心写法） chain = prompt | llm | StrOutputParser() # 把检索到的文档拼接成上下文字符串 context = "\n\n---\n\n".join([doc.page_content for doc in retrieved_docs]) # 执行链 answer = chain.invoke({ "context": context, "question": test_query, }) print(f"\n 问题：{test_query}") print(f"\n 回答：\n {answer}") # ===================================================== # Step 5: 端到端测试多个问题 # ===================================================== print("\n" + "=" * 60) print("🧪 端到端测试：多问题验证") print("=" * 60) test_questions = [ "列表和元组有什么区别？", "GIL 是什么，会影响多线程吗？", "Python 支持 Java 吗？", # 故意问一个文档里没有的问题，测试幻觉控制 ] for question in test_questions: print(f"\n❓ 问题：{question}") # 检索 docs = vectorstore.similarity_search(question, k=3) context = "\n\n---\n\n".join([doc.page_content for doc in docs]) # 生成 answer = chain.invoke({"context": context, "question": question}) print(f"💬 回答：{answer[:200]}...") # 只打印前200字
python scripts/simple_rag_demo.py
运行成功后，你应该能看到以下几个关键输出：
文档切分结果：观察切片数量和每个切片的内容，理解为什么切片大小的选择很重要——太小会导致语义不完整，太大会引入噪音。
检索结果：观察检索到的 3 个片段是否真的和问题相关。如果检索质量差，后面的生成再好也没用。
幻觉控制测试：最后一个问题"Python 支持 Java 吗"是文档里没有的内容，观察模型是否正确拒绝回答（说"文档中没有找到相关信息"），而不是自己编一个答案。
这个 Demo 非常简化，它有几个明显的局限：
这些局限不是 bug，而是从"能跑"到"跑得好"的工程进化路径。理解每一个改进点背后的原因，比直接给你一个完整项目更有价值。
在继续之前，再展示一个稍微进阶一点的写法——流式输出。这是后续第 5 篇要深入讲的内容，但先感受一下：
# scripts/stream_demo.py """ 流式输出演示 感受 SSE 流式生成的效果 """ import os import asyncio from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser load_dotenv() llm = ChatOpenAI( model=os.getenv("LLM_MODEL", "qwen-plus"), api_key=os.getenv("DASHSCOPE_API_KEY"), base_url=os.getenv("LLM_BASE_URL"), temperature=0.1, streaming=True, # 开启流式模式 ) prompt = ChatPromptTemplate.from_template( "请用中文简短回答：{question}" ) chain = prompt | llm | StrOutputParser() async def stream_answer(question: str): print(f"\n问题：{question}") print("回答：", end="", flush=True) # astream() 是 LCEL 的异步流式接口 async for chunk in chain.astream({"question": question}): print(chunk, end="", flush=True) # 逐 token 打印 print("\n") # 换行 async def main(): questions = [ "Python 的列表推导式是什么？", "什么情况下应该用 asyncio？", ] for q in questions: await stream_answer(q) if __name__ == "__main__": asyncio.run(main())
python scripts/stream_demo.py
观察输出：回答不是一次性出现的，而是像打字机一样逐字出现。这就是流式生成的效果，后续我们会把它封装成 SSE 接口供前端消费。
第1篇（本篇） ├── RAG 概念与架构 ├── 技术选型理解 └── 环境搭建 + 简单 Demo │ ▼ 第2篇：知识入库（离线链路上半段） ├── 多格式文档加载（PDF/MD/TXT） ├── 文本清洗管道 └── 父子文档切分策略 │ ▼ 第3篇：向量化与存储（离线链路下半段） ├── BGE-M3 稠密+稀疏双向量 ├── Milvus Schema 设计 └── 混合检索基础设施 │ ▼ 第4篇：检索优化（在线链路核心） ├── BERT 意图分类 + 路由 ├── Hybrid Search 实战 └── BGE-Reranker 二次精排 │ ▼ 第5篇：服务封装（在线链路收尾） ├── 多轮对话管理 ├── Redis 语义缓存 └── FastAPI + SSE 流式接口 │ ▼ 第6篇：评估与复盘 ├── RAGAS 四大指标 ├── 自动化评估流水线 └── 优化方向分析
先读后码：每篇文章的原理部分不要跳过。很多 RAG 工程坑的根源，是在不理解原理的情况下凭直觉堆代码。
动手跑通：每篇的代码都要自己打一遍，不要只是复制粘贴。在打代码的过程中遇到的报错，往往比文章本身更有学习价值。
循序渐进：每篇都会在上一篇的基础上迭代，不建议跳篇。如果某个概念没懂，先把 Demo 跑通，带着问题进入下一篇，有时候上下文能帮助理解。
理解改进点：每一个新引入的技术（父子切分、混合检索、Reranker……），都是为了解决一个具体问题。记住"这个技术解决了什么问题"比记住"这个技术怎么用"更重要。
在开始本系列之前，确保你具备以下基础：
不需要有深度学习或 NLP 的背景，遇到模型原理类的内容，本系列会用直觉性类比解释，不涉及数学推导。
本系列的代码使用了 OpenAI 兼容接口，这意味着你可以根据自己的情况选择任意支持该接口的服务：
qwen2.5:7b
三种方案只需要修改 .env 里的三个变量（API_KEY、BASE_URL、MODEL），代码本身完全一致。
API_KEY
BASE_URL
MODEL
本文我们做了三件事：
第一，厘清了 RAG 要解决的问题——传统 FAQ 系统在维护成本、覆盖率和响应速度上的根本性局限，以及纯 LLM 方案面临的私有知识、幻觉等挑战。RAG 的"先检索、再生成"思路是一个工程上简洁而有效的解答。
第二，建立了完整的 RAG 系统心智模型——离线链路负责把文档加工成可检索的向量索引，在线链路负责把用户问题转化为精准的检索+生成流程。理解这个两条链路的分工，是理解后续每一篇文章的基础。
第三，搭建了开发环境并跑通了第一个 RAG Demo——用最少的代码验证了"文档加载→切分→向量化→检索→生成"的完整流程，也直观看到了幻觉控制的效果，以及流式输出的体验。
下一篇，我们会深入离线链路的核心：如何处理真实世界的脏乱文档，以及为什么父子文档切分策略能显著提升检索质量。
本系列代码仓库：随系列更新，每篇文章对应一个 Git tag，你可以 checkout 到任意阶段的代码状态。 有问题？ 欢迎在评论区留言，或者直接提 Issue。RAG 工程的很多"坑"是环境和配置问题，多数可以快速解决。
本系列代码仓库：随系列更新，每篇文章对应一个 Git tag，你可以 checkout 到任意阶段的代码状态。
有问题？ 欢迎在评论区留言，或者直接提 Issue。RAG 工程的很多"坑"是环境和配置问题，多数可以快速解决。
下一篇：《知识入库：多格式文档的 ETL 与父子切分策略》
还没有评论，来抢沙发吧！
这个人很懒，什么都没写...
3 篇文章
Chuanyun
还没有评论，来抢沙发吧！