阿里云上有很多的服务。在今天的文章中,我重点介绍如何创供我们使用的推理端点,连接器,并使用代码写入我们的数据。在进行下面的操作之前,建议观看视频:阿里云 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 的知识,请在链接里进行查看。
好了。我基本上把我所想要讲的都讲了。祝大家学习愉快!