基于腾讯云ES混合搜索与DeepSeek，十分钟构建RAG 应用

点击蓝字⬆ 关注我们

随着数据智能技术的不断发展，以大语言模型（LLM）驱动的AIGC为代表的内容生成技术已经成为企业数据智能能力中不可或缺的一部分，但传统的内容生成技术存在信息更新不及时、垂直领域知识匮乏、模型幻觉等问题，如何推进大模型在各行业、各业务场景落地是各方普遍关注的问题，而检索增强生成（Retrieval-Augmented Generation，RAG）技术则为此提供了有效的解决方案，成为数据智能时代的一大趋势。

RAG是一种结合了检索和大语言模型内容生成的技术方案，它通过引用外部知识库，在用户输入Query时检索出知识，然后让模型基于可信的知识进行用户回答。RAG具有较高的可解释性和定制能力，可大幅降低大语言模型的幻觉，适用于问答系统、文档生成、智能助手等多种自然语言处理任务。本文将通过介绍腾讯云ES 一站式 RAG 方案，演示如何通过结合腾讯云 ES 与 DeepSeek 大模型，快速构建 RAG 应用。

腾讯云ES是云端全托管海量数据检索分析服务，拥有高性能自研内核，集成X-Pack，支持通过自治索引、存算分离、集群巡检等特性轻松管理集群，也支持免运维、自动弹性、按需使用的 Serverless 模式。在自研内核方面，腾讯云ES依托腾讯内外部海量业务的运营经验，针对ES内核进行了成本、性能、稳定性、拓展性等方面的优化，是全球第三方企业开源贡献第一的团队，使用腾讯云ES可以高效构建在线搜索、向量检索、日志分析、运维监控、智能问答等服务。

在RAG方面，腾讯云ES支持了一站式向量检索、文本+向量混合搜索、倒数排序融合、与大模型集成、GPU高性能推理、字段级别权限控制等能力，同时针对查询性能做了大量优化，有效的提升了数据检索效率，目前已落地微信读书“AI 问书”、微信输入法“问 AI”、腾讯地图、腾讯会议、IMA Copilot、乐享智能搜索等大型应用中。

同时，作为国内公有云首个从自然语言处理、到向量生成/存储/检索、并与大模型集成的端到端一站式技术平台，腾讯云ES作为核心参编单位参与了由中国信通院发起的的RAG标准制定，并成为首个通过RAG权威认证的企业。

购买ES 集群

1、登录 登录-腾讯云

2、点击「新建」：

3、计费模式为按量计费，产品版本为标准版、ES 版本为 8.13.3、商业特性为白金版：

ES 节点配置，测试环境可选择为ES.S1(2核4G)，节点数为2，磁盘为通用型SSD，磁盘容量为 20GB：

登录Kibana

访问 Kibana，设置公网访问策略：

2、获取当前 IP 地址并设置到 IP 白名单中：

3、点击Kibana公网访问地址访问Kibana。

部署embedding 模型

在集群购买完成后，就可以前往 Kibana 部署 Embedding 模型、创建知识库索引与向量化管道：

1、开启「节点出站访问」，仅开启数据节点即可，如有专用主节点，仅开启专用主节点即可。（该功能为白名单，请联系工单处理）：

注：如需上传自定义模型或第三方平台（如 Huggingface）模型，可参考

https://github.com/elastic/eland

2、登录 Kibana 之后，在左侧导航栏找到「Machine Learning」功能：

3、进入模型管理页面，并找到类型为「text_embedding」的模型：

如为未下载状态，选中模型，并点击「Add trained model」，本次演示我们使用「.multilingual-e5-small_linux-x86_64」模型，这是一个 384 维的多语言模型：

5、下载完成后，点击部署：

6、为快速体验，配置使用默认值即可：

创建索引与向量化管道

1、点击进入「Dev tools」

2、创建知识库索引：index-name为索引名称，实际可按需命名。

PUT index-name
{
  "mappings": {
    "properties": {
    "title": {
        "type": "keyword"
      },
      "content": {
        "type": "text"
      },
      "url": {
        "type": "keyword"
      },
      "content_embedding": {
        "type": "dense_vector",
        "dims": 384
      }
    }
  }
}

3、创建推理管道，该管道可用于写入数据时进行数据向量化。

PUT _ingest/pipeline/index-name-pipeline
{
  "description": "Text embedding pipeline",
  "processors": [
    {
      "inference": {
        "model_id": ".multilingual-e5-small_linux-x86_64",
        "input_output": [
          {
            "input_field": "content",
            "output_field": "content_embedding"
          },
          {
            "input_field": "title",
            "output_field": "title_embedding"


          }
        ]
      }
    }
  ]
}

上述管道，将字段「content、title」的内容，调用.multilingual-e5-small_linux-x86_64模型向量化之后存储到新的字段中。

写入知识库数据

通过 Bulk API 批量写入数据，可将 title 、content 、url 的内容替换为实际的知识库数据。

POST index-name/_bulk?pipeline=index-name-pipeline&refresh
{ "index" : {} }
{ "title" : "标题 1","content": "内容 1","url": "https：url1"  }
{ "index" : {} }
{ "title" : "标题 2","content": "内容 2","url": "https：url2"  }

调用DeepSeek大模型

1、python文件命名为deepseek.py。

2、获取DeepSeek（此处超链接），当前可免费使用 500 万 tokens：

3、 deepseek.py内容如下:

import json
import requests
from typing import Dict, Any


class DeepSeekAPI:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = " https://api.deepseek.com/v1" 
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }


    def chat_completion(self, system_prompt: str, user_message: str) -> Dict[str, Any]:
        url = f"{self.base_url}/chat/completions"
        payload = {
            "model": "deepseek-chat",
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_message}
            ],
            "temperature": 0.7,
            "max_tokens": 2000
        }


        try:
            response = requests.post(
                url,
                headers=self.headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": str(e)}


def deepseek_chat(system_prompt: str, content: str):
    api_key = "输入申请的 api_key"
    deepseek = DeepSeekAPI(api_key)
    return deepseek.chat_completion(system_prompt, content)

智能问答系统构建

1、安装 streamlit：

pip install streamlit

2、获取 ES 访问地址：用户名为 elastic、密码在创建集群时设置，用本地mac测试时，可开启公网访问，实际生产时，建议使用内网访问地址。

3、运行如下代码（ 可命名为 web.py，需与 deepseek.py在一个目录下）

import streamlit as st
from elasticsearch import Elasticsearch
from deepseek import deepseek_chat


es_client = Elasticsearch(
    "ES集群访问地址",basic_auth=("用户名", "密码"))


def get_elasticsearch_results(query):
    es_query = {
        "knn": {
            "field": "content_embedding",
            "num_candidates": 100,
            "query_vector_builder": {
                "text_embedding": {
                    "model_id": ".multilingual-e5-small_linux-x86_64",
                    "model_text": query
                }
            }
        },
        "query":{
            "match":{
                "content":query
            }
        },
        "rank":{
            "rrf":{
                "window_size":100,
                "rank_constant":20
            }
        }
    }
    result = es_client.search(index="index-name", body=es_query)
    return result["hits"]["hits"]


def create_deepseek_prompt(results,question):
    context = ""
    for hit in results:
        source_field = hit["_source"]["content"]
        hit_context = source_field
        context += f"{hit_context}\n"
    prompt = f"""
Instructions:


回答此问题：{question}
回答时，你只能参考文档{context} 生成答案，如果无法生成答案，请回复：对不起，该问题我无法回答


"""
    return prompt


def generate_deepseek_completion(user_prompt, question):
    response = deepseek_chat(user_prompt, question)
    if 'error' in response:
        return f"Error: {response['error']}"
    return response['choices'][0]['message']['content']


def format_result(hit):
    title = hit["_source"]["title"]
    return f'[{title}]'


def main():
    st.title("我的专属AI助手")


    # 创建输入框和查询按钮
    question = st.text_input("请输入您的问题：")
    if st.button("查询并生成答案"):
        elasticsearch_results = get_elasticsearch_results(question)
        user_prompt = create_deepseek_prompt(elasticsearch_results,question)
        system_prompt = f"""
你是一个问答任务的助手。使用呈现的上下文真实、实事求是地回答问题。
"""


        openai_completion = generate_deepseek_completion(system_prompt, user_prompt)


        # 显示结果
        st.write(openai_completion)


        # 展示Elasticsearch查询结果
        st.write("大模型参考的文档：")
        for hit in elasticsearch_results:
            st.markdown(format_result(hit))


if __name__ == "__main__":
    main()

在上述 python 文件的目录下，使用如下命令运行系统：

streamlit run web.py

生成的界面如下：

私域数据问答测试

1、索引中无数据：

2、AI助手无法回答：

3、写入相关数据:

POST index-name/_bulk?pipeline=index-name-pipeline&refresh
{ "index" : {} }
{ "title" : "腾讯云ES的产品经理","content": "腾讯云ES的产品经理是Kevin","url": "https：url1"  }

4、回答如下:

本文通过介绍腾讯云ES一站式 RAG 方案，演示如何通过结合腾讯云 ES 与DeepSeek大模型，快速构建RAG 应用。腾讯云ES凭借其在传统PB级日志和海量搜索场景中积累的丰富经验，通过深度重构底层系统，成功地将多年的性能优化、索引构建和运营管理经验应用于RAG领域，并积极探索向量召回与传统搜索技术的融合之道，旨在充分发挥两者的优势，为用户提供更加精准、高效的搜索体验。未来，腾讯云ES将持续深耕智能检索领域，在成本、性能、稳定性等方面持续提升，帮助客户降本增效的同时实现业务价值持续增长，欢迎持续关注！

腾讯云大数据始终致力于为各行业客户提供轻快、易用，智能的大数据平台。

END

⏬点击阅读原文

了解更多产品详情

我知道你在看哟