使用 LlamaIndex Workflows 与 Elasticsearch

作者：来自 Elastic Jeffrey Rengifo

在本文中，你将学习如何利用 LlamaIndex Workflows 与 Elasticsearch 快速构建一个使用 LLM 的自过滤搜索应用程序。

LlamaIndex Workflows 提出了一种不同的方式来处理将任务拆分给不同 agent 的问题，它引入了一个步骤与事件的架构。这种架构相比于基于 DAG（Directed Acyclic Graph - 有向无环图）的方法，例如 LangGraph，简化了设计。如果你想进一步了解关于 agent 的内容，推荐你阅读这篇文章。

LlamaIndex 的一项主要功能是能够在执行过程中轻松创建循环。循环可以帮助我们完成自动纠错任务，因为我们可以重复某个步骤，直到获得预期结果或达到设定的重试次数。

为了测试这一功能，我们将构建一个流程，使用 LLM 根据用户的问题生成 Elasticsearch 查询，并在生成的查询无效时启用自动纠错机制。如果在设定次数内 LLM 无法生成有效查询，我们将切换模型并继续尝试，直到超时为止。

为了优化资源使用，我们可以先使用更快且更便宜的模型生成初始查询，如果生成仍然失败，则改用更昂贵的模型。

理解步骤与事件

步骤是通过代码函数运行的操作。它接收一个事件以及一个上下文，该上下文可以被所有步骤共享。基础事件有两种类型：

• StartEvent，是启动流程的事件

• StopEvent，用于停止事件的执行

一个 Workflow 是一个包含所有步骤与交互的类，用于将它们组合在一起。

我们将创建一个 Workflow 来接收用户请求、展示映射和可筛选字段、生成查询语句，然后通过循环修复无效的查询。对于 Elasticsearch 来说，查询可能无效的原因包括：生成的 JSON 不合法，或语法错误。

为了展示其工作原理，我们将使用一个实际案例：搜索酒店房间。通过 workflow 提取值，根据用户的搜索来构建查询。

完整示例可在这个 Notebook 中查看。

步骤

• 安装依赖并导入包

• 准备数据

• LlamaIndex workflows

• 执行 workflow 任务

安装依赖并导入包

我们将使用 mistral-saba-24b 和 llama3-70b Groq 模型，因此除了 elasticsearch 和 llama-index，还需要安装 llama-index-llms-groq 包来处理与 LLM 的交互。

Groq 是一个推理服务，允许我们使用来自 Meta、Mistral 和 OpenAI 等提供商的多个开源模型。在本示例中，我们将使用它的免费层。你可以在此处获取稍后使用的 API KEY。

下面是所需依赖项的安装：Elasticsearch、LlamaIndex 核心库以及 LlamaIndex 的 Groq LLM 包。

pip install elasticsearch==8.17 llama-index llama-index-llms-groq

我们首先导入一些依赖项，用于处理环境变量（os）和管理 JSON 数据。

然后，我们导入 Elasticsearch 客户端，并使用 bulk helper 来通过 bulk API 进行索引。最后，我们导入 LlamaIndex 中的 Groq 类，以便与模型交互，并导入创建 workflow 所需的组件。

import os
import json
from getpass import getpassfrom elasticsearch import Elasticsearch
from elasticsearch.helpers import bulkfrom llama_index.llms.groq import Groq
from llama_index.core.workflow import (Event,StartEvent,StopEvent,Workflow,step,
)

2. 准备数据

设置密钥

我们设置 Groq 和 Elasticsearch 所需的环境变量。getpass 库允许我们通过提示输入密钥，并且不会显示输入的内容。

os.environ["GROQ_API_KEY"] = getpass("Groq Api key: ")
os.environ["ELASTIC_ENDPOINT"] = getpass("Elastic Endpoint: ")
os.environ["ELASTIC_API_KEY"] = getpass("Elastic Api key: ")

Elasticsearch 客户端

Elasticsearch 客户端负责与 Elasticsearch 进行连接，并允许我们使用 Python 库与 Elasticsearch 进行交互。

_client = Elasticsearch(os.environ["ELASTIC_ENDPOINT"],api_key=os.environ["ELASTIC_API_KEY"],
)

将数据导入 Elasticsearch

我们将以酒店房间为例创建一个索引：

INDEX_NAME="hotel-rooms"

映射

我们将使用文本类型字段（text）用于需要执行全文搜索的属性；使用 "keyword" 类型用于需要应用筛选或排序的字段，使用 "byte/integer" 类型用于数字。

try:_client.indices.create(index=INDEX_NAME,body={"mappings": {"properties": {"room_name": {"type": "text"},"description": {"type": "text"},"price_per_night": {"type": "integer"},"beds": {"type": "byte"},"features": {"type": "keyword"},}}},)print("index created successfully")
except Exception as e:print(f"Error creating inference endpoint: {e.info['error']['root_cause'][0]['reason'] }")

将文档导入 Elasticsearch

我们将导入一些酒店房间和设施，以便用户可以提问，我们可以将这些问题转化为针对文档的 Elasticsearch 查询。

documents = [{"room_name": "Standard Room","beds": 1,"description": "A cozy room with a comfortable queen-size bed, ideal for solo travelers or couples.","price_per_night": 80,"features": ["air conditioning", "wifi", "flat-screen TV", "mini fridge"]},{"room_name": "Deluxe Room","beds": 1,"description": "Spacious room with a king-size bed and modern amenities for a luxurious stay.","price_per_night": 120,"features": ["air conditioning", "wifi", "smart TV", "mini bar", "city view"]},{"room_name": "Family Room","beds": 2,"description": "A large room with two queen-size beds, perfect for families or small groups.","price_per_night": 150,"features": ["air conditioning", "wifi", "flat-screen TV", "sofa", "bath tub"]},{"room_name": "Suite","beds": 1,"description": "An elegant suite with a separate living area, offering maximum comfort and luxury.","price_per_night": 200,"features": ["air conditioning", "wifi", "smart TV", "jacuzzi", "balcony"]},{"room_name": "Penthouse Suite","beds": 1,"description": "The ultimate luxury experience with a panoramic view and top-notch amenities.","price_per_night": 350,"features": ["air conditioning", "wifi", "private terrace", "jacuzzi", "exclusive lounge access"]},{"room_name": "Single Room","beds": 1,"description": "A compact and comfortable room designed for solo travelers on a budget.","price_per_night": 60,"features": ["wifi", "air conditioning", "desk", "flat-screen TV"]},{"room_name": "Double Room","beds": 1,"description": "A well-furnished room with a queen-size bed, ideal for couples or business travelers.","price_per_night": 100,"features": ["air conditioning", "wifi", "mini fridge", "work desk"]},{"room_name": "Executive Suite","beds": 1,"description": "A high-end suite with premium furnishings and exclusive business amenities.","price_per_night": 250,"features": ["air conditioning", "wifi", "smart TV", "conference table", "city view"]},{"room_name": "Honeymoon Suite","beds": 1,"description": "A romantic suite with a king-size bed, perfect for newlyweds and special occasions.","price_per_night": 220,"features": ["air conditioning", "wifi", "hot tub", "romantic lighting", "balcony"]},{"room_name": "Presidential Suite","beds": 2,"description": "A luxurious suite with separate bedrooms and a living area, offering first-class comfort.","price_per_night": 500,"features": ["air conditioning", "wifi", "private dining area", "personal butler service", "exclusive lounge access"]}
]

我们将 JSON 文档解析为一个 bulk Elasticsearch 请求。

def build_data():for doc in documents:yield {"_index": INDEX_NAME, "_source": doc}try:success, errors = bulk(_client, build_data())print(f"{success} documents indexed successfully")if errors:print("Errors during indexing:", errors)except Exception as e:print(f"Error: {str(e)}")

3. LlamaIndex Workflows

我们需要创建一个类，包含将 Elasticsearch 映射发送到 LLM、运行查询并处理错误所需的函数。

class ElasticsearchRequest:@staticmethoddef get_mappings(_es_client: Elasticsearch):"""Get the mappings of the Elasticsearch index."""return _es_client.indices.get_mapping(index=INDEX_NAME)@staticmethodasync def do_es_query(query: str, _es_client: Elasticsearch):"""Execute an Elasticsearch query and return the results as a JSON string."""try:parsed_query = json.loads(query)if "query" not in parsed_query:return Exception("Error: Query JSON must contain a 'query' key")  # if the query is not a valid JSON return an errorresponse = _es_client.search(index=INDEX_NAME, body=parsed_query)hits = response["hits"]["hits"]if not hits or len(hits) == 0:return Exception("Query has not found any results")  # if the query has no results return an errorreturn json.dumps([hit["_source"] for hit in hits], indent=2)except json.JSONDecodeError:return Exception("Error: Query JSON no valid format")except Exception as e:return Exception(str(e))

Workflow 提示

EXTRACTION_PROMPT 将提供用户的问题，并将索引映射传递给 LLM，以便它返回一个 Elasticsearch 查询。

然后，REFLECTION_PROMPT 将帮助 LLM 在出现错误时进行修正，方法是提供来自 EXTRACTION_PROMPT 的输出以及查询引起的错误。

EXTRACTION_PROMPT = """
Context information is below:
---------------------
{passage}
---------------------Given the context information and not prior knowledge, create a Elasticsearch query from the information in the context.
The query must return the documents that match with query and the context information and the query used for retrieve the results.
{schema}"""REFLECTION_PROMPT = """
You already created this output previously:
---------------------
{wrong_answer}
---------------------This caused the error: {error}Try again; the response must contain only valid Elasticsearch queries. Do not add any sentence before or after the JSON object.
Do not repeat the query.
"""

Workflow 事件

我们创建了类来处理提取和查询验证事件：

class ExtractionDone(Event):output: strpassage: strclass ValidationErrorEvent(Event):error: strwrong_output: strpassage: str

Workflow

现在，让我们把一切结合起来。首先，我们需要将最大尝试次数设置为 3 次，以便更换模型。

接下来，我们将使用在 workflow 中配置的模型进行 extraction。我们验证事件是否为 StartEvent；如果是，我们捕获模型和问题（段落）。

然后，我们运行 validation 步骤，即尝试在 Elasticsearch 中运行提取的查询。如果没有错误，我们生成一个 StopEvent 并停止流程。否则，我们发出 ValidationErrorEvent 并重复第 1 步，提供错误信息以进行修正，然后返回验证步骤。如果在 3 次尝试后仍然没有有效查询，我们更换模型并重复此过程，直到达到 60 秒的超时运行时间参数。

class ReflectionWorkflow(Workflow):model_retries: int = 0max_retries: int = 3@step()async def extract(self, ev: StartEvent | ValidationErrorEvent) -> StopEvent | ExtractionDone:print("\n=== EXTRACT STEP ===\n")if isinstance(ev, StartEvent):model = ev.get("model")passage = ev.get("passage")if not passage:return StopEvent(result="Please provide some text in input")reflection_prompt = ""elif isinstance(ev, ValidationErrorEvent):passage = ev.passagemodel = ev.modelreflection_prompt = REFLECTION_PROMPT.format(wrong_answer=ev.wrong_output, error=ev.error)llm = Groq(model=model, api_key=os.environ["GROQ_API_KEY"])prompt = EXTRACTION_PROMPT.format(passage=passage, schema=ElasticsearchRequest.get_mappings(_client))if reflection_prompt:prompt += reflection_promptoutput = await llm.acomplete(prompt)print(f"MODEL: {model}")print(f"OUTPUT: {output}")print("=================\n")return ExtractionDone(output=str(output), passage=passage, model=model)@step()async def validate(self, ev: ExtractionDone) -> StopEvent | ValidationErrorEvent:print("\n=== VALIDATE STEP ===\n")try:results = await ElasticsearchRequest.do_es_query(ev.output, _client)self.model_retries += 1if self.model_retries > self.max_retries and ev.model != "llama3-70b-8192":print(f"Max retries for model {ev.model} reached, changing model\n")model = "llama3-70b-8192"  # if the some error occurs, the model will be changed to llama3-70b-8192else:model = ev.modelprint(f"Elasticsearch results: {results}")if isinstance(results, Exception):print("STATUS: Validation failed, retrying...\n")print("===================\n")return ValidationErrorEvent(error=str(results),wrong_output=ev.output,passage=ev.passage,model=model,)# print("results: ", results)except Exception as e:print("STATUS: Validation failed, retrying...\n")print("===================\n")return ValidationErrorEvent(error=str(e),wrong_output=ev.output,passage=ev.passage,model=model,)return StopEvent(result=ev.output)

4. 执行 workflow 任务

我们将进行以下搜索：具有智能电视、wifi、按摩浴缸且每晚价格低于 300 的房间。我们将首先使用 mistral-saba-24b 模型，并根据需要按照我们的流程切换到 llama3-70b-8192。

w = ReflectionWorkflow(timeout=60, verbose=True)user_prompt = "Rooms with smart TV, wifi, jacuzzi and price per night less than 300"result = await w.run(passage=f"I need the best possible query for documents that have: {user_prompt}",model="mistral-saba-24b",
)print(result)

结果

（格式化以便于阅读）

=== EXTRACT STEP ===

MODEL: mistral-saba-24b

OUTPUT:

{"query": {"bool": {"must": [{ "match": { "features": "smart TV" }},{ "match": { "features": "wifi" }},{ "match": { "features": "jacuzzi" }},{ "range": { "price_per_night": { "lt": 300 }}}]}}
}

Step extract produced event ExtractionDone

Running step validate

=== VALIDATE STEP ===

Max retries for model mistral-saba-24b reached, changing model

Elasticsearch 结果：

Error: Query JSON no valid format
STATUS: Validation failed, retrying...

Step validate produced event ValidationErrorEvent

Running step extract

=== EXTRACT STEP ===

MODEL: llama3-70b-8192

OUTPUT:

{"query": {"bool": {"filter": [{ "term": { "features": "smart TV" }},{ "term": { "features": "wifi" }},{ "term": { "features": "jacuzzi" }},{ "range": { "price_per_night": { "lt": 300 }}}]}}
}

Step extract produced event ExtractionDone

Running step validate

=== VALIDATE STEP ===

Elasticsearch 结果：

[{"room_name": "Suite","beds": 1,"description": "An elegant suite with a separate living area, offering maximum comfort and luxury.","price_per_night": 200,"features": ["air conditioning","wifi","smart TV","jacuzzi","balcony"]}
]

Step validate produced event StopEvent

{"query": {"bool": {"filter": [{ "term": { "features": "smart TV" }},{ "term": { "features": "wifi" }},{ "term": { "features": "jacuzzi" }},{ "range": { "price_per_night": { "lt": 300 }}}]}}
}

在上面的示例中，查询失败是因为 mistral-saba-24b 模型返回的结果是 Markdown 格式，开始时添加了 json，结尾处有 。而 llama3-70b-8192 模型直接返回了使用 JSON 格式的查询。根据我们的需求，我们可以捕获、验证并测试不同的错误，或在多次尝试后构建回退机制。

结论

LlamaIndex workflows 提供了一种有趣的替代方案，可以通过事件和步骤开发 agentic 流程。通过仅几行代码，我们成功创建了一个能够通过可互换模型进行自动纠错的系统。

我们如何改进这个流程？

• 除了映射外，我们可以将可能的确切过滤器值发送给 LLM，从而减少由于拼写错误的过滤器导致的无结果查询。为此，我们可以对特性进行 terms 聚合并将结果展示给 LLM。

• 添加对常见问题的代码修正，例如我们遇到的 Markdown 问题，以提高成功率。

• 添加处理有效查询但没有结果的方式。例如，移除某个过滤器并重新尝试，向用户提供建议。LLM 可以根据上下文选择要移除的过滤器。

• 向提示中添加更多上下文信息，如用户偏好或之前的搜索，以便我们可以在 Elasticsearch 结果的基础上提供定制化的建议。

你想尝试其中一个吗？

想要获得 Elastic 认证吗？了解下一期 Elasticsearch 工程师培训的时间！

Elasticsearch 拥有许多新特性，帮助你为特定用例构建最佳搜索解决方案。深入探索我们的示例笔记本，了解更多内容，开始免费云试用，或立即在本地机器上尝试 Elastic。

原文：Using LlamaIndex Workflows with Elasticsearch - Elasticsearch Labs