手把手教你基于腾讯云向量数据库和大模型构建企业级 RAG 应用并不只看表面做法，关键还要理解相关条件、限制和后续影响。

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

一、为什么你需要掌握这一代 AI 工具链

2026 年，大模型已从“聊天玩具”进化为核心生产工具。但现实场景中，幻觉、知识滞后、私有数据无法接入三大痛点仍未根本解决。检索增强生成（RAG） 成为工业界公认的最优解。

下文会带你硬核落地一套完整 RAG 系统，技术栈选用：

全文所有代码均可在腾讯云轻量服务器上直接运行，附性能调优参数。

二、系统架构与核心指标

复制

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

设计目标：

三、环境准备与腾讯云资源开通

3.1 向量数据库实例创建

3.2 混元大模型 API 开通

3.3 Python 环境

复制

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

四、数据预处理与索引构建（核心代码）

4.1 文档切片策略（非简单按字符截断）

采用 语义分块 重叠滑动窗口，保留上下文连续性：

复制

from langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import PyPDFLoaderdef load_and_chunk(file_path: str, chunk_size: int = 512, overlap: int = 50):loader = PyPDFLoader(file_path)docs = loader.load()# 使用递归分隔符（保留段落结构）splitter = RecursiveCharacterTextSplitter(separators=["", "", "。", "！", "？", "；", "，", " ", ""],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"] = ichunk.metadata["source"] = file_path.split("/")[-1]return chunks

4.2 生成 Embedding 并写入向量库

腾讯云 VectorDB 支持 自带 Embedding 接口（无需额外调用），也可自定义向量。本文使用 显式生成 批量写入 以最大化吞吐：

复制

import tcvectordbfrom tcvectordb.model.enum import FieldType, IndexType, MetricTypefrom tcvectordb.model.index import Index, VectorIndex, FilterIndexfrom tencentcloud.common import credentialfrom 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 = batchreq.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 = 100for 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")

五、检索链路优化（非简单 Top-K）

5.1 混合检索 重排序（Rerank）

仅靠向量相似度可能召回噪声，加入 关键词匹配（BM25） 和 重排序模型：

复制

from tcvectordb.model.document import SearchParamsdef 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_kreturn candidates[:top_k]

5.2 Query 改写（Query Rewriting）

用户口语化问题往往包含指代不清，使用大模型改写为独立检索式：

复制

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()

六、Prompt 工程与生成增强

6.1 上下文压缩（Context Compression）

检索到的 Top-5 可能总 token 超过模型窗口，采用 LLMLingua 压缩保留关键信息：

复制

from llmlingua import PromptCompressorcompressor = 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 = "---".join(documents)compressed = compressor.compress_prompt(combined,ratio=target_ratio,condition_in_question="",rank_method="longllmlingua")return compressed["compressed_prompt"]

6.2 生成 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.8req.Stream = Falseresp = client.ChatCompletions(req)answer = resp.Choices[0].Message.Content# 提取引用 ID（用于前端展示）return {"answer": answer,"references": [{"text": t[:100] "..."} for t in retrieved_texts]}

七、服务化部署（FastAPI 异步）

复制

from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelimport asynciofrom concurrent.futures import ThreadPoolExecutorapp = FastAPI(title="RAG Service", version="2.0")executor = ThreadPoolExecutor(max_workers=8)class QueryRequest(BaseModel):query: strtop_k: int = 5rewrite: bool = Trueclass QueryResponse(BaseModel):answer: strreferences: list[dict]latency_ms: [email protected]("/rag", response_model=QueryResponse)async def rag_endpoint(req: QueryRequest):import timestart = 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) * 1000return 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 uvicornuvicorn.run(app, host="0.0.0.0", port=8080, workers=4)

八、压测与调优结论（真实数据）

在腾讯云 SA5 实例（8C 32G）上，压测 1000 并发请求：

关键调优参数：

九、总结与扩展方向

本文完整实现了从 数据清洗 → 向量索引 → 混合检索 → Prompt 压缩 → 服务化 的全链路，所有代码已在腾讯云向量数据库 混元环境验证。

后续可扩展：

新一代 AI 工具不是黑盒，而是可组合、可调优的基础设施。希望本文能帮你越过“纸上谈兵”，真正将 RAG 落地到业务中。