feat: 添加 AI 搜索技能组配置脚本 configure_skillset.py，支持枚举数据源/技能组/索引器及通过 az 读取认知服务密钥并创建 Blob 数据源+技能组+索引器

Made-with: Cursor
2026-03-15 06:36:33 +00:00
parent 6cca1a0f9d
commit e03d3c7f52
2 changed files with 313 additions and 1 deletions
@@ -0,0 +1,307 @@
+#!/usr/bin/env python3
+"""
+在 Azure AI 搜索中枚举并配置：数据源、技能组、索引器。
+技能组是 AI 搜索内置功能，内置技能（如 OCR）在技能组内声明，由索引器执行。
+创建技能组时若未配置 COGNITIVE_SERVICES_KEY，会通过 az login 枚举 Azure 认知服务/内容识别并自动读取终结点与密钥。
+"""
+import json
+import os
+import subprocess
+import sys
+
+# 加载 local.settings.json
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+SETTINGS_PATH = os.path.join(SCRIPT_DIR, "local.settings.json")
+if os.path.exists(SETTINGS_PATH):
+    with open(SETTINGS_PATH, "r", encoding="utf-8") as f:
+        for k, v in json.load(f).get("Values", {}).items():
+            os.environ.setdefault(k, v)
+
+ENDPOINT = os.environ.get("AZURE_SEARCH_ENDPOINT", "").rstrip("/")
+API_KEY = os.environ.get("AZURE_SEARCH_API_KEY", "")
+API_VERSION = "2024-07-01"
+
+
+def _headers():
+    return {"Content-Type": "application/json", "api-key": API_KEY}
+
+
+def _get(path):
+    import urllib.request
+    req = urllib.request.Request(f"{ENDPOINT}{path}", headers=_headers(), method="GET")
+    with urllib.request.urlopen(req) as resp:
+        return json.loads(resp.read().decode())
+
+
+def _put(path, body):
+    import urllib.request
+    data = json.dumps(body).encode("utf-8")
+    req = urllib.request.Request(f"{ENDPOINT}{path}", data=data, headers=_headers(), method="PUT")
+    with urllib.request.urlopen(req) as resp:
+        return json.loads(resp.read().decode()) if resp.length else {}
+
+
+def _post(path, body):
+    import urllib.request
+    data = json.dumps(body).encode("utf-8")
+    req = urllib.request.Request(f"{ENDPOINT}{path}", data=data, headers=_headers(), method="POST")
+    with urllib.request.urlopen(req) as resp:
+        return json.loads(resp.read().decode()) if resp.length else {}
+
+
+def get_cognitive_services_from_az():
+    """
+    通过 az login 枚举当前订阅下的认知服务/内容识别（Microsoft.CognitiveServices/accounts），
+    并读取每个资源的终结点与 key。返回列表 [{name, resourceGroup, endpoint, key}, ...]。
+    若 az 未登录或未安装，返回 []。
+    """
+    out = []
+    try:
+        # 枚举认知服务账户（含多服务/内容识别/Foundry）
+        r = subprocess.run(
+            ["az", "cognitiveservices", "account", "list", "-o", "json"],
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        if r.returncode != 0 or not r.stdout.strip():
+            return out
+        accounts = json.loads(r.stdout)
+        if not isinstance(accounts, list):
+            return out
+        for acc in accounts:
+            name = acc.get("name") or ""
+            rg = acc.get("resourceGroup") or ""
+            # 终结点可能在 properties.endpoint 或 properties.endpoint 里
+            props = acc.get("properties") or {}
+            endpoint = (props.get("endpoint") or "").rstrip("/")
+            if not name or not rg:
+                continue
+            # 获取 key
+            kr = subprocess.run(
+                ["az", "cognitiveservices", "account", "keys", "list", "--name", name, "--resource-group", rg, "-o", "json"],
+                capture_output=True,
+                text=True,
+                timeout=15,
+            )
+            key = ""
+            if kr.returncode == 0 and kr.stdout.strip():
+                keys_obj = json.loads(kr.stdout)
+                key = (keys_obj.get("key1") or keys_obj.get("key2") or "").strip()
+            out.append({"name": name, "resourceGroup": rg, "endpoint": endpoint, "key": key})
+        return out
+    except (FileNotFoundError, json.JSONDecodeError, subprocess.TimeoutExpired) as e:
+        return out
+
+
+def list_cognitive_services():
+    """枚举并打印 Azure 中的认知服务/内容识别资源（终结点与密钥占位），便于确认。"""
+    print("=== Azure 认知服务 / 内容识别 (az login 枚举) ===\n")
+    items = get_cognitive_services_from_az()
+    if not items:
+        print("未发现认知服务资源，请确认：1) 已执行 az login  2) 当前订阅下已创建认知服务/内容识别。")
+        return
+    for i, x in enumerate(items, 1):
+        key_preview = (x["key"][:8] + "...") if x.get("key") else "(未读取)"
+        print(f"  [{i}] {x['name']}")
+        print(f"      资源组: {x['resourceGroup']}")
+        print(f"      终结点: {x.get('endpoint') or '(空)'}")
+        print(f"      Key:    {key_preview}")
+        print()
+    print("创建技能组时将自动使用上述第一个资源的 key（或通过 COGNITIVE_SERVICES_KEY 指定）。")
+
+
+def enumerate_services():
+    """枚举当前 Search 服务中的：数据源、技能组、索引器、索引"""
+    if not ENDPOINT or not API_KEY:
+        print("缺少 AZURE_SEARCH_ENDPOINT 或 AZURE_SEARCH_API_KEY（请在 local.settings.json 中配置）")
+        return
+    base = f"?api-version={API_VERSION}"
+    print("=== 数据源 (Data Sources) ===")
+    try:
+        r = _get(f"/datasources{base}")
+        for x in r.get("value", []):
+            print(f"  - {x.get('name')}  type={x.get('type')}")
+        if not r.get("value"):
+            print("  (无)")
+    except Exception as e:
+        print(f"  错误: {e}")
+    print("\n=== 技能组 (Skillsets) ===")
+    try:
+        r = _get(f"/skillsets{base}")
+        for x in r.get("value", []):
+            print(f"  - {x.get('name')}  skills={len(x.get('skills', []))}")
+        if not r.get("value"):
+            print("  (无)")
+    except Exception as e:
+        print(f"  错误: {e}")
+    print("\n=== 索引器 (Indexers) ===")
+    try:
+        r = _get(f"/indexers{base}")
+        for x in r.get("value", []):
+            print(f"  - {x.get('name')}  dataSource={x.get('dataSourceName')}  skillset={x.get('skillsetName')}  index={x.get('targetIndexName')}")
+        if not r.get("value"):
+            print("  (无)")
+    except Exception as e:
+        print(f"  错误: {e}")
+    print("\n=== 索引 (Indexes) ===")
+    try:
+        r = _get(f"/indexes{base}")
+        for x in r.get("value", []):
+            print(f"  - {x.get('name')}")
+        if not r.get("value"):
+            print("  (无)")
+    except Exception as e:
+        print(f"  错误: {e}")
+
+
+def create_datasource_skillset_indexer():
+    """
+    创建：Blob 数据源、技能组（OCR + Merge + 挂载 Azure 内容识别）、
+    用于 Blob 的索引、索引器。全部在 AI 搜索内完成。
+    """
+    conn_str = os.environ.get("INDEXER_BLOB_CONNECTION_STRING") or os.environ.get("BLOB_STORAGE_CONNECTION_STRING") or os.environ.get("AzureWebJobsStorage")
+    container = os.environ.get("INDEXER_BLOB_CONTAINER", "documents")
+    cognitive_key = os.environ.get("COGNITIVE_SERVICES_KEY") or os.environ.get("CONTENT_SAFETY_KEY")
+    index_name = os.environ.get("INDEXER_TARGET_INDEX", "openclaw-blob")
+
+    if not ENDPOINT or not API_KEY:
+        print("缺少 AZURE_SEARCH_ENDPOINT 或 AZURE_SEARCH_API_KEY")
+        return False
+    if not conn_str:
+        print("缺少 Blob 连接字符串，请设置 INDEXER_BLOB_CONNECTION_STRING 或 BLOB_STORAGE_CONNECTION_STRING")
+        return False
+    # 未配置 key 时，通过 az 枚举认知服务/内容识别并自动读取 key
+    if not cognitive_key:
+        candidates = get_cognitive_services_from_az()
+        # 优先名称含「内容识别」或 content / cognitive 的资源
+        for c in candidates:
+            if c.get("key") and ("内容识别" in (c.get("name") or "") or "content" in (c.get("name") or "").lower() or "cognitive" in (c.get("name") or "").lower() or "foundry" in (c.get("name") or "").lower()):
+                cognitive_key = c["key"]
+                print(f"已从 Azure 读取认知服务: {c['name']} (资源组: {c['resourceGroup']})，终结点: {c.get('endpoint') or '(无)'}")
+                break
+        if not cognitive_key and candidates:
+            c = candidates[0]
+            if c.get("key"):
+                cognitive_key = c["key"]
+                print(f"已从 Azure 读取认知服务: {c['name']} (资源组: {c['resourceGroup']})，终结点: {c.get('endpoint') or '(无)'}")
+    if not cognitive_key:
+        print("未配置 COGNITIVE_SERVICES_KEY 且 az 未枚举到可用认知服务，请先 az login 并确保订阅下有认知服务/内容识别，或手动设置 COGNITIVE_SERVICES_KEY")
+        return False
+
+    base = f"?api-version={API_VERSION}"
+
+    # 1. 数据源
+    ds_name = "openclaw-blob-datasource"
+    ds_body = {
+        "name": ds_name,
+        "type": "azureblob",
+        "credentials": {"connectionString": conn_str},
+        "container": {"name": container},
+    }
+    try:
+        _put(f"/datasources('{ds_name}'){base}", ds_body)
+        print(f"已创建/更新数据源: {ds_name}  container={container}")
+    except Exception as e:
+        print(f"创建数据源失败: {e}")
+        return False
+
+    # 2. 技能组（仅内置技能：OCR、Merge；计费挂载 Azure 内容识别）
+    ss_name = "openclaw-blob-skillset"
+    skillset_body = {
+        "name": ss_name,
+        "skills": [
+            {
+                "@odata.type": "#Microsoft.Skills.Vision.OcrSkill",
+                "context": "/document/normalized_images/*",
+                "defaultLanguageCode": None,
+                "detectOrientation": True,
+                "inputs": [{"name": "image", "source": "/document/normalized_images/*"}],
+                "outputs": [{"name": "text", "targetName": "ocr_text"}],
+            },
+            {
+                "@odata.type": "#Microsoft.Skills.Text.MergeSkill",
+                "context": "/document",
+                "insertPreTag": " ",
+                "insertPostTag": " ",
+                "inputs": [
+                    {"name": "text", "source": "/document/content"},
+                    {"name": "itemsToInsert", "source": "/document/normalized_images/*/ocr_text"},
+                    {"name": "offsets", "source": "/document/normalized_images/*/contentOffset"},
+                ],
+                "outputs": [{"name": "mergedText", "targetName": "merged_text"}],
+            },
+        ],
+        "cognitiveServices": {
+            "@odata.type": "#Microsoft.Azure.Search.CognitiveServicesByKey",
+            "description": "Azure 内容识别，用于 OCR 等内置技能计费",
+            "key": cognitive_key,
+        },
+    }
+    try:
+        _put(f"/skillsets('{ss_name}'){base}", skillset_body)
+        print(f"已创建/更新技能组: {ss_name}  (OCR + Merge，已挂载 cognitiveServices)")
+    except Exception as e:
+        print(f"创建技能组失败: {e}")
+        return False
+
+    # 3. 索引（供 Blob 管道写入）
+    index_body = {
+        "name": index_name,
+        "fields": [
+            {"name": "id", "type": "Edm.String", "key": True, "filterable": True},
+            {"name": "content", "type": "Edm.String", "searchable": True},
+            {"name": "metadata_storage_name", "type": "Edm.String", "filterable": True, "sortable": True},
+            {"name": "metadata_storage_path", "type": "Edm.String", "filterable": True},
+            {"name": "metadata_storage_last_modified", "type": "Edm.DateTimeOffset", "filterable": True, "sortable": True},
+        ],
+    }
+    try:
+        _put(f"/indexes('{index_name}'){base}", index_body)
+        print(f"已创建/更新索引: {index_name}")
+    except Exception as e:
+        print(f"创建索引失败: {e}")
+        return False
+
+    # 4. 索引器（拉取 Blob → 执行技能组 → 写入索引）
+    idx_name = "openclaw-blob-indexer"
+    indexer_body = {
+        "name": idx_name,
+        "dataSourceName": ds_name,
+        "targetIndexName": index_name,
+        "skillsetName": ss_name,
+        "parameters": {
+            "configuration": {
+                "dataToExtract": "contentAndMetadata",
+                "imageAction": "generateNormalizedImages",
+                "parsingMode": "default",
+            },
+        },
+        "fieldMappings": [
+            {"sourceFieldName": "metadata_storage_path", "targetFieldName": "id"},
+            {"sourceFieldName": "metadata_storage_name", "targetFieldName": "metadata_storage_name"},
+            {"sourceFieldName": "metadata_storage_path", "targetFieldName": "metadata_storage_path"},
+            {"sourceFieldName": "metadata_storage_last_modified", "targetFieldName": "metadata_storage_last_modified"},
+        ],
+        "outputFieldMappings": [
+            {"sourceFieldName": "/document/merged_text", "targetFieldName": "content"},
+        ],
+    }
+    try:
+        _put(f"/indexers('{idx_name}'){base}", indexer_body)
+        print(f"已创建/更新索引器: {idx_name}")
+    except Exception as e:
+        print(f"创建索引器失败: {e}")
+        return False
+
+    print("\n配置完成。可将文件放入 Blob 容器后运行索引器。")
+    return True
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1 and sys.argv[1] == "--create":
+        create_datasource_skillset_indexer()
+    elif len(sys.argv) > 1 and sys.argv[1] == "--list-cognitive":
+        list_cognitive_services()
+    else:
+        enumerate_services()
@@ -10,6 +10,11 @@

    "OPENAI_BASE_URL": "https://litellm.example.com/v1",
    "OPENAI_API_KEY": "sk-your-key",
-    "EMBEDDING_MODEL": "taiji/text-embedding-3-small"
+    "EMBEDDING_MODEL": "taiji/text-embedding-3-small",
+
+    "INDEXER_BLOB_CONTAINER": "documents",
+    "INDEXER_BLOB_CONNECTION_STRING": "<Blob 存储连接字符串，用于索引器数据源>",
+    "COGNITIVE_SERVICES_KEY": "<Azure 内容识别 / Foundry 资源 key，用于技能组计费>",
+    "INDEXER_TARGET_INDEX": "openclaw-blob"
  }
 }