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

作者：袖梨 2026-08-15

手把手教你基于腾讯云向量数据库和大模型构建企业级 RAG 应用的重点在于把前置条件、操作顺序和容易误判的地方分清楚。

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

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

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

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

腾讯云向量数据库（Tencent Cloud VectorDB） —— 全托管，百万级 QPS，毫秒召回腾讯混元大模型（Hunyuan）或开源 Qwen2.5（二选一，本文以混元 API 为例）LangChain 0.3 作为编排框架FastAPI 提供 HTTP 服务

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

二、系统架构与核心指标

代码语言：javascript

复制

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

设计目标：

召回延迟 ≤ 80ms（P95）首 Token 延迟 ≤ 1.2s准确率（Hit@5）≥ 92%（基于私有文档集）

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

3.1 向量数据库实例创建

登录腾讯云控制台 → 向量数据库 VectorDB → 新建实例选择性能型，副本数 2，分片数 2（生产环境建议 4 分片）获取 Endpoint 和 API Key（内网访问免流量费）

3.2 混元大模型 API 开通

进入腾讯混元大模型开通服务获取 SecretId 和 SecretKey本文使用 hunyuan-lite 模型（性价比极高，128K 上下文）

3.3 Python 环境

代码语言：javascript

复制

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 文档切片策略（非简单按字符截断）

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

代码语言：javascript

复制

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 接口（无需额外调用），也可自定义向量。本文使用显式生成批量写入以最大化吞吐：

代码语言：javascript

复制

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）和重排序模型：

代码语言：javascript

复制

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）

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

代码语言：javascript

复制

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 压缩保留关键信息：

代码语言：javascript

复制

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 模板（带引用溯源）

代码语言：javascript

复制

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 异步）

代码语言：javascript

复制

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)