RAG最佳实践：用 ElasticSearch 打造AI搜索系统与RAG 应用全流程详解！

今天来聊一个非常实际的话题：如何用 Elasticsearch 把 AI 搜索系统和 RAG 应用真正落地。

先拆解一下，在 Elasticsearch 里实现向量搜索，究竟需要哪几个核心组件。

首先是

。说白了，就是一个能把文本、图像这类非结构化数据“翻译”成数字向量的机器学习模型。向量是整个相似性搜索的基础——没有这个映射，机器就没办法理解语义，更别提什么“相似度”了。

然后是

。它主要负责两件事：一是在数据写入时跑模型、生成向量；二是在查询时，对用户输入也跑一遍同样的模型，生成查询向量。说白了，推理端点是数据流和模型之间的“翻译官”。特别注意，如果处理的是图片这类非文本数据，还得借助外部脚本来完成向量生成。

接下来是

层。Elasticsearch 会把生成的向量和原始元数据一起存入索引，然后通过近似的 k 最近邻搜索（k-NN），在向量空间里快速找到最匹配的结果。这套机制，是支撑大规模相似性搜索的底层引擎。

最后是

。这部分包括与用户的交互、业务逻辑的处理、搜索结果的呈现等等。核心搜索之外所有的“苦活累活”，都落在这里。

把这些组件整合到一起，就构成了一个完整的 AI 搜索链路。

接下来，我们进入正题：如何从零手写一个 AI 对话式搜索应用。先简单定义一下，

并不是简单的关键词匹配，而是一种基于自然语言处理和机器学习的交互方式。用户可以用自然语言提问，系统理解意图后，通过语义匹配给出答案。这种体验更贴近人类交流，在复杂查询和个性化推荐场景中，优势非常明显。

架构图如下所示（图中展示了从用户输入到检索再到生成回答的完整流程）。

确定数据来源——常见的有内部知识库、FAQ、文档资料。然后需要构建一个数据管道，把这些数据导入检索系统，准备好在 RAG 应用中直接使用。

这里需要准备索引映射、建立索引、并存储数据。代码示例：

from elasticsearch_serverless import Elasticsearch
import json
import os

client = Elasticsearch(
    os.getenv("ELASTICSEARCH_URL"),
    api_key=os.getenv("ES_API_KEY"),
    request_timeout=600
)

mappings={
    "properties":{
        "semantic":{"type":"semantic_text","inference_id":"e5-small"},
        "content":{"type":"text","copy_to":"semantic"}
    }
}

# Create index
client.indices.create(index="search-faq", mappings=mappings)

要创建一个推理服务来操作 E5 多语言 ML 模型：

inference_config={
    "service":"elasticsearch",
    "service_settings":{
        "num_allocations":1,
        "num_threads":1,
        "model_id":".multilingual-e5-small"
    }
}

# Create inference
client.inference.put(
    inference_id="e5-small",
    task_type="text_embedding",
    inference_config=inference_config
)

with open("faq.json") as f:
    documents = json.load(f)

def generate_docs():
    index_name = "search-faq"
    for row in documents:
        yield { "index" : { "_index" : index_name } }
        yield row

client.bulk(operations=generate_docs())

负责与用户交互并展示搜索结果：

// Retrieve relevant content from the knowledge base
async function findRelevantContent(question: string) {
   // Semantic search query
   const body = await client.search({
      size: 3,
      index: 'search-faq',
      body: {
         query: {
            semantic: {
               field: "semantic",
               query: question
            }
         }
      }
   });
   return body.hits.hits.map((hit: any) => ({
      content: hit._source.content
   }));
}

页面效果图如下所示（展示了最终的搜索界面和结果呈现）。

——

接下来，我们看如何用 Elasticsearch 搭建

。

ES 在 RAG 领域的解决方案如上图所示。一个典型的场景是：用户把问题直接抛给大模型，大模型只能依赖自己的内部知识回答。一旦涉及企业内部的私域信息，大模型就“抓瞎”了。这时 ES 就会启动 RAG 方案，把问题转交给知识库去检索。

知识库中不仅有文本，还有图片、视频等。我们提前把这些内容向量化，然后在检索阶段做文本和向量的混合召回，拿到 TopN 列表。之后将这份列表与用户原始问题拼成一个 prompt，再交给大模型。大模型根据这些上下文信息，就能给出更加精准的回答。

RAG 的核心思想就是“先检索、后生成”：先从外部知识库检索相关片段，再把这些片段和用户提问一起送入生成模型。用 Elasticsearch 作为向量数据库，解决大规模数据下的高效检索；而 LangChain 则负责构建和管理这种复杂应用的各层逻辑。

安装依赖：

pip install langchain-elasticsearch

配置连接：

from langchain_elasticsearch import ElasticsearchStore

es_store = ElasticsearchStore(
    es_cloud_id="your-cloud-id",
    es_api_key="your-api-key",
    index_name="rag-example",
    strategy=ElasticsearchStore.SparseVectorRetrievalStrategy(model_id=".elser_model_2"),
)

：将相关文档、技术资料存储到 Elasticsearch 中。

texts = [
   "LangChain is a framework for developing applications powered by large language models (LLMs).",
   "Elasticsearch is a distributed, RESTful search and analytics engine capable of addressing a growing number of use cases.",
   ...
]
es_store.add_texts(texts)

：用嵌入模型把文档转为向量，存入 ES 的向量字段。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(api_key="sk-...")

：

from langchain import hub
from langchain_core.runnables import RunnablePassthrough

prompt = hub.pull("rlm/rag-prompt")
retriever = es_store.as_retriever()

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

：用户输入查询，先通过同样的嵌入模型转为向量。

：LangChain 将检索到的上下文和用户原始问题拼成 prompt，交给 LLM 生成最终答案。

rag_chain.invoke("Which frameworks can help me build LLM apps?")

Elasticsearch Store 默认提供了多种开箱即用的检索策略，开发人员可以根据实际场景自由选择。但如果数据模型更复杂——比如索引里除了文本，还有标题、URL、标签等字段——该怎么办？Elasticsearch 的 Query DSL 给了你完全的控制权。

举个实际例子：假设我们想在检索环节加入语义重排序，用 Cohere 重排序模型提升顶部结果的相关性。可以用下面这种方式实现：

def text_similarity_reranking(search_query: str) -> Dict:
    return {
        "retriever": {
            "text_similarity_reranker": {
                "retriever": {
                    "standard": {
                        "query": {
                            "match": {
                                "text_field": search_query
                            }
                        }
                    }
                },
                "field": "text_field",
                "inference_id": "cohere-rerank-service",
                "inference_text": search_query,
                "window_size": 10
            }
        }
    }

retriever = ElasticsearchRetriever.from_es_params(
    es_cloud_id="your-cloud-id",
    es_api_key="your-api-key",
    index_name="rag-example",
    content_field=text_field,
    body_func=text_similarity_reranking,
)

总的来说，Elasticsearch 在 RAG 方向上的优势非常明显：

：独立技术栈，一站式完成向量生成、存储、索引和检索，通过配置就能实现大部分功能，大大降低了接入门槛。

：支持百万级 QPS 和千亿级数据量，分布式架构赋予了它极好的弹性。

：文本与向量的混合检索，显著提升搜索结果的准确性。

：无缝连接大模型，轻松构建 AI 智能问答应用。