如何为阿里云 Elasticsearch 创建推理端点,连接器,Workflow 并写入数据
阿里云上有很多的服务。在今天的文章中,我重点介绍如何创供我们使用的推理端点,连接器,并使用代码写入我们的数据。在进行下面的操作之前,建议观看视频:阿里云 Elasticsearch 开通试用到登陆 Kibana 教程。
我们可以使用阿里云提供的链接进入到试用页面:

如果你已经创建好了自己的 Elasticsearch 集群,那么再次点击那个链接,你就会看到上面的页面。点击上面的我的试用:

我们点击上面的控制台:




我们可以进行上面的测试。你也可以使用如下的命令来进行测试:
curl -u elastic:<YourPassword> http://es-cn-rcn4v4kcy0001wg52.public.elasticsearch.aliyuncs.com:9200/

同样地,你需要配置 Kibana 的公网地址:


设置完毕后,我们可以直接访问 Kibana:


这样我们就进入到 Kibana 界面了。
创建 API Key
我们进入到如下的页面来申请 阿里云 ES API Key:


我们保存创建的 API key,并在之后的配置中使用。
创建嵌入推理端点
点击 Kibana 上面的 Stack Management,并进入到连机器页面:



参考如下的文章:



注意:上面的 host 就是指的在 API key 申请的那个地址。

我们可以选择一个支持多语言的模型,比如上面的 ops-text-embedding-002:

上面的 URL 是由 API key 中的地址组成的:
http://default-8s7v.platform-cn-beijing.opensearch.aliyuncs.com/compatible-mode/v1/embeddings
注意:你需要根据自己在 API key 申请页面中的配置进行相应的修改。
点击上面的设置。我们可以进入到 Dev Tools 来进行测试:
POST _inference/alibaba_text_embedding
{
"input": "The sky above the port was the color of television tuned to a dead channel."
}

很显然,它能帮我们生成我们的向量。我们接下来使用之前在文章 “如何写入 IMDB 电影数据并针对它运用 AI Agent Builder 对它进行分享” 示范的那样。我们改写我们的程序如下:
#!/usr/bin/env python3
"""Ingest imdb_movies.csv into Elasticsearch, using connection settings from .env."""
import csv
import os
import sys
import urllib3
from dotenv import load_dotenv
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk, BulkIndexError
from elastic_transport import TlsError
INDEX_NAME = "imdb"
CSV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "imdb_movies.csv")
# Kept low (< 32) because the openai-text_embedding-pmja5fdpvkc inference endpoint's
# backing service returns a 500 (fastjson getInputValue error) when a single
# inference call batches more than ~32 input texts.
BULK_CHUNK_SIZE = 25
REQUEST_TIMEOUT = 300
INDEX_MAPPING = {
"mappings": {
"properties": {
"budget_x": {"type": "double"},
"country": {"type": "keyword"},
"crew": {"type": "text"},
"date_x": {"type": "keyword"},
"genre": {"type": "keyword"},
"names": {"type": "text"},
"orig_lang": {"type": "keyword"},
"orig_title": {"type": "text"},
"overview": {"type": "text", "copy_to": ["overview_semantic"]},
"overview_semantic": {
"type": "semantic_text",
"inference_id": "openai-text_embedding-pmja5fdpvkc"
},
"revenue": {"type": "double"},
"score": {"type": "double"},
"status": {"type": "keyword"},
},
}
}
def build_client(es_url: str, es_api_key: str) -> Elasticsearch:
"""Connect to Elasticsearch, working for both trusted and self-signed TLS certs."""
try:
client = Elasticsearch(
es_url, api_key=es_api_key, verify_certs=True, request_timeout=REQUEST_TIMEOUT
)
client.info()
return client
except TlsError:
print("Certificate could not be verified (self-signed?), retrying with verify_certs=False", file=sys.stderr)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
client = Elasticsearch(
es_url, api_key=es_api_key, verify_certs=False, request_timeout=REQUEST_TIMEOUT
)
client.info()
return client
def ensure_index(client: Elasticsearch) -> None:
if client.indices.exists(index=INDEX_NAME):
print(f"Index '{INDEX_NAME}' already exists, skipping creation")
return
client.indices.create(index=INDEX_NAME, body=INDEX_MAPPING)
client.cluster.health(index=INDEX_NAME, wait_for_status="yellow", timeout="30s")
print(f"Created index '{INDEX_NAME}'")
def to_float(value):
value = (value or "").strip()
if not value:
return None
try:
return float(value)
except ValueError:
return None
def read_docs(csv_path: str):
with open(csv_path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for i, row in enumerate(reader):
doc = {
"names": (row.get("names") or "").strip(),
"date_x": (row.get("date_x") or "").strip(),
"score": to_float(row.get("score")),
"genre": [g.strip() for g in (row.get("genre") or "").split(",") if g.strip()],
"overview": (row.get("overview") or "").strip(),
"crew": (row.get("crew") or "").strip(),
"orig_title": (row.get("orig_title") or "").strip(),
"status": (row.get("status") or "").strip(),
"orig_lang": (row.get("orig_lang") or "").strip(),
"budget_x": to_float(row.get("budget_x")),
"revenue": to_float(row.get("revenue")),
"country": (row.get("country") or "").strip(),
}
yield {"_index": INDEX_NAME, "_id": str(i), "_source": doc}
def main() -> None:
load_dotenv()
es_url = os.environ["ES_URL"]
es_api_key = os.environ["ES_API_KEY"]
client = build_client(es_url, es_api_key)
ensure_index(client)
try:
success, errors = bulk(
client,
read_docs(CSV_PATH),
chunk_size=BULK_CHUNK_SIZE,
raise_on_error=False,
)
except BulkIndexError as e:
print(f"Bulk indexing failed: {e}", file=sys.stderr)
sys.exit(1)
print(f"Indexed {success} documents into '{INDEX_NAME}'")
if errors:
print(f"{len(errors)} documents failed to index", file=sys.stderr)
for err in errors[:5]:
print(err, file=sys.stderr)
if __name__ == "__main__":
main()
请注意在上面,我们使用了 "inference_id": "alibaba_text_embedding"。
运行我们的程序,它就可以把我们的数据写入到 Elasticsearch 中。
创建大模型连接器
我们可以使用如下的方式来连接大模型:

我们点击链接来查看有哪些模型可以使用。这个位于 API 申请的页面:


我们查看到我们需要的模型,比如上面的 qwen_plus。



我们在连接器里针对它进行如下的配置:

其中 URL 为:
http://default-8s7v.platform-cn-beijing.opensearch.aliyuncs.com//compatible-mode/v1/chat/completions
注意:你需要根据自己在 API key 申请页面中的配置进行相应的修改。


它表明我们的配置是成功的。
在 Agents 中配置并使用它
我们在 Kibana 的 Agents 中打开并使用它:



Hurray! 我们的 LLM 现在可以开始工作了。
配置 Workflow
在 9.3 的发布中,Workflow 在默认的情况下,是没有展现的。我们需要启动它:




有关更多关于 Workflows 的知识,请在链接里进行查看。
好了。我基本上把我所想要讲的都讲了。祝大家学习愉快!
更多推荐

所有评论(0)