AIGC应用实战：从零构建企业级RAG检索增强生成系统的完整方案

为什么企业需要RAG检索增强生成系统

大语言模型的训练数据存在截止日期，无法获取企业内部知识库的实时信息。AIGC应用落地过程中，模型幻觉和知识时效性是最突出的两个问题。检索增强生成（Retrieval-Augmented Generation）通过外部知识库检索与模型生成相结合的方式，有效解决了这两个痛点。

一套可用的RAG系统包含四个核心模块：文档处理与向量化、向量存储与检索、Prompt组装与模型调用、结果后处理与评估。这篇文章会从工程实现的角度，逐步拆解每个模块的技术选型和配置细节。

文档处理与向量化模块设计

文档处理是RAG系统的入口，决定了后续检索的质量上限。企业场景下，知识文档格式多样——PDF、Word、Markdown、HTML、甚至扫描件。处理流程分为三步：

第一步：文档解析与文本提取

PDF解析推荐PyMuPDF，速度快且对表格和公式有较好的保留能力。Word文档用python-docx提取段落。扫描件需要先走OCR，PaddleOCR在中文场景下识别率较高。

import fitz  # PyMuPDF

def extract_pdf_text(file_path):
    doc = fitz.open(file_path)
    text_chunks = []
    for page_num, page in enumerate(doc):
        text = page.get_text("text")
        if text.strip():
            text_chunks.append({
                "page": page_num + 1,
                "content": text.strip()
            })
    return text_chunks

第二步：文本分块（Chunking）

分块策略直接影响检索精度。固定长度分块最简单，但会切断语义完整性。推荐按段落+滑动窗口的方式分块，块大小512 token，重叠128 token，在语义完整性和检索召回率之间取得平衡。

def chunk_text(text, chunk_size=512, overlap=128):
    tokens = text.split()
    chunks = []
    start = 0
    while start < len(tokens):
        end = start + chunk_size
        chunk = " ".join(tokens[start:end])
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks

第三步：向量化嵌入

中文场景下，bge-large-zh-v1.5和m3e-base是两个主流选择。bge-large在C-MTEB基准上综合排名第一，m3e-base在轻量级部署场景表现优秀。向量化用sentence-transformers库：

from sentence_transformers import SentenceTransformer

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

def embed_texts(texts, batch_size=32):
    embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        emb = model.encode(batch, normalize_embeddings=True)
        embeddings.extend(emb.tolist())
    return embeddings

注意启用normalize_embeddings，余弦相似度检索时归一化向量可以省去计算分母步骤。

向量存储与检索引擎选型

向量数据库选型需考虑数据规模和查询延迟：

- FAISS：本地部署首选，10万级以下文档性能最佳，支持IVF+PQ索引压缩
- Milvus：分布式场景推荐，支持多副本和水平扩展，百万级数据qps稳定
- Chroma：快速原型验证用，API简洁但不适合生产环境
- Elasticsearch 8.x：已有ES集群可直接启用dense_vector字段，减少组件维护

生产环境推荐Milvus 2.x，搭建示例如下：

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

connections.connect(host="localhost", port="19530")

# 定义Collection
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=4096),
    FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512),
]
schema = CollectionSchema(fields, description="enterprise_knowledge_base")
collection = Collection("kb_vectors", schema)

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

检索时采用混合检索策略——向量相似度检索+关键词BM25检索加权融合，能将Top-5召回率提升15%-25%。

Prompt工程与模型调用模块

RAG场景下的Prompt模板要明确区分"检索到的上下文"和"用户问题"，避免模型混淆信息来源。推荐的模板结构：

RAG_PROMPT_TEMPLATE = "你是一个专业的知识助手。请严格根据以下参考资料回答用户问题。如果参考资料中没有相关信息，请明确说明无法回答，不要编造内容。

参考资料：
{context}

用户问题：{question}

回答："

模型调用建议使用流式输出，配合SSE推送到前端提升用户体验。OpenAI兼容接口的调用方式：

import openai

client = openai.OpenAI(
    api_key="your-api-key",
    base_url="https://api.example.com/v1"
)

def generate_answer(query, retrieved_docs):
    context = "\n\n".join([f"[文档{i+1}] {doc}" 
                             for i, doc in enumerate(retrieved_docs)])
    
    response = client.chat.completions.create(
        model="your-model-name",
        messages=[{
            "role": "user",
            "content": RAG_PROMPT_TEMPLATE.format(
                context=context, question=query
            )
        }],
        temperature=0.1,
        stream=True
    )
    
    for chunk in response:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

temperature设为0.1是因为RAG场景下需要精确回答，降低随机性。max_tokens根据业务需求设置，一般2048够用。

结果后处理与评估体系

RAG系统的输出需要两层校验：

第一层：事实性校验。把模型输出与检索到的原文进行比对，检测是否存在幻觉。简单做法是用n-gram重叠度做快速过滤，重叠度低于30%的输出标记为"待人工审核"。

第二层：完整性校验。检查回答是否覆盖了用户问题的所有子问题。把用户问题拆解为子问题列表，逐条确认回答是否涵盖。

评估指标用RAGAS框架，重点关注三个维度：

- Faithfulness（忠实度）：回答是否严格基于检索到的上下文
- Answer Relevancy（相关性）：回答与用户问题的相关程度
- Context Precision（上下文精度）：检索到的文档片段中有效信息的比例

# RAGAS评估示例
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

results = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_precision]
)
print(results)

生产环境部署清单

一套完整的RAG系统上线前，需要确认以下配置项：

- 向量数据库配置nlist参数，10万文档建议nlist=128，100万文档建议nlist=256
- 嵌入模型部署用vLLM或TEI（Text Embeddings Inference），批量请求时吞吐量提升3-5倍
- 检索模块设置超时阈值，向量检索200ms内返回，超时降级为关键词检索
- 知识库增量更新机制：新文档入库后走全流程，向量写入后立即生效
- 日志记录每次检索的query、top-k文档ID、模型输出，用于后续分析检索质量

RAG系统的调优是持续过程。上线后通过日志分析低分query的模式，迭代优化分块策略和Prompt模板，逐步提升系统的可用性和准确性。