类别：provider

OpenSearch 向量搜索版

使用 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 查询示例。请按需替换 cluster/table 名称。
    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 &&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 错误：请检查 schema 字段定义与 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