系列导读:本文是「智答」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 篇文章
还没有评论,来抢沙发吧!