AI 技术

RAG 入门到实战：3 天搭一个企业知识库问答系统（2026 完整版）

Avatar photo
作者 airobot
2026年6月25日 7 分钟阅读
0
21-rag-3-days-enterprise-cover.png
RAG 入门到实战封面

RAG 入门到实战:3 天搭一个企业知识库问答系统(2026 完整版)

分类:AI 技术 (ai-tech) | 标签:RAG、大模型、知识库、向量数据库、LangChain

RAG(Retrieval-Augmented Generation)是 2026 年企业用大模型的核心技术。

简单说:让 AI 先查你公司的资料,再回答问题。

这样 AI 不会瞎编(幻觉),答案100% 来自你的文档。

这篇带你 3 天从 0 搭一个企业知识库问答系统:

  • Day 1:文档处理(上传/分段/向量化)
  • Day 2:智能问答(检索 + 生成)
  • Day 3:部署上线(API + Web UI)

完整代码 + 部署指南,直接抄作业。

RAG 到底是什么:用人话说清楚

传统 AI 的痛点

你问 GPT:”我们公司年假政策是什么?“

GPT 会编一个答案,因为它不知道你公司的 HR 手册。

RAG 怎么解决

先查 → 再答:

1. 你问问题:”年假政策是什么”

2. AI 去你公司文档库搜(“HR 手册.pdf”)

3. 找到相关段落

4. 把段落 + 你的问题 一起给大模型

5. 大模型基于真实资料生成答案

结果:答案 100% 基于公司资料,不会瞎编。

RAG vs 微调:什么时候用哪个

| 维度 | RAG | 微调 |

|—|—|—|

| 成本 | 低(几小时搞定) | 高(几周 + GPU 集群) |

| 数据量 | 10-10000 文档 | 1000+ 标注样本 |

| 实时更新 | ✅ 加文档即可 | ❌ 重新训练 |

| 适合 | 企业知识库/客服/法规 | 风格/口吻/特定任务 |

| 难度 | 入门 | 高级 |

结论:90% 企业场景用 RAG 就够了。只有需要”AI 学某种特定风格”才用微调。

Day 1:文档处理(上传/分段/向量化)

技术栈选型

| 模块 | 推荐工具 | 替代方案 |

|—|—|—|

| 大模型 | DeepSeek R2(便宜) | 通义千问 / Claude / GPT-4o |

| Embedding | BGE-M3(中文最强) | OpenAI text-embedding-3 |

| 向量数据库 | Milvus / Qdrant | Chroma / Weaviate / pgvector |

| 文档解析 | MinerU(国产) | Unstructured / PyMuPDF |

| 分段 | LangChain | LlamaIndex / 自定义 |

| Web 框架 | FastAPI | Flask / Django |

| 前端 | Next.js + shadcn/ui | Streamlit / Gradio |

为什么选这套:

  • DeepSeek R2:0.5 元/1M tokens(便宜 30 倍)
  • BGE-M3:中文 embedding 之王
  • Milvus:国产向量数据库,中文文档全
  • MinerU:国产 PDF 解析,中文友好

Step 1:环境准备

mkdir company-kb-qa
cd company-kb-qa
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install fastapi uvicorn langchain langchain-community \
  langchain-deepseek dashscope sentence-transformers \
  pymilvus python-multipart aiofiles

Step 2:文档解析(支持 PDF/Word/Excel/Markdown)

# document_parser.py
import os
from pathlib import Path
from langchain_community.document_loaders import (
    PyPDFLoader, UnstructuredWordDocumentLoader,
    UnstructuredExcelLoader, UnstructuredMarkdownLoader
)

def parse_document(filepath: str):
    """根据文件类型解析文档"""
    ext = Path(filepath).suffix.lower()
    
    if ext == '.pdf':
        loader = PyPDFLoader(filepath)
    elif ext in ['.docx', '.doc']:
        loader = UnstructuredWordDocumentLoader(filepath)
    elif ext in ['.xlsx', '.xls']:
        loader = UnstructuredExcelLoader(filepath)
    elif ext == '.md':
        loader = UnstructuredMarkdownLoader(filepath)
    else:
        raise ValueError(f"不支持的文件类型: {ext}")
    
    docs = loader.load()
    print(f"✅ 解析 {filepath}: {len(docs)} 页")
    return docs

# 测试
if __name__ == "__main__":
    docs = parse_document("test_docs/HR手册.pdf")
    print(f"第一页内容:\n{docs[0].page_content[:200]}")

Step 3:智能分段(关键!)

分段的好坏直接决定 RAG 效果。一段太长(超过 2000 字)AI 抓不住重点,太短(50 字)又缺上下文。

# text_splitter.py
from langchain.text_splitter import RecursiveCharacterTextSplitter

def smart_split(docs, chunk_size=500, chunk_overlap=50):
    """
    智能分段:
    - chunk_size=500:每段 500 字左右
    - chunk_overlap=50:段之间重叠 50 字(避免切断语义)
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],
        length_function=len,
    )
    
    chunks = splitter.split_documents(docs)
    print(f"✅ 分段完成: {len(docs)} 页 → {len(chunks)} 段")
    return chunks

# 测试
chunks = smart_split(docs)
print(f"第一段:\n{chunks[0].page_content}\n---")
print(f"第一段来源: {chunks[0].metadata}")

分段技巧:

  • 中文文档用 “\n\n / \n / 。” 分隔符(优于纯空格)
  • 表格/列表特殊处理(用 Unstructured 的 strategy="fast" 模式)
  • chunk_size = 500-1000 字(根据文档调整)
  • overlap = 10% chunk_size

Step 4:向量化(Embedding)

# embedder.py
from sentence_transformers import SentenceTransformer
import numpy as np

class Embedder:
    def __init__(self, model_name="BAAI/bge-m3"):
        """使用 BGE-M3(中文 embedding 王者)"""
        print(f"加载模型: {model_name}")
        self.model = SentenceTransformer(model_name)
        # 首次运行会自动下载约 2GB 模型
    
    def embed_texts(self, texts: list[str]) -> list[list[float]]:
        """批量向量化"""
        embeddings = self.model.encode(
            texts,
            batch_size=32,
            normalize_embeddings=True,  # 归一化,方便余弦相似度
            show_progress_bar=True
        )
        return embeddings.tolist()
    
    def embed_query(self, query: str) -> list[float]:
        """向量化单个查询"""
        return self.model.encode(
            query,
            normalize_embeddings=True
        ).tolist()

# 单例
_embedder = None
def get_embedder():
    global _embedder
    if _embedder is None:
        _embedder = Embedder()
    return _embedder

# 测试
if __name__ == "__main__":
    emb = get_embedder()
    vectors = emb.embed_texts(["年假怎么请", "请假流程"])
    print(f"✅ 向量化完成: 2 个文本 → 1024 维向量")
    print(f"第一段向量前 5 维: {vectors[0][:5]}")

Step 5:存入向量数据库

# vector_store.py
from pymilvus import MilvusClient, DataType
import uuid

class VectorStore:
    def __init__(self, db_path="./milvus.db", collection_name="company_kb"):
        self.client = MilvusClient(db_path)
        self.collection = collection_name
        self._init_collection()
    
    def _init_collection(self):
        """初始化集合(如果不存在)"""
        if self.client.has_collection(self.collection):
            return
        
        schema = self.client.create_schema(
            auto_id=False,
            enable_dynamic_field=True
        )
        schema.add_field("id", DataType.VARCHAR, max_length=64, is_primary=True)
        schema.add_field("vector", DataType.FLOAT_VECTOR, dim=1024)
        schema.add_field("text", DataType.VARCHAR, max_length=8000)
        schema.add_field("source", DataType.VARCHAR, max_length=512)
        
        index_params = self.client.prepare_index_params()
        index_params.add_index(
            field_name="vector",
            index_type="IVF_FLAT",
            metric_type="COSINE",
            params={"nlist": 128}
        )
        
        self.client.create_collection(
            collection_name=self.collection,
            schema=schema,
            index_params=index_params
        )
        print(f"✅ 创建集合: {self.collection}")
    
    def add_chunks(self, chunks, embeddings):
        """添加文档段到向量库"""
        data = []
        for chunk, vector in zip(chunks, embeddings):
            data.append({
                "id": str(uuid.uuid4()),
                "vector": vector,
                "text": chunk.page_content,
                "source": chunk.metadata.get("source", "unknown")
            })
        
        self.client.insert(collection_name=self.collection, data=data)
        print(f"✅ 插入 {len(data)} 段到向量库")
    
    def search(self, query_vector, top_k=5):
        """检索最相似的 top_k 段"""
        results = self.client.search(
            collection_name=self.collection,
            data=[query_vector],
            limit=top_k,
            output_fields=["text", "source"]
        )
        return results[0]  # 第一个查询的结果

# 单例
_vs = None
def get_vector_store():
    global _vs
    if _vs is None:
        _vs = VectorStore()
    return _vs

Day 1 测试

# test_day1.py
from document_parser import parse_document
from text_splitter import smart_split
from embedder import get_embedder
from vector_store import get_vector_store

# 1. 解析
docs = parse_document("test_docs/HR手册.pdf")
chunks = smart_split(docs)

# 2. 向量化
embedder = get_embedder()
texts = [c.page_content for c in chunks]
vectors = embedder.embed_texts(texts)

# 3. 入库
vs = get_vector_store()
vs.add_chunks(chunks, vectors)

print("🎉 Day 1 完成!文档已入库")

Day 2:智能问答(检索 + 生成)

Step 6:问答 Prompt 模板

# prompt_template.py

QA_PROMPT = """你是一个专业的企业知识库助手。请基于以下参考资料回答用户问题。

【严格要求】
1. 答案必须 100% 基于参考资料,不允许编造任何信息
2. 如果参考资料里没有答案,直接说"抱歉,文档里没有相关信息"
3. 回答时引用具体来源(文件名 + 段落)
4. 回答简洁清晰,不要废话

【参考资料】
{context}

【用户问题】
{question}

【回答格式】
答案: [你的回答]
来源: [引用了哪些文档,具体哪段]
"""

CONTEXT_TEMPLATE = """文档 {index}:
---
来源:{source}
内容:{text}
---"""

Step 7:问答主逻辑

# qa_engine.py
from langchain_deepseek import ChatDeepSeek
from langchain_core.messages import HumanMessage
from prompt_template import QA_PROMPT, CONTEXT_TEMPLATE
from embedder import get_embedder
from vector_store import get_vector_store

class QAEngine:
    def __init__(self):
        self.embedder = get_embedder()
        self.vs = get_vector_store()
        self.llm = ChatDeepSeek(
            model="deepseek-reasoner-r2",
            temperature=0.1,  # 低温度,避免编造
            max_tokens=2000
        )
    
    def ask(self, question: str, top_k: int = 5):
        """问答主流程"""
        # 1. 向量化问题
        q_vector = self.embedder.embed_query(question)
        
        # 2. 检索 top_k 相关段落
        results = self.vs.search(q_vector, top_k=top_k)
        
        if not results:
            return {
                "answer": "抱歉,知识库里没有相关信息。",
                "sources": []
            }
        
        # 3. 拼装 prompt
        context_parts = []
        sources = []
        for i, hit in enumerate(results, 1):
            entity = hit["entity"]
            context_parts.append(CONTEXT_TEMPLATE.format(
                index=i,
                source=entity["source"],
                text=entity["text"][:1000]  # 截断,避免超长
            ))
            sources.append({
                "source": entity["source"],
                "score": hit["distance"],
                "text": entity["text"][:200]
            })
        
        context = "\n\n".join(context_parts)
        prompt = QA_PROMPT.format(context=context, question=question)
        
        # 4. 调用大模型
        response = self.llm.invoke([HumanMessage(content=prompt)])
        
        return {
            "answer": response.content,
            "sources": sources
        }

# 单例
_qa = None
def get_qa_engine():
    global _qa
    if _qa is None:
        _qa = QAEngine()
    return _qa

Step 8:测试问答

# test_day2.py
from qa_engine import get_qa_engine

qa = get_qa_engine()

# 测试 1
result = qa.ask("年假怎么请?需要提前几天申请?")
print("=" * 50)
print(f"问题:年假怎么请?需要提前几天申请?")
print(f"\n答案:\n{result['answer']}")
print(f"\n来源({len(result['sources'])} 条):")
for s in result['sources']:
    print(f"  - {s['source']} (相似度:{s['score']:.3f})")

# 测试 2
result = qa.ask("公司年假是几天?")
print("=" * 50)
print(f"问题:公司年假是几天?")
print(f"\n答案:\n{result['answer']}")

预期输出:

问题:年假怎么请?需要提前几天申请?

答案:根据 HR 手册规定,员工请年假需提前 3 个工作日在 OA 系统提交申请,
单次年假不超过 15 天。连续年假超过 5 天需部门负责人审批。

来源(3 条):
  - HR手册.pdf (相似度:0.892)
  - OA系统使用指南.pdf (相似度:0.745)
  - 假期管理制度.pdf (相似度:0.621)

Day 3:部署上线(API + Web UI)

Step 9:FastAPI 后端

# main.py
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List
import shutil
import os

from document_parser import parse_document
from text_splitter import smart_split
from embedder import get_embedder
from vector_store import get_vector_store
from qa_engine import get_qa_engine

app = FastAPI(title="企业知识库问答系统", version="1.0")

# 跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

UPLOAD_DIR = "./uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

# 数据模型
class QuestionRequest(BaseModel):
    question: str
    top_k: int = 5

class QuestionResponse(BaseModel):
    question: str
    answer: str
    sources: List[dict]

# API:上传文档
@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """上传并处理文档"""
    # 保存文件
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    with open(file_path, "wb") as f:
        shutil.copyfileobj(file.file, f)
    
    try:
        # 处理流程
        docs = parse_document(file_path)
        chunks = smart_split(docs)
        
        embedder = get_embedder()
        vectors = embedder.embed_texts([c.page_content for c in chunks])
        
        vs = get_vector_store()
        vs.add_chunks(chunks, vectors)
        
        return {
            "filename": file.filename,
            "pages": len(docs),
            "chunks": len(chunks),
            "status": "success"
        }
    except Exception as e:
        raise HTTPException(500, f"处理失败: {str(e)}")

# API:问答
@app.post("/ask", response_model=QuestionResponse)
async def ask_question(req: QuestionRequest):
    """智能问答"""
    qa = get_qa_engine()
    result = qa.ask(req.question, top_k=req.top_k)
    return QuestionResponse(
        question=req.question,
        answer=result["answer"],
        sources=result["sources"]
    )

# 健康检查
@app.get("/health")
async def health():
    return {"status": "ok"}

# 启动: uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Step 10:Next.js 前端(简化版)

// app/page.tsx
'use client';
import { useState } from 'react';

export default function Home() {
  const [question, setQuestion] = useState('');
  const [answer, setAnswer] = useState('');
  const [sources, setSources] = useState([]);
  const [loading, setLoading] = useState(false);

  const ask = async () => {
    if (!question.trim()) return;
    setLoading(true);
    const res = await fetch('http://localhost:8000/ask', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ question, top_k: 5 })
    });
    const data = await res.json();
    setAnswer(data.answer);
    setSources(data.sources);
    setLoading(false);
  };

  return (
    

      

        
📚 企业知识库