本文聚焦一个实际命题：如何借助 Elasticsearch 将 AI 搜索系统与 RAG 应用真正落地部署。这是一份完整的实战指南。

首先，我们来梳理在 Elasticsearch 中实现向量搜索所需的几大核心组件。

首先是嵌入模型。简而言之，它是一种将文本、图像等非结构化数据转换成数字向量的机器学习模型。向量是相似性搜索的基石——没有这一映射，机器便无法理解语义，更无法计算相似度。

其次是推理终端节点。它承担两个核心任务：一是在数据写入阶段运行模型生成向量；二是在查询阶段对用户输入执行相同模型，生成查询向量。推理端点本质上扮演着数据流与模型之间的“翻译官”角色。特别提醒，若处理图片等非文本数据，通常需要借助外部脚本来完成向量生成。

接下来是搜索层。Elasticsearch 将生成的向量与原始元数据一同存入索引，并利用近似 k 最近邻搜索（k-NN），在向量空间中快速定位最匹配的结果。这一机制构成了大规模相似性搜索的底层引擎。

最后是应用程序逻辑。它涵盖用户交互、业务逻辑处理、搜索结果展示等环节。所有核心搜索之外的复杂工作都由这一层承担。

将这些组件整合，便构成了一条完整的端到端 AI 搜索链路。

现在进入核心内容：从零构建一个 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 构建完整的 RAG 系统。

ES 在 RAG 领域的解决方案如上图所示。一个典型场景：用户将问题直接提交给大语言模型（LLM），而模型只能依赖其内部知识作答。一旦涉及企业内部私域信息，模型便无法正确回复。此时，Elasticsearch 启动 RAG 方案，将问题转交给知识库进行检索。

知识库中包含文本、图片、视频等多种数据。我们提前将其向量化，在检索阶段执行文本与向量的混合召回，获取 TopN 候选结果。随后将这些结果与用户原始问题组合成 prompt，提交给大模型。大模型基于上下文信息，输出更精准的回答。

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

首先安装所需依赖包：

pip install langchain-elasticsearch

配置 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 提供了完全的自定义能力。

这里推荐一个更灵活的做法：在 LangChain 中使用 ElasticsearchRetriever，通过自定义函数将用户查询映射至 ES 请求。

举一个实际场景：假设希望在检索环节加入语义重排序，利用 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 智能问答应用。