【腾讯云ES×AI实战】告别复杂！在一个技术栈内，基于混合搜索快速搭建RAG

# 腾讯云API客户端配置，获取方式见【3.获取API秘钥】
SECRET_ID = "AKID******"    # 替换为你的云API SecretId
SECRET_KEY = "******"       # 替换为你的云API SecretKey
 
# es集群相关配置，获取方式见【1.购买集群】【2.获取ES访问地址】
# 用户名为 elastic、密码在创建集群时设置。用本地环境测试时，可开启公网访问，实际生产时，建议使用内网访问地址。
ES_ENDPOINT = "https://es-******.tencentelasticsearch.com:9200"
ES_USERNAME = "elastic"  # Elasticsearch集群用户名
ES_PASSWORD = "******"  # Elasticsearch集群密码
INDEX_NAME = "索引名称"  # 在写入时要保证索引未被创建过，会默认创建索引并设置mapping，写入数据和在线问答的索引保持一致
 
# 数据入库时，选择COS方式需设置你的COS文件路径
# 要处理的文档URL（推荐将文件存放到COS存储，并开启「公有读私有写」权限)
document_url = "https://xxx.cos.xxx/sample.doc"

# Python版本：3.9及以上
pip install elasticsearch==8.16.0
pip install tencentcloud-sdk-python
pip install urllib3 certifi
# 安装文档处理包
pip install python-docx pypdf2 lxml
pip install streamlit    #使用COS数据入库时，无需安装

# 要处理的文档URL（推荐将文件存放到COS存储，并开启「公有读私有写」权限。
document_url = "https://xxx.cos.xxx/sample.doc"

python data_input_inference.py

streamlit run data_input_inference_streamlit.py

streamlit run search_inference.py

MAX_CHUNK_SIZE = 2000  # 文本切片单条文本最大字符个数（根据具体选用的Embedding模型tokens设定,最小值不小于1000）
DOCUMENT_CHUNK_MODEL_NAME = "doc-tree-chunk"
# 调用ChunkDocumentAsync接口创建文档切片任务
params = {
    "Document": {
        "FileType": file_type,
        "FileUrl": file_url,
        "FileName": os.path.basename(urlparse(file_url).path)
    },
    "Config": {
        "MaxChunkSize": MAX_CHUNK_SIZE  # 可根据需要调整切片大小
    },
    "ModelName": DOCUMENT_CHUNK_MODEL_NAME
}
response = retry_request(client.call_json, "ChunkDocumentAsync", params)
task_id = response['Response']['TaskId']
logger.info(f"Created document chunking task with ID: {task_id}")

try:
    with zipfile.ZipFile(io.BytesIO(response.content)) as zip_file:
        # 检查zip中是否有.jsonl文件
        jsonl_files = [f for f in zip_file.namelist() if f.endswith('.jsonl')]
        if not jsonl_files:
            error_msg = "No .jsonl file found in the zip archive"
            logger.error(error_msg)
            raise Exception(error_msg)
 
        # 处理每个jsonl文件
        for file_name in jsonl_files:
            with zip_file.open(file_name) as f:
                # 逐行读取jsonl文件
                for line in f:
                    try:
                        data = json.loads(line.decode('utf-8'))
                        if 'page_content' in data:
                            doc_list.append(data['page_content'])
                    except json.JSONDecodeError as e:
                        logger.warning(f"Failed to parse JSON line: {str(e)}")
                        continue

INFERENCE_MODEL_ID = "model_bge_base_zh-v1.5" 
es_client.perform_request(
        'PUT',
        f'/_inference/text_embedding/{INFERENCE_MODEL_ID}',
        body=inference_config
)

PIPELINE_NAME = "model_bge_base_zh-v1.5_embeddings_pipeline"
pipeline_config = {
    "processors": [
        {
            "inference": {
                "model_id": INFERENCE_MODEL_ID,
                "input_output": {
                    "input_field": "content",
                    "output_field": "content_embedding"
                }
            }
        }
    ]
}
 
es_client.ingest.put_pipeline(
        id=PIPELINE_NAME,
        body=pipeline_config
)

from elasticsearch.helpers import bulk
actions = [
    {
        "_index": INDEX_NAME,
        "_source": {
            "content": content,  # 原始文本内容，根据定义的pipeline，默认向量化后写入pipeline中定义的向量字段
        }
    }
    for index, content in enumerate(chunk_list)
]
success, failed = bulk(
        es_client,
        actions,
        pipeline=PIPELINE_NAME
)

# 使用8.16集群的混合搜索语法(并行召回：返回 query 和 knn 召回结果的并集)
mix_query = {
    "query": {"match": {"content": query}},
    "knn": [{
        "field": "content_embedding",
        "query_vector_builder": {
            "text_embedding": {
                "model_id": INFERENCE_MODEL_ID,
                "model_text": query
            }
        },
        "k": 5,
        "num_candidates": 100
    }],
    "size": 5
}

RERANK_MODEL_NAME = "bge-reranker-large"
json_object = {
    "ModelName": RERANK_MODEL_NAME,
    "Query": query,
    "ReturnDocuments": True,
    "Documents": docs,
    "TopN":5    # 排序返回的top文档数量, 如果没有指定则返回全部候选doc，如果指定的TopN值大于输入的候选doc数量，返回全部doc
}

PROMPT_TEMPLATE = """
# 任务说明
你是一位专业的AI助手，需要基于提供的参考文档准确回答用户问题。请严格遵循以下要求：
 
# 用户问题
{user_query}
 
# 参考文档（已按相关性排序）
{documents_str}
 
# 回答要求
1. 必须基于上述参考文档内容生成答案，不得编造文档中不存在的信息
2. 回答结构应包含：
   - 简明扼要的核心答案（1-2句话）
   - 详细解释或分点说明（3-5点，视问题复杂度而定）
   - 可选的补充信息或建议（如适用）
3. 根据问题类型采用合适的回答方式：
   - 对于事实性问题：提供准确数据或事实
   - 对于观点性问题：呈现不同角度的分析
   - 对于操作性问题：给出步骤清晰的指导
   - 对于比较性问题：进行客观对比分析
4. 若文档内容与问题无关或信息不足，请回复：
   "根据现有资料，我暂时无法提供令人满意的答案。建议您补充更多相关信息或重新提问。"
 
# 特别注意
1. 保持回答专业、准确且流畅
2. 如文档中有数据或研究结果，请优先引用
4. 对于专业术语，必要时提供简单解释
5. 避免使用"根据文档"等冗余表述，直接整合信息到回答中
"""

# 设置deepseek-v3模型（这里可以设置为你想要的模型，参考：https://cloud.tencent.com/document/api/845/117810）
LLM_MODEL_NAME = "deepseek-v3"
 
# 调用大模型回答，content即为build_llm_prompt返回组装后的prompt
def generate_llm_response(content: str) -> Dict[str, Any]:
    """
    调用大模型生成回答
 
    Args:
        content: 要发送给大模型的prompt内容
 
    Returns:
        大模型的响应结果
 
    Raises:
        Exception: 如果调用大模型过程中出现错误
    """
    if not content.strip():
        raise ValueError("空内容无法生成回答")
 
    json_object = {
        "Messages": [{"Role": "user", "Content": content}],
        "ModelName": LLM_MODEL_NAME,
        "Stream": False    # 可以设置成True，流式输出
    }
 
    try:
        response = tencent_cloud_client.call_json("ChatCompletions", json_object)
        return response
    except Exception as e:
        logger.error(f"大模型调用失败: {e}")
        raise