RAG Pipeline 从零搭建：架构设计、技术选型到生产部署

什么是 RAG？为什么它比”直接问 LLM”好用？

假设你问 ChatGPT：“我们公司 2025 年的销售策略是什么？”

ChatGPT 会告诉你：“我不知道，因为这是你们公司的内部信息。”

但如果我们把公司的销售文档先存入向量数据库，然后：

1. 把你的问题转换成向量
2. 在向量数据库中找到最相关的文档片段
3. 把文档片段和问题一起发给 LLM

ChatGPT 就能基于你的内部文档给出精准回答。这就是 RAG（Retrieval-Augmented Generation，检索增强生成）。

RAG Pipeline 全架构

用户问题
    │
    ▼
┌─────────────────┐
│  Query Embedding │  ← 用 Embedding 模型把问题转为向量
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Vector Search   │  ← 在向量数据库中检索 Top-K 相关片段
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Prompt Assembly │  ← 把检索结果 + 问题组装成 Prompt
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   LLM Generation │  ← LLM 基于上下文生成回答
└────────┬────────┘
         │
         ▼
     最终回答

第一步：文档解析与分块（Chunking）

为什么要分块？

LLM 的上下文窗口有限（即使是 128K 也只能装几十万字）。如果你把整本 500 页的手册塞进去：（1）超出上下文限制；（2）LLM 会在海量文本中迷失重点。

所以，我们需要把文档切成小块（chunk），每块 500-1000 字。

分 chunk 的三种策略

1. 固定大小分块（最简单）

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每块最多 500 字符
    chunk_overlap=50,    # 相邻块重叠 50 字符，避免切断关键信息
    separators=["

", "
", "。", "！", "？", " ", ""]
)
chunks = splitter.split_text(long_text)

2. 语义分块（效果更好）

按段落、标题、列表等语义边界切分，而不是机械地按字符数切。LangChain 的 MarkdownHeaderTextSplitter 可以做到：

from langchain.text_splitter import MarkdownHeaderTextSplitter

splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "Header_1"),
        ("##", "Header_2"),
        ("###", "Header_3"),
    ]
)
chunks = splitter.split_text(markdown_text)

3. 智能分块（最贵但最好）

用 LLM 判断哪里该切、哪里不该切。适合对精度要求极高的场景。

分块大小的黄金法则

场景	chunk_size	chunk_overlap	说明
客服问答	300-500	30-50	问题通常有明确答案，小 chunk 更精准
技术文档	500-1000	50-100	技术细节多，需要更多上下文
法律合同	200-400	20-50	法律条款精确，不宜过长
学术论文	800-1500	100-200	论文需要大上下文理解

经验：先用 500/50 跑 baseline，根据检索结果调优。

第二步：Embedding 模型选型

Embedding 模型决定了你的检索质量——如果模型把不相关的文档判定为”相似”，后面的步骤全白费。

主流 Embedding 模型对比

模型	维度	中文效果	价格	部署方式
text-embedding-3-small	1536	⭐⭐⭐⭐	$0.02/百万 token	API
text-embedding-3-large	3072	⭐⭐⭐⭐⭐	$0.13/百万 token	API
百炼 text-embedding-v3	1024	⭐⭐⭐⭐⭐	¥0.7/百万 token	API
BGE-m3	1024	⭐⭐⭐⭐⭐	免费	自建
GTE-Qwen2-7B-instruct	3584	⭐⭐⭐⭐⭐	免费	自建

选型建议：

• 中文场景首选：百炼 text-embedding-v3 或 BGE-m3（中文优化最好）
• 预算充足：OpenAI text-embedding-3-large
• 需要自建/离线：BGE-m3（HuggingFace 免费模型，效果接近 OpenAI）
• 端侧部署：BGE-m3 的 int8 量化版本，内存只需 1.5GB

代码示例：调用百炼 Embedding

from dashscope import TextEmbedding

response = TextEmbedding.call(
    model='text-embedding-v3',
    input=['什么是 RAG？', 'RAG 的核心原理是检索增强生成'],
    dimensions=1024,
    text_type='document'  # 'document' 或 'query'
)

# 获取向量
vectors = [item['embedding'] for item in response.output['embeddings']]
print(f"向量维度: {len(vectors[0])}")  # 1024

第三步：向量数据库存储与检索

用 Qdrant 搭建最小可用的向量检索

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

# 1. 连接（本地 Docker 模式）
client = QdrantClient("localhost", port=6333)

# 2. 创建 collection
client.create_collection(
    collection_name="rag_docs",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
)

# 3. 写入向量
points = [
    PointStruct(id=1, vector=vector1, payload={"text": "什么是 RAG？", "source": "doc1"}),
    PointStruct(id=2, vector=vector2, payload={"text": "RAG 的核心原理...", "source": "doc2"}),
]
client.upsert(collection_name="rag_docs", points=points)

# 4. 检索
results = client.search(
    collection_name="rag_docs",
    query_vector=query_vector,
    limit=3  # Top-3 相关片段
)

for r in results:
    print(f"得分: {r.score:.4f} | 文本: {r.payload['text'][:100]}")

检索策略：Top-K 和分数阈值

Top-K 怎么选？

• K=3-5：适合精准问答（客服、知识检索）
• K=5-10：适合内容生成（写作、摘要）
• K>10：通常引入噪声，不推荐

分数阈值（可选）：

results = client.search(..., score_threshold=0.7)

只返回相似度 > 0.7 的结果，避免检索到不相关的内容。

第四步：Prompt 组装

这是 RAG 中最关键也最容易被忽视的一步。Prompt 写得好不好，直接决定了 LLM 回答的质量。

好的 RAG Prompt 模板

你是一个专业的知识助手。请基于以下检索到的文档片段回答用户问题。

## 相关文档
{context_1}

{context_2}

{context_3}

## 用户问题
{question}

## 回答要求
1. 如果文档中有相关信息，请基于文档内容回答
2. 如果文档中没有相关信息，请说"根据现有资料无法回答该问题"
3. 不要编造文档中没有的信息
4. 回答要简洁明了，不超过 500 字

常见错误

❌ 错误 1：把检索结果直接塞进去，不告诉 LLM 怎么用

以下是相关资料：{context}
问题：{question}
请回答。

LLM 会困惑：这些资料和我的问题什么关系？我应该全部用上还是选部分？

✅ 正确：明确告诉 LLM 资料是什么、怎么用、边界在哪里。

❌ 错误 2：检索到 10 个片段全部塞给 LLM 上下文太长会导致 LLM “迷失在中间”（Lost in the Middle 现象）——开头和结尾的信息记得牢，中间的信息容易忽略。

✅ 正确：选 Top-3 到 Top-5 最相关的片段，按相关性排序（最相关的放最后，利用 LLM 的”近因效应”）。

第五步：LLM 生成

模型选择

模型	中文能力	推理速度	价格（百万 token）	适合场景
GPT-4o	⭐⭐⭐⭐	快	$2.50-10	通用
Claude Sonnet 4	⭐⭐⭐⭐	快	$3-15	写作/分析
DeepSeek-V3	⭐⭐⭐⭐⭐	中	¥0.14	中文/性价比
Qwen2.5-72B	⭐⭐⭐⭐⭐	中	免费自建	中文/开源

代码示例

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

response = client.chat.completions.create(
    model="qwen-plus",
    messages=[
        {"role": "system", "content": "你是专业的知识助手..."},
        {"role": "user", "content": f"## 相关文档
{context}

## 问题
{question}"}
    ],
    temperature=0.3  # 低温度，让回答更确定
)

print(response.choices[0].message.content)

第六步：生产部署架构

最小可用架构（适合 100 人以内）

用户 → FastAPI 服务 → Embedding API
                      ↓
                   Qdrant (Docker)
                      ↓
                   LLM API

一台 8 核 16GB 服务器跑 Qdrant + FastAPI，月成本 ¥300-500。

生产级架构（适合 1000+ 人）

                    ┌──────────┐
用户 → API Gateway →│ FastAPI  │→ Embedding API
                    │  Cluster │↓
                    └──────────┘→ Qdrant Cluster
                                  ↓
                               LLM API

需要 K8s + Qdrant 分布式部署 + 负载均衡，月成本 ¥2000-5000。

关键优化

1. 缓存：对常见问题做缓存（Redis），相同问题直接返回，不调用 LLM
2. 流式输出：用 stream=True 让 LLM 逐字输出，用户体验从”等 5 秒”变成”立刻有反馈”
3. 降级策略：当 LLM API 超时或报错时，返回”系统繁忙”而不是 500 错误
4. 日志记录：记录每次检索的 query、检索结果、LLM 回答，用于持续优化

总结

搭建 RAG Pipeline 的关键不在于技术选型有多高级，而在于每个环节都做对：

1. 分块：500 字符 + 50 字符重叠，起步配置
2. Embedding：中文选百炼 v3 或 BGE-m3
3. 向量库：小规模用 Qdrant，大规模用 Milvus
4. Prompt：明确告诉 LLM 资料是什么、怎么用、边界在哪
5. LLM：追求性价比用 DeepSeek 或 Qwen，追求质量用 GPT-4o 或 Claude

先用最小可用架构跑起来，确认效果后再扩展。

[[AFFILIATE:阿里云]]