阿里云上有很多的服务。在今天的文章中我重点介绍如何创供我们使用的推理端点连接器并使用代码写入我们的数据。在进行下面的操作之前建议观看视频阿里云 Elasticsearch 开通试用到登陆 Kibana 教程。https://www.bilibili.com/video/BV13pNJ6oEf6/我们可以使用阿里云提供的链接进入到试用页面如果你已经创建好了自己的 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 endpoints # 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_keyes_api_key, verify_certsTrue, request_timeoutREQUEST_TIMEOUT ) client.info() return client except TlsError: print(Certificate could not be verified (self-signed?), retrying with verify_certsFalse, filesys.stderr) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) client Elasticsearch( es_url, api_keyes_api_key, verify_certsFalse, request_timeoutREQUEST_TIMEOUT ) client.info() return client def ensure_index(client: Elasticsearch) - None: if client.indices.exists(indexINDEX_NAME): print(fIndex {INDEX_NAME} already exists, skipping creation) return client.indices.create(indexINDEX_NAME, bodyINDEX_MAPPING) client.cluster.health(indexINDEX_NAME, wait_for_statusyellow, timeout30s) print(fCreated 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, encodingutf-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_sizeBULK_CHUNK_SIZE, raise_on_errorFalse, ) except BulkIndexError as e: print(fBulk indexing failed: {e}, filesys.stderr) sys.exit(1) print(fIndexed {success} documents into {INDEX_NAME}) if errors: print(f{len(errors)} documents failed to index, filesys.stderr) for err in errors[:5]: print(err, filesys.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 的知识请在链接里进行查看。好了。我基本上把我所想要讲的都讲了。祝大家学习愉快
如何为阿里云 Elasticsearch 创建推理端点，连接器，Workflow 并写入数据
阿里云上有很多的服务。在今天的文章中我重点介绍如何创供我们使用的推理端点连接器并使用代码写入我们的数据。在进行下面的操作之前建议观看视频阿里云 Elasticsearch 开通试用到登陆 Kibana 教程。https://www.bilibili.com/video/BV13pNJ6oEf6/我们可以使用阿里云提供的链接进入到试用页面如果你已经创建好了自己的 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 endpoints # 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_keyes_api_key, verify_certsTrue, request_timeoutREQUEST_TIMEOUT ) client.info() return client except TlsError: print(Certificate could not be verified (self-signed?), retrying with verify_certsFalse, filesys.stderr) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) client Elasticsearch( es_url, api_keyes_api_key, verify_certsFalse, request_timeoutREQUEST_TIMEOUT ) client.info() return client def ensure_index(client: Elasticsearch) - None: if client.indices.exists(indexINDEX_NAME): print(fIndex {INDEX_NAME} already exists, skipping creation) return client.indices.create(indexINDEX_NAME, bodyINDEX_MAPPING) client.cluster.health(indexINDEX_NAME, wait_for_statusyellow, timeout30s) print(fCreated 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, encodingutf-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_sizeBULK_CHUNK_SIZE, raise_on_errorFalse, ) except BulkIndexError as e: print(fBulk indexing failed: {e}, filesys.stderr) sys.exit(1) print(fIndexed {success} documents into {INDEX_NAME}) if errors: print(f{len(errors)} documents failed to index, filesys.stderr) for err in errors[:5]: print(err, filesys.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 的知识请在链接里进行查看。好了。我基本上把我所想要讲的都讲了。祝大家学习愉快
