refactor(auto-config): 端点/路径职责反转对齐端点可互换约定——提取提示词base_url改主机根(禁路径段)+chat_path改完整相对路径(含版本段);apply端点归一主机根+同(主机根,region)去重+timeout取max+删裸域拦截与协议补绑自检;_gen_templates/_ensure_step_profile path经_norm_relative_path归一(全URL剥scheme+host,保留版本段,不再剥base_url前缀)。配套pipeline-llm 3ea03be

pipeline_platform/platform_ability.py

@@ -353,10 +353,10 @@ _EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型
 只输出一个 JSON 对象（不要 markdown 代码块），字段：
 {
   "vendor_name": "供应商名称",
-  "base_url": "API 基础地址——必须取文档示例请求 URL 中除接口路径外的完整前缀（含 /api/v1 或 /compatible-mode/v1 等路径段，如 https://dashscope.aliyuncs.com/api/v1）。禁止只填裸域（如 https://dashscope.aliyuncs.com）——裸域拼接口路径必 404",
+  "base_url": "API 主机根——只填 scheme+域名（如 https://dashscope.aliyuncs.com），禁止带任何路径段（/api/v1、/compatible-mode/v1 等一律不进 base_url）。端点是纯连接点、对所有模型可互换；接口路径（含版本段）完整放进 chat_path / 各步骤 path",
-  "protocol": "openai_compat | dashscope_sync | dashscope_async | custom（能走 /chat/completions 的填 openai_compat；DashScope 原生接口按交互形态二选一，见下方协议判定规则）",
+  "protocol": "openai_compat | dashscope_sync | dashscope_async | custom（能走 chat/completions 语义的填 openai_compat；DashScope 原生接口按交互形态二选一，见下方协议判定规则）",
-  "endpoints": [{"base_url": "...", "region": "domestic|international", "timeout": 60}],
+  "endpoints": [{"base_url": "主机根，如 https://dashscope.aliyuncs.com", "region": "domestic|international", "timeout": 60}],
-  "chat_path": "对话接口路径（如 /chat/completions）",
+  "chat_path": "对话接口从主机根起的完整路径（含版本段，如 /compatible-mode/v1/chat/completions 或 /api/v3/chat/completions）",
   "auth_header": "认证头格式说明（如 Bearer API_KEY）",
   "request_headers": {"说明": "文档调用示例里除认证外的必需请求头，逐字照抄（如 DashScope 异步的 X-DashScope-Async: enable）；无则 null"},
   "request_fields": ["请求体字段名列表"],
@@ -896,12 +896,12 @@ async def _h_apply_llm_config(sor, params, ctx):
 
     from appPublic.uniqueID import getID
 
-    # 1. 供应商：按名称复用。端点带协议标签（2026-09-05 404 实测教训）：
+    # 1. 供应商：按名称复用。端点 = 纯连接点（2026-09-12 用户定夺：模型和端点
-    #    同一供应商可同时有 openai_compat(compatible-mode/v1) 与原生异步(api/v1)
+    #    不能强绑定——多端点=冗余/区域/多key，每个端点都必须支持全部模型）。
-    #    两类端点，账号绑定/运行时选择必须按协议匹配，否则原生模型被拼到
+    #    base_url 一律归一为主机根（scheme+域名）：提取 LLM 若给了带路径段的
-    #    compatible-mode 前缀下 → 404。另拦截「裸域退化」：提取 LLM 偶发把
+    #    URL（旧约定残留），确定性剥掉路径段；路径段归 profile.path 承载
-    #    base_url 提成 https://host（丢了 /api/v1），与已有带路径端点同 host
+    #    （完整相对路径，含版本段）。同 (主机根, region) 去重，timeout 取最大值
-    #    时该裸域无意义，跳过不入目录。
+    #    （同主机不同形态接口的超时需求不同，单端点必须覆盖最长）。
     from urllib.parse import urlparse as _urlparse
     recs = await sor.sqlExe(
         "SELECT id, endpoints FROM llm_vendor WHERE name=${n}$", {"n": vendor_name})
@@ -913,53 +913,81 @@ async def _h_apply_llm_config(sor, params, ctx):
             eps_old = json.loads(getattr(recs[0], "endpoints", "") or "[]")
         except Exception:
             eps_old = []
-        merged = list(eps_old)
+        # 存量端点也归一到主机根（旧数据带 /api/v1 等路径段时收敛合并）
+        merged = []
+        for e in eps_old:
+            pu = _urlparse((e.get("base_url") or "").rstrip("/"))
+            root = (pu.scheme + "://" + pu.netloc) if pu.netloc else (e.get("base_url") or "").rstrip("/")
+            if not root:
+                continue
+            hit = None
+            for m2 in merged:
+                if m2["_root"] == root and m2["region"] == (e.get("region") or "domestic"):
+                    hit = m2
+                    break
+            if hit is None:
+                merged.append({"_root": root, "region": e.get("region") or "domestic",
+                               "timeout": int(_f(e.get("timeout"), 60))})
+            else:
+                hit["timeout"] = max(hit["timeout"], int(_f(e.get("timeout"), 60)))
         added_eps = []
         skipped_eps = []
         for ep in endpoints:
             bu = (ep.get("base_url") or "").rstrip("/")
             if not bu:
                 continue
-            # 裸域退化拦截（新增与回填打标两条路径都要拦，2026-09-05 e2e 实测：
-            # 提取 LLM 偶发把 base_url 提成裸域，回填打标会让裸域变「协议匹配」
-            # 端点被运行时选中 → 拼出 https://host/services/... → 404）
             pu = _urlparse(bu)
-            if not pu.path.strip("/"):
+            if not pu.netloc:
-                same_host = any(_urlparse((e.get("base_url") or "")).netloc == pu.netloc
+                skipped_eps.append("%s（非法 URL：无主机名）" % bu)
-                                and _urlparse((e.get("base_url") or "")).path.strip("/")
-                                for e in merged)
-                if same_host:
-                    skipped_eps.append("%s（裸域退化：同 host 已有带路径端点）" % bu)
-                    continue
-            # 同 base_url 已存在：只补协议标签（旧端点无 protocol 时回填）
-            hit = None
-            for e in merged:
-                if (e.get("base_url") or "").rstrip("/") == bu:
-                    hit = e
-                    break
-            if hit is not None:
-                if protocol and not hit.get("protocol"):
-                    hit["protocol"] = protocol
                 continue
-            merged.append({"base_url": bu, "region": ep.get("region") or "domestic",
+            root = pu.scheme + "://" + pu.netloc
-                           "timeout": int(_f(ep.get("timeout"), 60)),
+            if pu.path.strip("/"):
-                           "protocol": protocol})
+                skipped_eps.append("%s 的路径段 /%s 不入端点（归 profile.path 承载）"
-            added_eps.append(bu)
+                                   % (bu, pu.path.strip("/")))
+            region = ep.get("region") or "domestic"
+            to = int(_f(ep.get("timeout"), 60))
+            hit = None
+            for m2 in merged:
+                if m2["_root"] == root and m2["region"] == region:
+                    hit = m2
+                    break
+            if hit is None:
+                merged.append({"_root": root, "region": region, "timeout": to})
+                added_eps.append(root)
+            else:
+                hit["timeout"] = max(hit["timeout"], to)
+        merged_out = [{"base_url": m2["_root"], "region": m2["region"],
+                       "timeout": m2["timeout"], "protocol": protocol}
+                      for m2 in merged]
         await sor.sqlExe(
             "UPDATE llm_vendor SET endpoints=${e}$, updated_at=NOW() WHERE id=${i}$",
-            {"e": json.dumps(merged, ensure_ascii=False), "i": vendor_id})
+            {"e": json.dumps(merged_out, ensure_ascii=False), "i": vendor_id})
         await sor.sqlExe("COMMIT", {})
         vendor_action = "复用供应商 %s（新增端点 %d 个%s）" % (
             vendor_id, len(added_eps),
-            ("，跳过 %s" % "；".join(skipped_eps)) if skipped_eps else "")
+            ("，说明：%s" % "；".join(skipped_eps)) if skipped_eps else "")
-        vendor_endpoints = merged
+        vendor_endpoints = merged_out
     else:
         vendor_id = getID()
-        eps_norm = [{"base_url": (ep.get("base_url") or "").rstrip("/"),
+        eps_norm = []
-                     "region": ep.get("region") or "domestic",
+        for ep in endpoints:
-                     "timeout": int(_f(ep.get("timeout"), 60)),
+            bu = (ep.get("base_url") or "").rstrip("/")
-                     "protocol": protocol} for ep in endpoints
+            if not bu:
-                    if (ep.get("base_url") or "").strip()]
+                continue
+            pu = _urlparse(bu)
+            root = (pu.scheme + "://" + pu.netloc) if pu.netloc else bu
+            hit = None
+            for m2 in eps_norm:
+                if m2["base_url"] == root and m2["region"] == (ep.get("region") or "domestic"):
+                    hit = m2
+                    break
+            to = int(_f(ep.get("timeout"), 60))
+            if hit is None:
+                eps_norm.append({"base_url": root,
+                                 "region": ep.get("region") or "domestic",
+                                 "timeout": to, "protocol": protocol})
+            else:
+                hit["timeout"] = max(hit["timeout"], to)
         await sor.C("llm_vendor", {
             "id": vendor_id, "name": vendor_name,
             "endpoints": json.dumps(eps_norm, ensure_ascii=False),
@@ -1134,32 +1162,10 @@ async def _h_apply_llm_config(sor, params, ctx):
     if protocol not in RUNTIME_PROTOCOLS:
         rt_note = ("⚠️ 协议「%s」的适配模板已存档，但当前运行时调用链仅支持 %s——"
                    "该供应商模型暂不可被产线直接调用" % (protocol, "/".join(RUNTIME_PROTOCOLS)))
-    # 4. 账号端点绑定自检（2026-09-05 404 教训）：账号只绑 compatible-mode
+    # 4. 账号端点绑定自检已删（2026-09-12 端点可互换重构）：端点是纯连接点、
-    #    端点时，原生协议模型运行时会被选到错误端点 → 404。自动补绑本协议
+    #    对所有模型等价，账号 endpoint_ids 空=选用全部端点，不存在「协议不匹配
-    #    端点到该供应商全部活跃账号（不删已有绑定），如实报告。
+    #    会 404」需要补绑的场景；请求形态由 profile.path（完整相对路径）承载。
     account_notes = []
-    proto_idx = [i for i, e in enumerate(vendor_endpoints or [])
-                 if (e.get("protocol") or "") == protocol]
-    if proto_idx:
-        recs = await sor.sqlExe(
-            "SELECT id, name, endpoint_ids FROM llm_account "
-            "WHERE vendor_id=${v}$ AND status='active'", {"v": vendor_id})
-        await sor.sqlExe("COMMIT", {})
-        for acc in recs or []:
-            try:
-                bound = [int(b) for b in json.loads(getattr(acc, "endpoint_ids", "") or "[]")]
-            except Exception:
-                bound = []
-            if any(i in proto_idx for i in bound):
-                continue  # 已绑本协议端点，可用
-            newb = sorted(set(bound) | set(proto_idx))
-            await sor.sqlExe(
-                "UPDATE llm_account SET endpoint_ids=${e}$, updated_at=NOW() WHERE id=${i}$",
-                {"e": json.dumps(newb), "i": getattr(acc, "id", "")})
-            await sor.sqlExe("COMMIT", {})
-            account_notes.append("账号「%s」原绑定端点 %s 不含协议 %s 的端点（运行时会 404），"
-                                 "已自动补绑为 %s" % (getattr(acc, "name", ""), bound,
-                                                      protocol, newb))
     audit_note = await _audit_media_convention(sor, list(profile_ids.values()))
     # 5. 模型配套技能自动生成（2026-09-08 用户定夺：每个模型的加入需要有
     #    配套 skill——记录该模型需要什么样的输入；invoke_model 完备性判断由
@@ -1639,8 +1645,12 @@ def _gen_templates(protocol: str, capability: str, spec: dict) -> dict:
     """按文档示例动态生成适配模板（path/headers/request/response/param_schema）。
 
     返回 {path, headers, req, resp, param_schema, media_params, biz_params, notes}。
+    path 约定（2026-09-12 用户定夺）：从主机根起的完整相对路径（含版本段，如
+    /compatible-mode/v1/chat/completions）——端点 base_url 是纯主机根，
+    运行时 url = 端点主机根 + path。提取 LLM 给全 URL 时确定性剥掉 scheme+host。
     """
-    chat_path = (spec.get("chat_path") or "/chat/completions").strip()
+    chat_path = _norm_relative_path(
+        spec.get("chat_path") or "/chat/completions")
     headers = _headers_from_spec(spec)
     notes = []
     if protocol == "openai_compat":
@@ -1731,6 +1741,29 @@ def _gen_templates(protocol: str, capability: str, spec: dict) -> dict:
             "biz_params": biz_params, "notes": notes}
 
 
+def _norm_relative_path(p: str) -> str:
+    """接口路径归一为「主机根起的完整相对路径」（2026-09-12 端点可互换约定）。
+
+    - 全 URL → 剥掉 scheme+host，保留完整路径段（含版本段）；
+    - 相对路径 → 保证以 / 开头；
+    - 空 → 原样返回（调用方有自己的缺省）。
+    旧约定「path 相对带版本段的 base_url」已废弃：端点是纯连接点（主机根），
+    path 必须自带版本段，任意端点可拼。
+    """
+    s = (p or '').strip()
+    if not s:
+        return s
+    if s.startswith('http://') or s.startswith('https://'):
+        from urllib.parse import urlparse
+        pu = urlparse(s)
+        s = pu.path or '/'
+        if pu.query:
+            s = s + '?' + pu.query
+    if not s.startswith('/'):
+        s = '/' + s
+    return s
+
+
 # 旧骨架模板标记：含这些标记的 profile 视为不可用，apply 时按文档自愈重建
 _SKELETON_MARKERS = ('__from_doc__', '__note__', 'xxx_file',
                      # 旧单值媒体契约（2026-09-06 前）：params.image_file) 等。
@@ -2005,25 +2038,26 @@ async def _ensure_step_profile(sor, protocol, cap, vmid, step, spec):
       幂等键=名称「{protocol}-{purpose}-{path}」（去掉模型维度）——同供应商同协议
       下多个模型共享同一份查询模板；旧的 {vmid}-{purpose} 命名兼容复用，
       但若旧模板仍是骨架（__from_doc__）同样弃用重建。
-      path 归一化（对齐运行时契约，实测踩过三个坑）：
+      path 归一化（2026-09-12 端点可互换约定，见 _norm_relative_path）：
-        - 任务号占位符用**单花括号** {task_id}（运行时 str.replace 消费，
+        - 保留从主机根起的完整相对路径（含 /api/v1 版本段），不再剥前缀
-          Jinja 双花括号不会被替换）；文档给 {{task_id}} 也归一为单花括号
+        - 任务号占位符用**单花括号** {task_id}（运行时 str.replace 消费）
-        - 剥掉 base_url 已有的路径前缀（提取给 /api/v1/tasks/{task_id}，
+        - 缺占位符时按 DashScope 惯例补 /api/v1/tasks/{task_id}
-          运行时 url=base_url+path，base_url 已含 /api/v1 → 不归一会双前缀 404）
-        - 缺占位符时按 DashScope 惯例补 /tasks/{task_id}
     返回 profile id。
     """
     purpose = (step.get("purpose") or "query").strip() or "query"
     path = (step.get("path") or "").strip()
     method = (step.get("method") or ("GET" if purpose == "query" else "POST")).strip().upper()
-    # path 归一化
+    # path 归一化（2026-09-12 端点可互换约定）：
-    from urllib.parse import urlparse
+    #   - 保留从主机根起的完整相对路径（含 /api/v1 等版本段），不再剥前缀——
-    bu_path = urlparse((spec.get("base_url") or "").strip()).path.rstrip("/")
+    #     端点 base_url 已是纯主机根，运行时 url = 主机根 + path，剥了反而 404
-    if bu_path and path.startswith(bu_path + "/"):
+    #   - 全 URL → 剥掉 scheme+host（_norm_relative_path）
-        path = path[len(bu_path):]
+    #   - 任务号占位符用**单花括号** {task_id}（运行时 str.replace 消费，
+    #     Jinja 双花括号不会被替换）；文档给 {{task_id}} 也归一为单花括号
+    #   - 缺占位符时按 DashScope 惯例补 /api/v1/tasks/{task_id}
+    path = _norm_relative_path(path)
     path = path.replace("{{task_id}}", "{task_id}")
     if "{task_id}" not in path and purpose == "query":
-        path = (path.rstrip("/") + "/{task_id}") if path else "/tasks/{task_id}"
+        path = (path.rstrip("/") + "/{task_id}") if path and path != "/" else "/api/v1/tasks/{task_id}"
     shared_name = "%s-%s-%s" % (protocol, purpose, path or "default")
     legacy_name = "%s-%s" % (vmid, purpose)
     for cand in (shared_name, legacy_name):