第三章：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 人规模的企业法务团队
核心痛点：合同审查、条款查询需要大量人工时间，法律顾问昂贵
产品功能：

1. 上传合同文档（PDF、Word）
2. AI 自动解析并建立知识库
3. 用自然语言问任何关于合同的问题
4. 高亮显示原文出处，可追溯

定价：

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 不只是回答问题，而是自主完成任务。