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

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

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"

          }
        ]
      }
    }
  ]
}

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"  }

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)


pip install streamlit

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()

streamlit run web.py

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