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

点击蓝字⬆ 关注我们
  

  本文共计1669字 预计阅读时长6分钟
  随着数据智能技能的不断发展，以大语言模子（LLM）驱动的AIGC为代表的内容生成技能已经成为企业数据智能本事中不可或缺的一部分，但传统的内容生成技能存在信息更新不实时、垂直范畴知识匮乏、模子幻觉等问题，怎样推进大模子在各行业、各业务场景落地是各方普遍关注的问题，而检索增强生成（Retrieval-Augmented Generation，RAG）技能则为此提供了有效的办理方案，成为数据智能时代的一大趋势。
  RAG是一种联合了检索和大语言模子内容生成的技能方案，它通过引用外部知识库，在用户输入Query时检索出知识，然后让模子基于可信的知识进行用户回答。RAG具有较高的可表明性和定制本事，可大幅低落大语言模子的幻觉，适用于问答系统、文档生成、智能助手等多种自然语言处理任务。本文将通过先容腾讯云ES 一站式 RAG 方案，演示怎样通过联合腾讯云 ES 与 DeepSeek 大模子，快速构建 RAG 应用。
  腾讯云ES 一站式 RAG 方案
  腾讯云ES是云端全托管海量数据检索分析服务，拥有高性能自研内核，集成X-Pack，支持通过自治索引、存算分离、集群巡检等特性轻松管理集群，也支持免运维、主动弹性、按需使用的 Serverless 模式。在自研内核方面，腾讯云ES依托腾讯表里部海量业务的运营经验，针对ES内核进行了成本、性能、稳定性、拓展性等方面的优化，是举世第三方企业开源贡献第一的团队，使用腾讯云ES可以高效构建在线搜索、向量检索、日志分析、运维监控、智能问答等服务。
  在RAG方面，腾讯云ES支持了一站式向量检索、文本+向量混合搜索、倒数排序融合、与大模子集成、GPU高性能推理、字段级别权限控制等本事，同时针对查询性能做了大量优化，有效的提升了数据检索效率，目前已落地微信读书“AI 问书”、微信输入法“问 AI”、腾讯地图、腾讯集会、IMA Copilot、乐享智能搜索等大型应用中。
  

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

  AI 助手构建
  购买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为索引名称，实际可按需命名。

1. PUT /index-name
2. {
3.  "mappings": {
4.    "properties": {
5.    "title": {
6.        "type": "keyword"
7.      },
8.      "content": {
9.        "type": "text"
10.      },
11.      "url": {
12.        "type": "keyword"
13.      },
14.      "content_embedding": {
15.        "type": "dense_vector",
16.        "dims": 384
17.      }
18.    }
19.  }
20. }

复制代码

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

1. PUT /_ingest/pipeline/index-name-pipeline
2. {
3.  "description": "Text embedding pipeline",
4.  "processors": [
5.    {
6.      "inference": {
7.        "model_id": ".multilingual-e5-small_linux-x86_64",
8.        "input_output": [
9.          {
10.            "input_field": "content",
11.            "output_field": "content_embedding"
12.          },
13.          {
14.            "input_field": "title",
15.            "output_field": "title_embedding"
18.          }
19.        ]
20.      }
21.    }
22.  ]
23. }

复制代码

上述管道，将字段「content、title」的内容，调用.multilingual-e5-small_linux-x86_64模子向量化之后存储到新的字段中。
  写入知识库数据
  通过 Bulk API 批量写入数据，可将 title 、content 、url 的内容更换为实际的知识库数据。

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

复制代码

调用DeepSeek大模子
  1、python文件命名为deepseek.py。
  2、获取DeepSeek（此处超链接），当前可免费使用 500 万 tokens：
  

  3、 deepseek.py内容如下:

1. import json
2. import requests
3. from typing import Dict, Any
6. class DeepSeekAPI:
7.    def __init__(self, api_key: str):
8.        self.api_key = api_key
9.        self.base_url = " https://api.deepseek.com/v1"
10.        self.headers = {
11.            "Authorization": f"Bearer {self.api_key}",
12.            "Content-Type": "application/json"
13.        }
16.    def chat_completion(self, system_prompt: str, user_message: str) -> Dict[str, Any]:
17.        url = f"{self.base_url}/chat/completions"
18.        payload = {
19.            "model": "deepseek-chat",
20.            "messages": [
21.                {"role": "system", "content": system_prompt},
22.                {"role": "user", "content": user_message}
23.            ],
24.            "temperature": 0.7,
25.            "max_tokens": 2000
26.        }
29.        try:
30.            response = requests.post(
31.                url,
32.                headers=self.headers,
33.                json=payload,
34.                timeout=30
35.            )
36.            response.raise_for_status()
37.            return response.json()
38.        except requests.exceptions.RequestException as e:
39.            return {"error": str(e)}
42. def deepseek_chat(system_prompt: str, content: str):
43.    api_key = "输入申请的 api_key"
44.    deepseek = DeepSeekAPI(api_key)
45.    return deepseek.chat_completion(system_prompt, content)

复制代码

智能问答系统构建
  1、安装 streamlit：

1. pip install streamlit

复制代码

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

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

1. import streamlit as st
2. from elasticsearch import Elasticsearch
3. from deepseek import deepseek_chat
6. es_client = Elasticsearch(
7.    "ES集群访问地址",basic_auth=("用户名", "密码"))
10. def get_elasticsearch_results(query):
11.    es_query = {
12.        "knn": {
13.            "field": "content_embedding",
14.            "num_candidates": 100,
15.            "query_vector_builder": {
16.                "text_embedding": {
17.                    "model_id": ".multilingual-e5-small_linux-x86_64",
18.                    "model_text": query
19.                }
20.            }
21.        },
22.        "query":{
23.            "match":{
24.                "content":query
25.            }
26.        },
27.        "rank":{
28.            "rrf":{
29.                "window_size":100,
30.                "rank_constant":20
31.            }
32.        }
33.    }
34.    result = es_client.search(index="index-name", body=es_query)
35.    return result["hits"]["hits"]
38. def create_deepseek_prompt(results,question):
39.    context = ""
40.    for hit in results:
41.        source_field = hit["_source"]["content"]
42.        hit_context = source_field
43.        context += f"{hit_context}\n"
44.    prompt = f"""
45. Instructions:
48. 回答此问题：{question}
49. 回答时，你只能参考文档{context} 生成答案，如果无法生成答案，请回复：对不起，该问题我无法回答
52. """
53.    return prompt
56. def generate_deepseek_completion(user_prompt, question):
57.    response = deepseek_chat(user_prompt, question)
58.    if 'error' in response:
59.        return f"Error: {response['error']}"
60.    return response['choices'][0]['message']['content']
63. def format_result(hit):
64.    title = hit["_source"]["title"]
65.    return f'[{title}]'
68. def main():
69.    st.title("我的专属AI助手")
72.    # 创建输入框和查询按钮
73.    question = st.text_input("请输入您的问题：")
74.    if st.button("查询并生成答案"):
75.        elasticsearch_results = get_elasticsearch_results(question)
76.        user_prompt = create_deepseek_prompt(elasticsearch_results,question)
77.        system_prompt = f"""
78. 你是一个问答任务的助手。使用呈现的上下文真实、实事求是地回答问题。
79. """
82.        openai_completion = generate_deepseek_completion(system_prompt, user_prompt)
85.        # 显示结果
86.        st.write(openai_completion)
89.        # 展示Elasticsearch查询结果
90.        st.write("大模型参考的文档：")
91.        for hit in elasticsearch_results:
92.            st.markdown(format_result(hit))
95. if __name__ == "__main__":
96.    main()

复制代码

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

1. streamlit run web.py

复制代码

生成的界面如下：
  

  私域数据问答测试
  1、索引中无数据：
  

  2、AI助手无法回答：
  

  3、写入相关数据:

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

复制代码

4、回答如下:
  

  总结
  本文通过先容腾讯云ES一站式 RAG 方案，演示怎样通过联合腾讯云 ES 与DeepSeek大模子，快速构建RAG 应用。腾讯云ES依附其在传统PB级日志和海量搜索场景中积聚的丰富经验，通过深度重构底层系统，成功地将多年的性能优化、索引构建和运营管理经验应用于RAG范畴，并积极探索向量召回与传统搜索技能的融合之道，旨在充分发挥两者的优势，为用户提供更加精准、高效的搜索体验。未来，腾讯云ES将持续深耕智能检索范畴，在成本、性能、稳定性等方面持续提升，帮助客户降本增效的同时实现业务价值持续增长，欢迎持续关注！
  腾讯云大数据始终致力于为各行业客户提供轻快、易用，智能的大数据平台。
  END
  关注腾讯云大数据╳探索数据的无穷可能
  

  ⏬点击阅读原文
  相识更多产品详情
  

  我知道你在看哟
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。