aliyun-opensearch-search

Category：提供商

OpenSearch Vector Search Edition

使用 ha3engine SDK 推送文档并执行 HA/SQL 搜索。此技能仅聚焦于 API/SDK 用法（不涉及控制台操作）。

前置条件

• 安装 SDK（建议在 venv 中安装，以避免 PEP 668 限制）：

python3 -m venv .venv
. .venv/bin/activate
python -m pip install alibabacloud-ha3engine

• 通过环境变量提供连接配置：
  • OPENSEARCH_ENDPOINT（API 域名）
  • OPENSEARCH_INSTANCE_ID
  • OPENSEARCH_USERNAME
  • OPENSEARCH_PASSWORD
  • OPENSEARCH_DATASOURCE（数据源名称）
  • OPENSEARCH_PK_FIELD（主键字段名称）

快速开始（推送 + 搜索）

import os
from alibabacloud_ha3engine import models, client
from Tea.exceptions import TeaException, RetryError

cfg = models.Config(
    endpoint=os.getenv("OPENSEARCH_ENDPOINT"),
    instance_id=os.getenv("OPENSEARCH_INSTANCE_ID"),
    protocol="http",
    access_user_name=os.getenv("OPENSEARCH_USERNAME"),
    access_pass_word=os.getenv("OPENSEARCH_PASSWORD"),
)
ha3 = client.Client(cfg)

def push_docs():
    data_source = os.getenv("OPENSEARCH_DATASOURCE")
    pk_field = os.getenv("OPENSEARCH_PK_FIELD", "id")

    documents = [
        {"fields": {"id": 1, "title": "hello", "content": "world"}, "cmd": "add"},
        {"fields": {"id": 2, "title": "faq", "content": "vector search"}, "cmd": "add"},
    ]
    req = models.PushDocumentsRequestModel({}, documents)
    return ha3.push_documents(data_source, pk_field, req)


def search_ha():
    # HA query example. Replace cluster/table names as needed.
    query_str = (
        "config=hit:5,format:json,qrs_chain:search"
        "&&query=title:hello"
        "&&cluster=general"
    )
    ha_query = models.SearchQuery(query=query_str)
    req = models.SearchRequestModel({}, ha_query)
    return ha3.search(req)

try:
    print(push_docs().body)
    print(search_ha())
except (TeaException, RetryError) as e:
    print(e)

脚本快速开始

python skills/ai/search/aliyun-opensearch-search/scripts/quickstart.py

环境变量：

• OPENSEARCH_ENDPOINT
• OPENSEARCH_INSTANCE_ID
• OPENSEARCH_USERNAME
• OPENSEARCH_PASSWORD
• OPENSEARCH_DATASOURCE
• OPENSEARCH_PK_FIELD（可选，默认值为 id）
• OPENSEARCH_CLUSTER（可选，默认值为 general）

可选参数：--cluster、--hit、--query。

SQL 风格搜索

from alibabacloud_ha3engine import models

sql = "select * from <indexTableName>&&kvpair=trace:INFO;formatType:json"
sql_query = models.SearchQuery(sql=sql)
req = models.SearchRequestModel({}, sql_query)
resp = ha3.search(req)
print(resp)

Claude Code/Codex 使用说明

• 使用 push_documents 执行添加/删除更新。
• 较大的查询字符串（>30KB）应使用 RESTful 搜索 API。
• HA 查询适用于快速、灵活地进行向量 + 关键词检索；SQL 则适用于结构化数据。

错误处理

• 身份验证错误：检查用户名/密码和实例访问权限。
• 推送时出现 4xx 错误：检查架构字段和 pk_field 是否一致。
• 5xx 错误：采用退避策略重试。

验证

mkdir -p output/aliyun-opensearch-search
for f in skills/ai/search/aliyun-opensearch-search/scripts/*.py; do
  python3 -m py_compile "$f"
done
echo "py_compile_ok" > output/aliyun-opensearch-search/validate.txt

通过标准：命令以 0 退出，且已生成 output/aliyun-opensearch-search/validate.txt。

输出与证据

• 将构件、命令输出和 API 响应摘要保存在 output/aliyun-opensearch-search/ 下。
• 在证据文件中包含关键参数（区域/资源 ID/时间范围），以便复现。

工作流

1. 确认用户意图、区域、标识符，以及操作是只读还是会修改数据。
2. 先运行一个最小的只读查询，以验证连接和权限。
3. 使用明确的参数和受限范围执行目标操作。
4. 验证结果并保存输出/证据文件。

参考资料

• SDK 软件包：alibabacloud-ha3engine

• 示例：OpenSearch 文档中的数据推送和 HA/SQL 搜索示例

• 来源列表：references/sources.md