如何通过 LlamaIndex 将数据导入 Elasticsearch

原创

点火三周

发布于 2025-03-05 18:08:42

4800

代码可运行

运行总次数：0

代码可运行

本文将介绍如何使用 LlamaIndex 将数据索引到 Elasticsearch 中，以实现 FAQ 搜索引擎。Elasticsearch 将作为我们的向量数据库，支持向量搜索，而 RAG（检索增强生成）将丰富搜索上下文，提供更准确的响应。

什么是 LlamaIndex？

LlamaIndex 是一个框架，旨在通过大型语言模型（LLMs）与特定或私有数据进行交互，简化智能代理和工作流程的创建。它允许将来自各种来源（API、PDF、数据库）的数据与 LLM 结合，实现如研究、信息提取和上下文化响应生成等任务。

关键概念:

代理（Agents）：使用 LLM 执行任务的智能助手，从简单的响应到复杂的操作。
工作流程（Workflows）：结合代理、数据连接器和工具以执行高级任务的多步流程。
上下文增强（Context augmentation）：利用外部数据丰富 LLM，克服其训练局限性。

LlamaIndex 与 Elasticsearch 的集成：

Elasticsearch 可以在多种方式中与 LlamaIndex 集成：

数据源：使用 Elasticsearch Reader 提取文档。
嵌入模型：将数据编码为向量用于语义搜索。
向量存储：将 Elasticsearch 用作向量化文档的搜索存储库。
高级存储：配置如文档摘要或知识图谱等结构。

使用 LlamaIndex 和 Elasticsearch 构建 FAQ 搜索引擎

数据准备

我们将使用 Elasticsearch 服务 FAQ 作为示例。每个问题都从网站中提取并保存在单独的文本文件中。你可以使用任何方法来组织数据；在这个例子中，我们选择将文件保存在本地。

示例文件：

文件名: what-is-elasticsearch-service.txt
内容: Elasticsearch Service 是由 Elasticsearch 的创建者提供托管和管理的 Elasticsearch 和 Kibana 服务。Elasticsearch Service 是 Elastic Cloud 的一部分，配备了只有 Elasticsearch、Kibana、Beats 和 Logstash 的公司才能提供的功能。Elasticsearch 是一个全文搜索引擎，适用于从网站搜索到大数据分析等各种用途。

保存所有问题后，目录将如下所示：

安装依赖

我们将使用 Python 语言实现数据摄取和搜索，我使用的版本是 3.9。作为前提，需要安装以下依赖：

llama-index
vector-stores-elasticsearch
openai

Elasticsearch 和 Kibana 将通过 Docker 创建，并通过 docker-compose.yml 配置为运行版本 8.16.2。这使得创建本地环境更加容易。

version: '3.8'
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.16.2
    container_name: elasticsearch-8.16.2
    environment:
      - node.name=elasticsearch
      - xpack.security.enabled=false
      - discovery.type=single-node
      - "ES_JAVA_OPTS=-Xms1024m -Xmx1024m"
    ports:
      - 9200:9200
    networks:
      - shared_network
  kibana:
    image: docker.elastic.co/kibana/kibana:8.16.2
    container_name: kibana-8.16.2
    restart: always
    environment:
      - ELASTICSEARCH_URL=http://elasticsearch:9200
    ports:
      - 5601:5601
    depends_on:
      - elasticsearch
    networks:
      - shared_network
networks:
  shared_network:

文档摄取

将使用 LlamaIndex 将文档索引到 Elasticsearch 中。首先，我们使用 SimpleDirectoryReader 加载本地目录中的文件。加载文档后，我们将使用 VectorStoreIndex 对其进行索引。

documents = SimpleDirectoryReader("./faq").load_data()
storage_context = StorageContext.from_defaults(vector_store=es)
index = VectorStoreIndex(documents, storage_context=storage_context, embed_model=embed_model)

LlamaIndex 中的向量存储负责存储和管理文档嵌入。LlamaIndex 支持多种类型的向量存储，这里我们使用 Elasticsearch。在 StorageContext 中，我们配置 Elasticsearch 实例。由于上下文是本地的，不需要额外参数。对于其他环境中的配置，请参考文档检查所需参数：ElasticsearchStore 配置。

默认情况下，LlamaIndex 使用 OpenAI 的 text-embedding-ada-002 模型生成嵌入。然而，在这个例子中，我们将使用 text-embedding-3-small 模型。使用该模型需要一个 OpenAI API 密钥。

以下是完整的文档摄取代码。

import openai
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.elasticsearch import ElasticsearchStore

openai.api_key = os.environ["OPENAI_API_KEY"]

es = ElasticsearchStore(
    index_name="faq",
    es_url="http://localhost:9200"
)

def format_title(filename):
    filename_without_ext = filename.replace('.txt', '')
    text_with_spaces = filename_without_ext.replace('-', ' ')
    formatted_text = text_with_spaces.title()
    return formatted_text

embed_model = OpenAIEmbedding(model="text-embedding-3-small")
documents = SimpleDirectoryReader("./faq").load_data()

for doc in documents:
    doc.metadata['title'] = format_title(doc.metadata['file_name'])

storage_context = StorageContext.from_defaults(vector_store=es)
index = VectorStoreIndex(documents, storage_context=storage_context, embed_model=embed_model)

执行后，文档将被索引到 faq 索引中，如下所示：

使用 RAG 进行搜索

为了进行搜索，我们配置 ElasticsearchStore 客户端，设置 index_name 和 es_url 字段。我们定义 AsyncDenseVectorStrategy 作为向量搜索策略。其他策略，如 AsyncBM25Strategy（关键字搜索）和 AsyncSparseVectorStrategy（稀疏向量），也可用。更多详情请参考官方文档。

es = ElasticsearchStore(
    index_name="faq",
    es_url="http://localhost:9200",
    retrieval_strategy=AsyncDenseVectorStrategy()
)

接下来，我们将创建一个 VectorStoreIndex 对象，在其中配置 vector_store 使用 ElasticsearchStore 对象。通过 as_retriever 方法，我们可以检索与查询最相关的文档，设置返回结果的数量为 5。

index = VectorStoreIndex.from_vector_store(vector_store=es)
retriever = index.as_retriever(similarity_top_k=5)
results = retriever.retrieve(query)

下一步是 RAG。向量搜索的结果被整合到一个格式化的提示中，供 LLM 使用，从而基于检索到的信息生成上下文化的响应。

在 PromptTemplate 中，我们定义了提示的格式，其中包括：

上下文（{context_str}）：由检索器检索到的文档。
查询（{query_str}）：用户的问题。
说明：指导模型基于上下文进行回答，而不依赖外部知识。

qa_prompt = PromptTemplate(
    "你是一个有帮助且知识渊博的助手。"
    "你的任务是根据下面提供的上下文回答用户的问题。"
    "不要使用任何先前知识或外部信息。\n"
    "---------------------\n"
    "上下文：\n"
    "{context_str}\n"
    "---------------------\n"
    "查询：{query_str}\n"
    "说明：\n"
    "1. 仔细阅读并理解提供的上下文。\n"
    "2. 如果上下文包含足够的信息来回答查询，请提供一个清晰简洁的答案。\n"
    "3. 不要编造或猜测任何信息。\n"
    "答案："
)

最终，LLM 处理提示并返回一个精确且上下文化的响应。

llm = OpenAI(model="gpt-4o")
context_str = "\n\n".join([n.node.get_content() for n in results])
response = llm.complete(
    qa_prompt.format(context_str=context_str, query_str=query)
)
print("答案:")
print(response)

完整代码如下：

es = ElasticsearchStore(
    index_name="faq",
    es_url="http://localhost:9200",
    retrieval_strategy=AsyncDenseVectorStrategy()
)

def print_results(results):
    for rank, result in enumerate(results, start=1):
        title = result.metadata.get("title")
        score = result.get_score()
        text = result.get_text()
        print(f"{rank}. title={title} \nscore={score} \ncontent={text}")

def search(query: str):
    index = VectorStoreIndex.from_vector_store(vector_store=es)
    retriever = index.as_retriever(similarity_top_k=10)
    results = retriever.retrieve(QueryBundle(query_str=query))
    print_results(results)
    
    qa_prompt = PromptTemplate(
        "你是一个有帮助且知识渊博的助手。"
        "你的任务是根据下面提供的上下文回答用户的问题。"
        "不要使用任何先前知识或外部信息。\n"
        "---------------------\n"
        "上下文：\n"
        "{context_str}\n"
        "---------------------\n"
        "查询：{query_str}\n"
        "说明：\n"
        "1. 仔细阅读并理解提供的上下文。\n"
        "2. 如果上下文包含足够的信息来回答查询，请提供一个清晰简洁的答案。\n"
        "3. 不要编造或猜测任何信息。\n"
        "答案："
    )
    
    llm = OpenAI(model="gpt-4o")
    context_str = "\n\n".join([n.node.get_content() for n in results])
    response = llm.complete(
        qa_prompt.format(context_str=context_str, query_str=query)
    )
    print("答案:")
    print(response)

question = "Elastic 服务是免费的吗？"
print(f"问题: {question}")
search(question)

现在我们可以执行搜索，例如 "Elastic 服务是免费的吗？" 并根据 FAQ 数据本身得到上下文化的响应。

问题: Elastic 服务是免费的吗？
答案:
Elastic 服务并非完全免费。然而，有一个 14 天的免费试用期可用于探索 Elastic 解决方案。在试用期结束后，对功能和服务的访问取决于订阅级别。

生成此响应所使用的文档如下：

1. 标题=我可以免费试用 Elasticsearch 服务吗 
分数=1.0 
内容=是的，注册后可享受 14 天的免费试用。试用期从创建集群的那一刻开始。在免费试用期内，可以访问一个部署以探索企业搜索、可观察性、安全性或最新版本的 Elastic Stack 解决方案。

2. 标题=你们提供 Elastic 的商业产品吗 
分数=0.9941274512218439 
内容=是的，所有 Elasticsearch 服务客户都可以使用基本身份验证、基于角色的访问控制和监控。Elasticsearch 服务的金、白金和企业客户可以完全访问 X-Pack 中的所有功能：安全性、警报、监控、报告、图形分析和可视化。联系我们了解更多信息。

3. 标题=什么是 Elasticsearch 服务 
分数=0.9896776845746571 
内容=Elasticsearch 服务是由 Elasticsearch 的创建者提供托管和管理的 Elasticsearch 和 Kibana 服务。Elasticsearch 服务是 Elastic Cloud 的一部分，配备了只有 Elasticsearch、Kibana、Beats 和 Logstash 的公司才能提供的功能。Elasticsearch 是一个全文搜索引擎，适用于从网站搜索到大数据分析等各种用途。

4. 标题=我可以在 Elasticsearch 服务中运行完整的 Elastic Stack 吗 
分数=0.9880631561979476 
内容=许多属于 Elastic Stack 的产品在 Elasticsearch 服务中均可用，包括 Elasticsearch、Kibana、插件和如监控和安全性等功能。可以直接使用其他 Elastic Stack 产品与 Elasticsearch 服务。例如，Logstash 和 Beats 都可以将其数据发送到 Elasticsearch 服务。运行的内容取决于订阅级别。

5. 标题=Elasticsearch 服务与 Amazon Elasticsearch 服务有什么区别 
分数=0.9835054890793161 
内容=Elasticsearch 服务是唯一由 Elasticsearch、Kibana、Beats 和 Logstash 的创建公司提供、管理和支持的托管 Elasticsearch 服务。使用 Elasticsearch 服务，您始终可以获得最新版本的软件。我们的服务基于最佳实践和多年托管和管理数千个 Elasticsearch 集群的经验。有关更多信息，请查看以下 Amazon 和 Elastic Elasticsearch 服务比较页面。请注意，Elastic 和 Amazon Web Services (AWS) 之间没有正式合作伙伴关系，Elastic 不对 AWS Elasticsearch 服务提供任何支持。