【硬核实战】基于腾讯云混元Embedding+向量数据库构建企业级RAG知识助手（附完整Python源码与踩坑记录）

pip install tencentcloud-sdk-python>=3.0.1050  # 注意：低于此版本不支持混元Embedding
pip install tcvectordb  # 腾讯云向量数据库官方SDK

import ssl
import sys
if sys.platform == 'win32':
    ssl._create_default_https_context = ssl._create_unverified_context

from tencentcloud.common import credential
from tencentcloud.hunyuan.v20230901 import hunyuan_client, models

# 初始化混元客户端
cred = credential.Credential("你的SecretId", "你的SecretKey")
client = hunyuan_client.HunyuanClient(cred, "ap-guangzhou")

# 初始化向量数据库客户端
from tcvectordb import VectorDBClient
vdb_client = VectorDBClient(
    url="http://你的实例地址",
    username="root",
    key="你的API密钥",
    timeout=30
)

import json, requests
def get_embedding(text_list):
    """批量获取向量，单次最多支持 512 条"""
    url = "https://hunyuan.tencentcloudapi.com/"
    payload = {
        "Input": text_list,  # 注意：这里传入的是列表
        "Model": "hunyuan-embedding"
    }
    # 签名逻辑省略（建议使用SDK的CommonClient），此处展示核心参数
    resp = requests.post(url, headers=headers, json=payload)
    return [item["Embedding"] for item in resp.json()["Response"]["Data"]]

from tcvectordb.model.enum import FieldType, IndexType, MetricType
from tcvectordb.model.index import Index, VectorIndex, FilterIndex

# 创建索引
index = Index(
    FilterIndex(name="id", field_type=FieldType.String, index_type=IndexType.PRIMARY_KEY),
    FilterIndex(name="department", field_type=FieldType.String, index_type=IndexType.FILTER),  # 用于标量过滤
    VectorIndex(name="vector", dimension=1024, index_type=IndexType.HNSW,
                metric_type=MetricType.COSINE, params={"M": 32, "efConstruction": 200})
)
vdb_client.create_collection(database_name="kb_db", collection_name="doc_chunks", index=index)

def search_knowledge(query, department, top_k=5):
    # 1. 问题向量化
    q_vector = get_embedding([query])[0]
    # 2. 向量检索 + Filter 过滤
    results = vdb_client.search(
        database_name="kb_db",
        collection_name="doc_chunks",
        vectors=[q_vector],
        limit=top_k,
        filter=f'department="{department}"'  # 关键：标量过滤减少干扰
    )
    return [doc["text"] for doc in results[0]]

def build_prompt(context_list, user_query):
    context_str = "\n\n".join([f"[文档片段{i+1}]: {c}" for i, c in enumerate(context_list)])
    return {
        "system": "你是一个严谨的技术助手。仅根据提供的文档片段回答。",
        "user": f"文档片段：\n{context_str}\n\n问题：{user_query}"
    }

from tencentcloud.hunyuan.v20230901 import models

def chat_stream(system, user):
    req = models.ChatCompletionsRequest()
    req.Model = "hunyuan-lite"  # 追求性价比选lite，复杂逻辑选pro
    req.Messages = [
        {"Role": "system", "Content": system},
        {"Role": "user", "Content": user}
    ]
    req.Stream = True  # 开启流式
    resp = client.ChatCompletions(req)
    for event in resp:
        # 解析SSE事件
        print(event["Choices"][0]["Delta"]["Content"], end="")

import time
from functools import wraps

def retry_on_throttle(max_retries=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if "LimitExceeded" in str(e):
                        time.sleep(2 ** i)  # 退避
                        continue
                    raise e
        return wrapper
    return decorator