第三章：RAG 与知识库产品化
第三章：RAG 与知识库产品化
基础大模型的知识截止于某个时间点，不包含你的内部文档，不了解你的产品。RAG 是让大模型"懂你"的技术桥梁。
3.1 RAG 是什么
RAG（Retrieval-Augmented Generation，检索增强生成）的原理很简单：
传统 LLM：用户提问 → 模型用"训练时记住的"知识回答
RAG：用户提问 → 在知识库里搜索相关文档 → 把文档 + 问题一起发给 LLM → LLM 基于文档回答
为什么需要 RAG 而不是直接把文档塞进 Prompt：
GPT-4o 支持 128K context window。理论上可以把大量文档直接塞进去。但问题是：
- 成本：100 页文档 ≈ 80,000 tokens，每次调用约 $0.08（用 GPT-4o）
- 准确性：模型在超长 context 中"找信息"能力下降（中间部分容易被忽视）
- 效率：RAG 只传递最相关的片段（通常 3-10 个），而不是全部
RAG 的实际价值：
- 让 AI 回答基于最新数据（不受训练截止日期限制）
- 让 AI 回答基于私有/专有知识
- 大幅降低幻觉（AI 有依据才回答，而不是"编造"）
- 降低每次调用的 token 成本
3.2 向量数据库原理
RAG 的核心是语义搜索，语义搜索的核心是向量（Embedding）。
文字 "苹果公司的市值" → Embedding 模型 → [0.023, -0.145, 0.891, ...] (1536 维向量)
文字 "Apple 公司股票价格" → Embedding 模型 → [0.019, -0.138, 0.879, ...] (类似的向量)
余弦相似度高 → 语义相近
构建 RAG 知识库的步骤：
from openai import OpenAI
import numpy as np
client = OpenAI()
# 1. 文档分块（Chunking）
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
"""将长文档切成小块，每块有一定重叠以保持上下文"""
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
if chunk:
chunks.append(chunk)
return chunks
# 2. 生成 Embedding
def get_embedding(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small", # 便宜：$0.02/百万 token
input=text
)
return response.data[0].embedding
# 3. 语义搜索
def cosine_similarity(v1: list[float], v2: list[float]) -> float:
v1, v2 = np.array(v1), np.array(v2)
return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
def search_knowledge_base(query: str, documents: list[dict], top_k: int = 3) -> list[dict]:
query_embedding = get_embedding(query)
results = []
for doc in documents:
similarity = cosine_similarity(query_embedding, doc["embedding"])
results.append({**doc, "similarity": similarity})
return sorted(results, key=lambda x: x["similarity"], reverse=True)[:top_k]
3.3 向量数据库对比
| 数据库 | 类型 | 价格 | 适合场景 |
|---|---|---|---|
| Pinecone | 云 SaaS | 免费 1 个 index（100 万向量），$70/月起 | 快速上线，不想管基础设施 |
| Weaviate | 开源/云 | 开源免费自托管；云版本按使用付费 | 需要混合搜索（向量+关键词） |
| Qdrant | 开源/云 | 开源免费；云 $25/月起 | 高性能，Rust 实现，低内存占用 |
| pgvector | PostgreSQL 扩展 | 免费（需要自己的 PG 实例） | 已有 PostgreSQL 的项目，避免引入新数据库 |
| Chroma | 开源 | 完全免费，本地或云 | 本地开发、测试、小规模应用 |
| Supabase Vector | 云托管 pgvector | 包含在 Supabase 套餐中 | Supabase 用户的最省事选择 |
实际选择建议：
- 个人项目/MVP：Chroma 本地 + pgvector 生产
- 快速上线 SaaS：Pinecone（免费层已够用）
- 已有 PostgreSQL：pgvector（避免额外数据库）
3.4 用 pgvector 构建完整 RAG 系统
-- 安装 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;
-- 知识库表
CREATE TABLE knowledge_chunks (
id BIGSERIAL PRIMARY KEY,
source_id INTEGER, -- 来源文档 ID
content TEXT NOT NULL, -- 原始文本
embedding vector(1536), -- OpenAI text-embedding-3-small 维度
metadata JSONB, -- 文档名称、页码、标签等
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- 向量索引（HNSW 算法，适合高召回率）
CREATE INDEX ON knowledge_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
import asyncpg
from openai import OpenAI
client = OpenAI()
async def add_document_to_knowledge_base(
db_pool: asyncpg.Pool,
content: str,
source_name: str,
metadata: dict = None
):
"""将文档添加到知识库"""
chunks = chunk_text(content, chunk_size=400, overlap=50)
async with db_pool.acquire() as conn:
for chunk in chunks:
embedding = get_embedding(chunk)
await conn.execute(
"""
INSERT INTO knowledge_chunks (content, embedding, metadata)
VALUES ($1, $2, $3)
""",
chunk,
embedding,
{"source": source_name, **(metadata or {})}
)
async def rag_query(
db_pool: asyncpg.Pool,
question: str,
top_k: int = 5,
system_context: str = ""
) -> str:
"""RAG 查询：检索相关文档 → 发给 LLM 回答"""
# 1. 生成问题的 Embedding
query_embedding = get_embedding(question)
# 2. 语义搜索最相关的文档块
async with db_pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT content, metadata,
1 - (embedding <=> $1::vector) AS similarity
FROM knowledge_chunks
ORDER BY embedding <=> $1::vector
LIMIT $2
""",
query_embedding,
top_k
)
if not rows:
return "未找到相关信息。"
# 3. 组装 Context
context = "\n\n---\n\n".join([
f"[来源：{row['metadata'].get('source', '未知')}]\n{row['content']}"
for row in rows
])
# 4. 发给 LLM
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": f"""你是一个知识库助手。请基于以下参考文档回答用户问题。
如果文档中没有相关信息，请明确说明"我的知识库中没有这方面的信息"，不要编造答案。
{system_context}
【参考文档】
{context}"""
},
{"role": "user", "content": question}
]
)
return response.choices[0].message.content
3.5 文档 Q&A SaaS 产品化
产品定位示例：企业合同知识库
目标客户：100-1,000 人规模的企业法务团队
核心痛点：合同审查、条款查询需要大量人工时间，法律顾问昂贵
产品功能：
- 上传合同文档（PDF、Word）
- AI 自动解析并建立知识库
- 用自然语言问任何关于合同的问题
- 高亮显示原文出处，可追溯
定价：
Starter: $49/月 — 50 个文档，2 个用户
Professional: $149/月 — 500 个文档，10 个用户
Enterprise: $499+/月 — 无限文档，SAML SSO，API 访问
文件解析层
import fitz # PyMuPDF，处理 PDF
import docx
from pathlib import Path
class DocumentParser:
def parse_pdf(self, file_path: str) -> str:
"""提取 PDF 文本，保留段落结构"""
doc = fitz.open(file_path)
text_parts = []
for page_num, page in enumerate(doc):
text = page.get_text("text")
if text.strip():
text_parts.append(f"[第{page_num+1}页]\n{text}")
return "\n\n".join(text_parts)
def parse_docx(self, file_path: str) -> str:
"""提取 Word 文档文本"""
doc = docx.Document(file_path)
paragraphs = [para.text for para in doc.paragraphs if para.text.strip()]
return "\n\n".join(paragraphs)
def parse(self, file_path: str) -> str:
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
return self.parse_pdf(file_path)
elif suffix in [".docx", ".doc"]:
return self.parse_docx(file_path)
elif suffix == ".txt":
return Path(file_path).read_text()
else:
raise ValueError(f"Unsupported file type: {suffix}")
3.6 RAG 质量优化
分块策略对质量影响巨大：
- 固定大小分块（简单，但可能切断句子）：适合结构化文档
- 按句/段落分块（保持语义完整性）：适合文章/报告
- 按标题层级分块（保留文档结构）：适合有章节的手册
提升 RAG 准确率的技术：
# Hybrid Search：向量搜索 + 关键词搜索 结合
async def hybrid_search(query: str, top_k: int = 5) -> list[dict]:
"""
结合语义搜索和全文搜索，取两者最佳结果
"""
# 1. 向量搜索（语义相似）
vector_results = await vector_search(query, top_k=top_k * 2)
# 2. 全文搜索（关键词匹配）
text_results = await full_text_search(query, top_k=top_k * 2)
# 3. RRF（Reciprocal Rank Fusion）合并排名
scores = {}
k = 60 # RRF 常数
for rank, result in enumerate(vector_results):
doc_id = result["id"]
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
for rank, result in enumerate(text_results):
doc_id = result["id"]
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
# 返回融合排名后的 top_k 结果
sorted_ids = sorted(scores, key=scores.get, reverse=True)[:top_k]
return [get_chunk_by_id(doc_id) for doc_id in sorted_ids]
小结
RAG 的商业本质是：把"数据资产"转化为"可交互的知识"。一个拥有 10 年合同库的律师事务所，它的真正资产不只是合同文本，而是这些文本中积累的法律判断——RAG 让这个资产变得可检索、可问答。
这是 AI 时代最值钱的产品形态之一：你不需要更好的基础模型，你需要更好的数据 + 更好的检索系统。
下一章，AI Agent：让 AI 不只是回答问题，而是自主完成任务。