AI Agent知识库构建：RAG系统最佳实践

一、开场：AI怎么”记住”你的业务知识？

大家好，我是老金。

上周老板提了个需求：”能不能让AI客服知道我们公司的所有产品知识？”

我说：”简单啊，把产品手册喂给AI。”

老板：”那售后知识呢？技术文档呢？历史案例呢？”

我算了一下，加起来50万字的文档。

这时候我就知道，光靠提示词是不行了，得上RAG（检索增强生成）。

今天分享一下RAG系统的搭建经验，从踩坑到最佳实践。

二、RAG是什么？为什么需要它？

LLM的知识局限

大语言模型有两个问题：

1. 知识有时效性：训练完就不更新了
2. 上下文窗口有限：塞不进所有知识

RAG的解决方案：

用户问题 → 检索相关知识 → 拼接提示词 → LLM生成回答

RAG vs 微调 vs 长上下文

方案	成本	时效性	适用场景
微调	高	差	风格迁移、领域适配
长上下文	中	好	文档不长、更新频繁
RAG	低	好	知识量大、需要实时更新

RAG的优势：知识更新只要改向量库，不需要重新训练模型。

三、RAG系统架构

基础架构

┌─────────────────────────────────────────────────────────┐
│                     RAG系统架构                          │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐            │
│  │ 文档    │ → │ 分块    │ → │ 向量化  │            │
│  │ (原始)  │    │ Chunking│    │Embedding│            │
│  └─────────┘    └─────────┘    └────┬────┘            │
│                                      │                  │
│                                      ▼                  │
│                              ┌──────────────┐          │
│                              │   向量数据库   │          │
│                              └──────┬───────┘          │
│                                     │                   │
│  ┌─────────┐    ┌─────────┐        │                   │
│  │ 用户    │ → │ 问题    │        │                   │
│  │ 问题    │    │ 向量化  │────────┼─→ 检索相似文档    │
│  └─────────┘    └─────────┘        │                   │
│                                     ▼                   │
│                              ┌──────────────┐          │
│                              │ 提示词拼接    │          │
│                              └──────┬───────┘          │
│                                     │                   │
│                                     ▼                   │
│                              ┌──────────────┐          │
│                              │  LLM生成     │          │
│                              └──────────────┘          │
│                                                         │
└─────────────────────────────────────────────────────────┘

核心组件

class RAGSystem:
    """RAG系统核心"""

    def __init__(self, 
                 embedder,      # 向量化模型
                 vector_store,  # 向量数据库
                 llm,           # 大语言模型
                 chunk_size=500,
                 chunk_overlap=50):
        self.embedder = embedder
        self.vector_store = vector_store
        self.llm = llm
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    async def index_documents(self, documents):
        """索引文档"""
        chunks = self.chunk_documents(documents)
        embeddings = await self.embedder.embed_batch(chunks)
        self.vector_store.add_batch(chunks, embeddings)

    async def query(self, question, top_k=5):
        """查询"""
        # 1. 问题向量化
        question_embedding = await self.embedder.embed(question)

        # 2. 检索相关文档
        relevant_chunks = self.vector_store.search(
            question_embedding, top_k=top_k
        )

        # 3. 构建提示词
        context = self.build_context(relevant_chunks)
        prompt = self.build_prompt(question, context)

        # 4. 生成回答
        answer = await self.llm.generate(prompt)

        return {
            "answer": answer,
            "sources": relevant_chunks
        }

四、文档分块策略

为什么分块很重要？

分块不好，检索就不好，回答就不好。这是RAG系统的第一道关卡。

分块方法对比

方法	优点	缺点	适用场景
固定长度	简单	可能截断语义	格式统一的文档
句子分割	语义完整	块太小信息不足	短文本
段落分割	语义完整	块大小不一	结构化文档
递归分割	灵活	需要调参	混合格式文档
语义分割	最佳语义	计算成本高	高质量要求场景

推荐方案：递归字符分割

from langchain.text_splitter import RecursiveCharacterTextSplitter

def create_chunker(chunk_size=500, chunk_overlap=50):
    """创建文档分块器"""
    return RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=[
            "nn",  # 双换行（段落）
            "n",    # 单换行
            "。",    # 中文句号
            ".",     # 英文句号
            "！",    # 中文感叹号
            "！",    # 中文问号
            " ",     # 空格
            ""       # 字符
        ],
        length_function=len,
        is_separator_regex=False
    )

分块大小的选择

# 根据场景选择分块大小
CHUNK_SIZES = {
    "faq": 200,           # FAQ问答：小块，精准匹配
    "product_docs": 500,  # 产品文档：中等块，上下文完整
    "technical": 1000,    # 技术文档：大块，保留代码块
    "legal": 1500,        # 法律文档：大块，条款完整
}

进阶：元数据增强

class EnhancedChunker:
    """带元数据的分块器"""

    def chunk_with_metadata(self, document, metadata):
        """分块并添加元数据"""
        chunks = self.splitter.split_text(document)

        enhanced_chunks = []
        for i, chunk in enumerate(chunks):
            enhanced_chunks.append({
                "content": chunk,
                "metadata": {
                    **metadata,
                    "chunk_index": i,
                    "total_chunks": len(chunks),
                    "char_start": self._get_char_offset(document, i),
                    "char_end": self._get_char_offset(document, i + 1)
                }
            })

        return enhanced_chunks

五、向量化模型选择

模型对比

模型	维度	性能	成本	推荐场景
OpenAI text-embedding-3-small	1536	好	低	通用场景
OpenAI text-embedding-3-large	3072	很好	中	高精度需求
BGE-large-zh	1024	好	免费	中文场景
M3E-large	1024	好	免费	中文多领域
Cohere embed-v3	1024	很好	中	多语言

我的推荐

# 中文为主：BGE
from sentence_transformers import SentenceTransformer

embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 多语言/英文为主：OpenAI
from openai import OpenAI
client = OpenAI()

def embed_openai(texts):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [d.embedding for d in response.data]

向量维度的影响

# 维度越高，表达能力越强，但存储成本也越高
dimensions = {
    384: "轻量级，适合大规模低精度场景",
    768: "标准，平衡性能和成本",
    1024: "高质量，推荐默认选择",
    1536: "高质量，OpenAI默认",
    3072: "最高质量，成本也最高"
}

六、向量数据库选择

主流向量数据库对比

数据库	类型	优点	缺点	推荐场景
Pinecone	云服务	开箱即用	收费	生产环境快速上线
Milvus	开源	高性能	部署复杂	大规模生产
Chroma	开源	简单易用	不适合大规模	原型开发
Weaviate	开源	功能丰富	学习曲线	需要混合检索
Qdrant	开源	高性能Rust	生态较小	性能优先
PGVector	PostgreSQL扩展	与SQL结合	性能一般	已有PG环境

快速上手：Chroma

import chromadb
from chromadb.config import Settings

# 初始化
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

# 创建集合
collection = client.create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}
)

# 添加文档
collection.add(
    documents=["文档内容1", "文档内容2"],
    metadatas=[{"source": "file1"}, {"source": "file2"}],
    ids=["id1", "id2"]
)

# 查询
results = collection.query(
    query_texts=["用户问题"],
    n_results=5
)

生产推荐：Milvus

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 连接
connections.connect("default", host="localhost", port="19530")

# 定义schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
    FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=256)
]

schema = CollectionSchema(fields, "knowledge_base")
collection = Collection("knowledge_base", schema)

# 创建索引
index_params = {
    "metric_type": "COSINE",
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}
collection.create_index(field_name="embedding", index_params=index_params)

七、检索策略优化

基础检索的问题

单纯用向量相似度检索有几个问题：

1. 语义漂移：相似但不相关的文档
2. 关键词丢失：专有名词匹配不到
3. 上下文缺失：检索到的块不完整

优化方案1：混合检索

class HybridRetriever:
    """混合检索：向量 + 关键词"""

    def __init__(self, vector_store, bm25_index, alpha=0.5):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.alpha = alpha  # 向量检索权重

    def retrieve(self, query, top_k=10):
        # 向量检索
        vector_results = self.vector_store.search(query, top_k=top_k * 2)

        # BM25检索
        bm25_results = self.bm25_index.search(query, top_k=top_k * 2)

        # 融合结果
        return self.rrf_fusion(vector_results, bm25_results, top_k)

    def rrf_fusion(self, vector_results, bm25_results, top_k):
        """Reciprocal Rank Fusion融合"""
        scores = {}

        for i, doc in enumerate(vector_results):
            scores[doc.id] = scores.get(doc.id, 0) + 1 / (i + 60)

        for i, doc in enumerate(bm25_results):
            scores[doc.id] = scores.get(doc.id, 0) + 1 / (i + 60)

        # 排序返回top_k
        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_docs[:top_k]

优化方案2：重排序

class Reranker:
    """重排序器"""

    def __init__(self, rerank_model):
        self.rerank_model = rerank_model

    async def rerank(self, query, documents, top_k=5):
        """重排序检索结果"""
        # 计算query与每个文档的相关性分数
        pairs = [(query, doc.content) for doc in documents]
        scores = await self.rerank_model.score(pairs)

        # 按分数重排序
        scored_docs = list(zip(documents, scores))
        scored_docs.sort(key=lambda x: x[1], reverse=True)

        return [doc for doc, score in scored_docs[:top_k]]

优化方案3：查询改写

async def rewrite_query(original_query, llm):
    """改写用户查询，提高检索效果"""
    prompt = f"""
    用户问题：{original_query}

    请将这个问题改写为更适合检索的形式：
    1. 补充关键词
    2. 明确搜索意图
    3. 扩展相关概念

    改写后的问题：
    """

    rewritten = await llm.generate(prompt)
    return rewritten

# 示例
# 原始：怎么退款
# 改写：订单退款流程 退货退款条件 售后退款申请

八、提示词工程

RAG提示词模板

RAG_PROMPT_TEMPLATE = """
你是一个知识库问答助手。请根据以下参考文档回答用户问题。

## 参考文档
{context}

## 用户问题
{question}

## 回答要求
1. 优先使用参考文档中的信息回答
2. 如果参考文档中没有相关信息，请明确说明"根据现有知识库无法回答"
3. 引用具体来源时使用[文档名]格式
4. 回答要简洁准确

## 回答
"""

来源引用

def build_context_with_citations(chunks):
    """构建带引用标注的上下文"""
    context_parts = []

    for i, chunk in enumerate(chunks):
        citation = f"[{chunk.metadata.get('source', '未知来源')}]"
        context_parts.append(f"{citation}n{chunk.content}")

    return "nn---nn".join(context_parts)

九、实战案例：客服知识库

需求

• 产品手册：100个产品，每个10页
• FAQ：500条问答
• 技术文档：50篇
• 历史工单：10000条

实现方案

class CustomerServiceKnowledgeBase:
    """客服知识库"""

    def __init__(self):
        self.embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5')
        self.vector_store = ChromaVectorStore()
        self.llm = GPT4Client()
        self.reranker = CrossEncoderReranker()

    async def index_knowledge(self, knowledge_dir):
        """索引知识库"""
        # 分类处理不同类型文档
        for doc_type in ["products", "faq", "tech_docs", "tickets"]:
            docs = self.load_documents(f"{knowledge_dir}/{doc_type}")
            chunks = self.chunk_documents(docs, doc_type)
            await self.index_chunks(chunks, doc_type)

    def chunk_documents(self, docs, doc_type):
        """根据文档类型选择分块策略"""
        chunk_sizes = {
            "products": 500,
            "faq": 200,      # FAQ保持完整
            "tech_docs": 1000,
            "tickets": 300
        }
        chunker = create_chunker(
            chunk_size=chunk_sizes[doc_type]
        )
        return chunker.split_documents(docs)

    async def answer(self, question):
        """回答问题"""
        # 1. 查询改写
        rewritten_query = await rewrite_query(question, self.llm)

        # 2. 混合检索
        candidates = self.hybrid_retrieve(rewritten_query, top_k=20)

        # 3. 重排序
        ranked_docs = await self.reranker.rerank(
            question, candidates, top_k=5
        )

        # 4. 生成回答
        context = build_context_with_citations(ranked_docs)
        answer = await self.llm.generate(
            RAG_PROMPT_TEMPLATE.format(context=context, question=question)
        )

        return {
            "answer": answer,
            "sources": [doc.metadata for doc in ranked_docs]
        }

十、评估与优化

RAG评估指标

class RAGEvaluator:
    """RAG系统评估器"""

    def evaluate(self, test_cases):
        """评估RAG系统"""
        results = {
            "retrieval_precision": [],
            "retrieval_recall": [],
            "answer_relevance": [],
            "answer_faithfulness": [],
            "answer_correctness": []
        }

        for case in test_cases:
            # 检索评估
            retrieved = self.rag.retrieve(case.question)
            results["retrieval_precision"].append(
                self.calc_precision(retrieved, case.relevant_docs)
            )

            # 生成评估
            answer = self.rag.answer(case.question)
            results["answer_relevance"].append(
                self.eval_relevance(answer, case.question)
            )
            results["answer_faithfulness"].append(
                self.eval_faithfulness(answer, retrieved)
            )

        return {k: mean(v) for k, v in results.items()}

持续优化

问题	诊断方法	优化方案
检索不到相关文档	检查查准率	调整分块大小、使用混合检索
检索到不相关文档	检查查准率	添加重排序、优化查询
回答不准确	检查忠实度	优化提示词、添加约束
回答不完整	检查相关性	增加检索数量、改写查询

十一、总结与行动清单

关键要点

1. 分块是基础：好的分块才有好的检索
2. 检索是核心：混合检索+重排序效果最佳
3. 评估是保障：定期评估，持续优化
4. 成本要平衡：不是所有场景都需要最复杂的方案

行动清单

步骤	任务	时间
1	收集整理知识文档	1天
2	选择向量化模型和向量库	半天
3	实现基础RAG流程	1天
4	优化分块策略	半天
5	添加混合检索	1天
6	添加重排序	半天
7	构建评估数据集	1天
8	持续优化迭代	持续

下期预告

明天聊聊自定义工具开发——让AI Agent连接你的业务系统！

---

往期回顾

• AI Agent成本控制指南：Token花得肉疼怎么办？
• AI Agent评估体系：如何量化”智能程度”？
• AI Agent生产部署指南：从灰度发布到快速回滚