从零搭建企业级RAG系统：基于LLM的智能问答实战

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  文档上传    │────▶│  文档解析    │────▶│  文本分块    │
└─────────────┘     └─────────────┘     └──────┬──────┘
                                                 ▼
┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  用户提问    │────▶│  向量检索    │◀────│  向量嵌入    │
└─────────────┘     └──────┬──────┘     │  + 向量库   │
                            │              └─────────────┘
                            ▼
                   ┌─────────────┐
                   │  上下文组装  │
                   │ + Prompt    │
                   └──────┬──────┘
                            ▼
                   ┌─────────────┐
                   │  LLM生成    │
                   │  (流式)     │
                   └─────────────┘

# 创建虚拟环境
python -m venv rag_env
source rag_env/bin/activate  # Windows: rag_env\Scripts\activate

# 核心依赖
pip install fastapi uvicorn python-multipart
pip install langchain langchain-community langchain-text-splitters
pip install chromadb sentence-transformers
pip install pypdf python-docx markdown
pip install tencentcloud-sdk-python  # 腾讯云SDK
pip install tiktoken

rag_system/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI入口
│   ├── config.py            # 配置管理
│   ├── document_loader.py   # 文档解析与分块
│   ├── embedding.py         # 嵌入模型封装
│   ├── vector_store.py      # 向量库操作
│   ├── retriever.py         # 检索与重排
│   ├── generator.py         # LLM生成（含流式）
│   └── schemas.py           # Pydantic模型
├── data/                    # 原始文档存放
├── scripts/                 # 初始化脚本
└── requirements.txt

import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # 向量库配置
    vector_db_type: str = "chroma"  # 或 "tencent"
    chroma_persist_dir: str = "./chroma_db"
    collection_name: str = "medical_kb"
    
    # 嵌入模型
    embedding_model: str = "BAAI/bge-large-zh-v1.5"
    embedding_device: str = "cpu"  # 或 "cuda"
    
    # 腾讯云混元配置（若使用）
    tencent_secret_id: str = os.getenv("TENCENT_SECRET_ID", "")
    tencent_secret_key: str = os.getenv("TENCENT_SECRET_KEY", "")
    hunyuan_model: str = "hunyuan-lite"
    
    # 检索参数
    top_k: int = 5
    score_threshold: float = 0.7
    chunk_size: int = 512
    chunk_overlap: int = 50
    
    # 服务
    api_port: int = 8000

settings = Settings()

import os
import re
from typing import List
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, UnstructuredMarkdownLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

class DocumentProcessor:
    def __init__(self, chunk_size=512, chunk_overlap=50):
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
            length_function=len,
        )
    
    def load_document(self, file_path: str) -> List[Document]:
        ext = os.path.splitext(file_path)[1].lower()
        if ext == ".pdf":
            loader = PyPDFLoader(file_path)
        elif ext == ".docx":
            loader = Docx2txtLoader(file_path)
        elif ext == ".md":
            loader = UnstructuredMarkdownLoader(file_path)
        elif ext == ".txt":
            # 手动读取
            with open(file_path, "r", encoding="utf-8") as f:
                text = f.read()
            return [Document(page_content=text, metadata={"source": file_path})]
        else:
            raise ValueError(f"Unsupported file type: {ext}")
        return loader.load()
    
    def split_documents(self, docs: List[Document]) -> List[Document]:
        """分块并添加元数据（如段落标题推断）"""
        chunks = self.text_splitter.split_documents(docs)
        # 为每个chunk增加索引和来源
        for i, chunk in enumerate(chunks):
            chunk.metadata["chunk_id"] = i
            chunk.metadata["source_file"] = os.path.basename(chunk.metadata.get("source", "unknown"))
        return chunks
    
    def process_file(self, file_path: str) -> List[Document]:
        docs = self.load_document(file_path)
        return self.split_documents(docs)

from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List

class LocalEmbedding:
    def __init__(self, model_name: str = "BAAI/bge-large-zh-v1.5", device: str = "cpu"):
        self.model = SentenceTransformer(model_name, device=device)
        # BGE模型需要添加指令前缀以获得更好效果
        self.query_prefix = "为这个句子生成表示以用于检索相关文章："
    
    def encode(self, texts: List[str], is_query: bool = False) -> np.ndarray:
        if is_query:
            texts = [self.query_prefix + t for t in texts]
        return self.model.encode(texts, normalize_embeddings=True)  # 归一化便于余弦相似度
    
    @property
    def dimension(self) -> int:
        return self.model.get_sentence_embedding_dimension()

# embedding_tencent.py 片段
from tencentcloud.common import credential
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models
import json

class TencentEmbedding:
    def __init__(self, secret_id, secret_key):
        cred = credential.Credential(secret_id, secret_key)
        self.client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou")
    
    def encode(self, texts: List[str]) -> List[List[float]]:
        req = models.GetEmbeddingRequest()
        req.Input = texts
        resp = self.client.GetEmbedding(req)
        return [item.Embedding for item in resp.Data]

import chromadb
from chromadb.config import Settings as ChromaSettings
from app.config import settings
from app.embedding import LocalEmbedding
import uuid

class VectorStore:
    def __init__(self):
        self.embedder = LocalEmbedding(
            model_name=settings.embedding_model,
            device=settings.embedding_device
        )
        self.client = chromadb.PersistentClient(
            path=settings.chroma_persist_dir,
            settings=ChromaSettings(anonymized_telemetry=False)
        )
        self.collection = self.client.get_or_create_collection(
            name=settings.collection_name,
            metadata={"hnsw:space": "cosine"}  # 余弦距离
        )
    
    def add_documents(self, documents: List[dict]) -> List[str]:
        """
        documents: [{"content": str, "metadata": dict}]
        返回插入的ID列表
        """
        ids = [str(uuid.uuid4()) for _ in documents]
        texts = [doc["content"] for doc in documents]
        metadatas = [doc.get("metadata", {}) for doc in documents]
        
        # 生成向量
        embeddings = self.embedder.encode(texts, is_query=False)
        
        self.collection.add(
            ids=ids,
            embeddings=embeddings.tolist(),
            documents=texts,
            metadatas=metadatas
        )
        return ids
    
    def search(self, query: str, top_k: int = 5) -> List[dict]:
        """返回最相似的文档片段"""
        query_vec = self.embedder.encode([query], is_query=True)
        results = self.collection.query(
            query_embeddings=query_vec.tolist(),
            n_results=top_k,
            include=["documents", "metadatas", "distances"]
        )
        # 组装返回
        items = []
        for i in range(len(results["ids"][0])):
            items.append({
                "id": results["ids"][0][i],
                "content": results["documents"][0][i],
                "metadata": results["metadatas"][0][i],
                "score": 1 - results["distances"][0][i]  # 余弦相似度
            })
        return items
    
    def delete_collection(self):
        self.client.delete_collection(settings.collection_name)

from app.vector_store import VectorStore
from typing import List, Tuple
import numpy as np

class AdvancedRetriever:
    def __init__(self, vector_store: VectorStore):
        self.vector_store = vector_store
    
    def retrieve(self, query: str, top_k: int = 5, use_mmr: bool = True, diversity: float = 0.3) -> List[dict]:
        """
        MMR: 在相关性和多样性之间平衡，避免重复内容。
        """
        # 先获取更多候选（top_k * 2）
        candidates = self.vector_store.search(query, top_k=top_k * 2)
        if not candidates or not use_mmr:
            return candidates[:top_k]
        
        # 计算候选向量（重新获取embedding，实际可缓存）
        from app.embedding import LocalEmbedding
        embedder = LocalEmbedding()
        candidate_texts = [c["content"] for c in candidates]
        cand_embs = embedder.encode(candidate_texts, is_query=False)
        query_emb = embedder.encode([query], is_query=True)
        
        # MMR选择
        selected_indices = []
        remaining = set(range(len(candidates)))
        # 第一轮选择与query最相似的
        sim_to_query = np.dot(cand_embs, query_emb.T).flatten()
        first = int(np.argmax(sim_to_query))
        selected_indices.append(first)
        remaining.remove(first)
        
        while len(selected_indices) < top_k and remaining:
            best_score = -np.inf
            best_idx = -1
            for idx in remaining:
                # 与query相似度
                sim_q = sim_to_query[idx]
                # 与已选集合的最大相似度（惩罚项）
                max_sim_sel = max([np.dot(cand_embs[idx], cand_embs[s]) for s in selected_indices])
                mmr_score = sim_q - diversity * max_sim_sel
                if mmr_score > best_score:
                    best_score = mmr_score
                    best_idx = idx
            selected_indices.append(best_idx)
            remaining.remove(best_idx)
        
        return [candidates[i] for i in selected_indices]

import json
import asyncio
from typing import AsyncGenerator, List
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models
from app.config import settings
from app.retriever import AdvancedRetriever

class StreamGenerator:
    def __init__(self, retriever: AdvancedRetriever):
        self.retriever = retriever
        # 初始化腾讯混元客户端
        from tencentcloud.common import credential
        cred = credential.Credential(settings.tencent_secret_id, settings.tencent_secret_key)
        self.client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou")
    
    def build_prompt(self, query: str, contexts: List[dict]) -> str:
        """构建包含上下文的提示词，加入引用标记"""
        context_text = "\n\n".join([
            f"[{i+1}] {ctx['content']}" for i, ctx in enumerate(contexts)
        ])
        prompt = f"""你是一个专业的医疗知识助手，请基于以下参考资料回答用户问题。如果参考资料不足以回答，请明确告知。
参考资料：
{context_text}

用户问题：{query}
请给出准确、简洁的回答，并在回答末尾标注引用的参考资料编号（如[1]）。"""
        return prompt
    
    async def generate_stream(self, query: str) -> AsyncGenerator[str, None]:
        # 1. 检索
        contexts = self.retriever.retrieve(query, top_k=settings.top_k)
        if not contexts:
            yield json.dumps({"error": "未找到相关知识"}) + "\n"
            return
        
        # 2. 构建prompt
        prompt = self.build_prompt(query, contexts)
        
        # 3. 调用混元流式API
        req = models.ChatCompletionsRequest()
        req.Model = settings.hunyuan_model
        req.Messages = [{"Role": "user", "Content": prompt}]
        req.Stream = True
        req.Temperature = 0.3
        req.TopP = 0.9
        
        resp = self.client.ChatCompletions(req)
        # 处理流式响应
        for event in resp:
            if event.data:
                data = json.loads(event.data)
                if "Choices" in data and len(data["Choices"]) > 0:
                    delta = data["Choices"][0].get("Delta", {})
                    content = delta.get("Content", "")
                    if content:
                        # 以SSE格式输出
                        yield f"data: {json.dumps({'content': content, 'source': [c['metadata'] for c in contexts]})}\n\n"
        yield "data: [DONE]\n\n"

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import StreamingResponse, JSONResponse
from pydantic import BaseModel
from app.document_loader import DocumentProcessor
from app.vector_store import VectorStore
from app.retriever import AdvancedRetriever
from app.generator import StreamGenerator
from app.config import settings
import tempfile
import os

app = FastAPI(title="医疗知识库RAG系统", version="1.0")

# 初始化组件
vector_store = VectorStore()
retriever = AdvancedRetriever(vector_store)
generator = StreamGenerator(retriever)
processor = DocumentProcessor(
    chunk_size=settings.chunk_size,
    chunk_overlap=settings.chunk_overlap
)

class QueryRequest(BaseModel):
    query: str
    top_k: int = settings.top_k

@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """上传并索引文档"""
    try:
        # 保存临时文件
        suffix = os.path.splitext(file.filename)[1]
        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
            content = await file.read()
            tmp.write(content)
            tmp_path = tmp.name
        
        # 解析与分块
        chunks = processor.process_file(tmp_path)
        # 准备存储格式
        docs_to_add = [
            {"content": chunk.page_content, "metadata": chunk.metadata}
            for chunk in chunks
        ]
        ids = vector_store.add_documents(docs_to_add)
        os.unlink(tmp_path)
        return {"status": "success", "chunks": len(chunks), "ids": ids}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/query")
async def query(request: QueryRequest):
    """非流式查询（返回完整结果）"""
    contexts = retriever.retrieve(request.query, top_k=request.top_k)
    if not contexts:
        return JSONResponse({"answer": "未找到相关知识", "sources": []})
    # 使用同步生成（省略，实际可调用非流式API）
    # 此处简单拼接
    prompt = generator.build_prompt(request.query, contexts)
    # 实际调用混元非流式...
    return {"prompt": prompt, "contexts": contexts}  # 示意

@app.post("/stream")
async def stream_query(request: QueryRequest):
    """流式问答接口"""
    return StreamingResponse(
        generator.generate_stream(request.query),
        media_type="text/event-stream"
    )

@app.get("/health")
async def health():
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=settings.api_port)

# 伪代码集成
from rank_bm25 import BM25Okapi
def hybrid_search(query, alpha=0.5):
    vec_results = vector_search(query)
    bm25_results = bm25_search(query)
    # RRF融合
    ...

FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]