手把手教你基于腾讯云向量数据库和大模型构建企业级 RAG 应用

┌─────────────┐     ┌──────────────┐     ┌─────────────────┐
│  用户 Query  │────▶│  Query 改写  │────▶│  Embedding 编码  │
└─────────────┘     └──────────────┘     └────────┬────────┘
                                                   ▼
                                          ┌─────────────────┐
                                          │ 向量检索 (Top-K) │
                                          └────────┬────────┘
                                                   ▼
┌─────────────┐     ┌──────────────┐     ┌─────────────────┐
│  最终回复    │◀────│  LLM 生成    │◀────│  上下文拼接 +   │
└─────────────┘     └──────────────┘     │  Prompt 工程   │
                                          └─────────────────┘

python3.10 -m venv rag_env
source rag_env/bin/activate
pip install -U langchain langchain-community tcvectordb tencentcloud-sdk-python fastapi uvicorn pypdf tiktoken

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader

def load_and_chunk(file_path: str, chunk_size: int = 512, overlap: int = 50):
    loader = PyPDFLoader(file_path)
    docs = loader.load()
    
    # 使用递归分隔符（保留段落结构）
    splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        length_function=len,
        add_start_index=True,
    )
    chunks = splitter.split_documents(docs)
    
    # 附加元数据（文件来源、页码、时间戳）
    for i, chunk in enumerate(chunks):
        chunk.metadata["chunk_id"] = i
        chunk.metadata["source"] = file_path.split("/")[-1]
    return chunks

import tcvectordb
from tcvectordb.model.enum import FieldType, IndexType, MetricType
from tcvectordb.model.index import Index, VectorIndex, FilterIndex
from tencentcloud.common import credential
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models

# 初始化混元 Embedding 客户端
cred = credential.Credential("YOUR_SECRET_ID", "YOUR_SECRET_KEY")
client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou")

def get_embeddings(texts: list[str], batch_size=16) -> list[list[float]]:
    """批量获取 Embedding，支持重试"""
    all_vecs = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        req = models.GetEmbeddingRequest()
        req.Input = batch
        req.Model = "hunyuan-embedding"  # 官方 embedding 模型
        resp = client.GetEmbedding(req)
        vecs = [item.Embedding for item in resp.Data]
        all_vecs.extend(vecs)
    return all_vecs

# 连接腾讯云向量数据库
vdb_client = tcvectordb.VectorDBClient(
    url="http://your-endpoint.vectordb.tencentcloudapi.com",
    username="root",
    key="YOUR_API_KEY",
    timeout=30
)

# 创建数据库和集合（若不存在）
db = vdb_client.create_database("rag_db")
coll = db.create_collection(
    name="doc_chunks",
    shard=2,
    replicas=2,
    indexes=[
        VectorIndex("vector", Dimension=1024, IndexType=IndexType.HNSW,
                    MetricType=MetricType.COSINE, Params={"M": 16, "efConstruction": 200}),
        FilterIndex("chunk_id", FieldType.String, IndexType.PRIMARY_KEY),
        FilterIndex("source", FieldType.String, IndexType.FILTER),
        FilterIndex("page_num", FieldType.Uint64, IndexType.FILTER),
    ]
)

# 批量写入（每批 100 条，利用 upsert）
def index_documents(chunks):
    batch_size = 100
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i+batch_size]
        texts = [c.page_content for c in batch]
        vectors = get_embeddings(texts)
        
        docs = []
        for chunk, vec in zip(batch, vectors):
            docs.append({
                "chunk_id": chunk.metadata["chunk_id"],
                "vector": vec,
                "text": chunk.page_content,
                "source": chunk.metadata["source"],
                "page_num": chunk.metadata.get("page", 0),
                "metadata": str(chunk.metadata)  # 冗余存储便于调试
            })
        coll.upsert(docs)
        print(f"Indexed {i+len(batch)} / {len(chunks)} chunks")

from tcvectordb.model.document import SearchParams

def hybrid_search(query: str, top_k: int = 10) -> list[dict]:
    # 1. 向量检索
    query_vec = get_embeddings([query])[0]
    vec_results = coll.search(
        vectors=[query_vec],
        filter=None,
        params=SearchParams(ef=200),  # HNSW 检索参数
        limit=top_k * 2,
        retrieve_vector=False,
        output_fields=["text", "source", "chunk_id"]
    )
    
    # 2. 获取候选文本
    candidates = []
    for res in vec_results[0]:
        candidates.append({
            "text": res["text"],
            "score": res["score"],
            "chunk_id": res["chunk_id"]
        })
    
    # 3. 使用 bge-reranker-v2-m3 进行精排（本地部署或调用 API）
    # 此处简化：按向量得分降序，实际可叠加 BM25 线性加权
    # 为演示，直接取 top_k
    return candidates[:top_k]

def rewrite_query(original: str, history: list[str] = None) -> str:
    prompt = f"""你是一个搜索专家。将用户问题改写为更清晰、适合向量检索的表述，保持核心实体。
    原问题：{original}
    改写后："""
    # 调用混元生成
    req = models.ChatCompletionsRequest()
    req.Model = "hunyuan-lite"
    req.Messages = [{"Role": "user", "Content": prompt}]
    resp = client.ChatCompletions(req)
    return resp.Choices[0].Message.Content.strip()

from llmlingua import PromptCompressor

compressor = PromptCompressor(
    model_name="microsoft/llmlingua-2-bert-base-multilingual-cased",
    device_map="cpu"  # 可选 cuda
)

def compress_context(documents: list[str], target_ratio=0.5) -> str:
    combined = "\n---\n".join(documents)
    compressed = compressor.compress_prompt(
        combined,
        ratio=target_ratio,
        condition_in_question="",
        rank_method="longllmlingua"
    )
    return compressed["compressed_prompt"]

SYSTEM_TEMPLATE = """你是一个专业的知识助手。根据以下参考文档回答用户问题。
如果无法从文档中找到答案，请明确说"资料中未涉及"，不要编造。
回答时请用 [1] [2] 格式标注信息来源。

参考文档：
{context}

用户问题：{question}
回答："""

def generate_response(query: str, retrieved_texts: list[str]) -> dict:
    compressed = compress_context(retrieved_texts, target_ratio=0.6)
    prompt = SYSTEM_TEMPLATE.format(context=compressed, question=query)
    
    req = models.ChatCompletionsRequest()
    req.Model = "hunyuan-lite"
    req.Messages = [{"Role": "user", "Content": prompt}]
    req.Temperature = 0.1  # 降低随机性
    req.TopP = 0.8
    req.Stream = False
    resp = client.ChatCompletions(req)
    answer = resp.Choices[0].Message.Content
    
    # 提取引用 ID（用于前端展示）
    return {
        "answer": answer,
        "references": [{"text": t[:100] + "..."} for t in retrieved_texts]
    }

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
from concurrent.futures import ThreadPoolExecutor

app = FastAPI(title="RAG Service", version="2.0")
executor = ThreadPoolExecutor(max_workers=8)

class QueryRequest(BaseModel):
    query: str
    top_k: int = 5
    rewrite: bool = True

class QueryResponse(BaseModel):
    answer: str
    references: list[dict]
    latency_ms: float

@app.post("/rag", response_model=QueryResponse)
async def rag_endpoint(req: QueryRequest):
    import time
    start = time.perf_counter()
    
    # 异步执行（避免阻塞）
    loop = asyncio.get_event_loop()
    
    # 1. Query 改写（可选）
    if req.rewrite:
        rewritten = await loop.run_in_executor(executor, rewrite_query, req.query)
    else:
        rewritten = req.query
    
    # 2. 混合检索
    docs = await loop.run_in_executor(executor, hybrid_search, rewritten, req.top_k)
    texts = [d["text"] for d in docs]
    
    # 3. 生成回复
    result = await loop.run_in_executor(executor, generate_response, req.query, texts)
    
    elapsed = (time.perf_counter() - start) * 1000
    return QueryResponse(
        answer=result["answer"],
        references=result["references"],
        latency_ms=round(elapsed, 2)
    )

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

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080, workers=4)