From 706ccdba90665f099150c7772c38d73e8e395254 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 12 Sep 2026 12:35:27 +0800 Subject: [PATCH] =?UTF-8?q?refactor(auto-config):=20=E7=AB=AF=E7=82=B9/?= =?UTF-8?q?=E8=B7=AF=E5=BE=84=E8=81=8C=E8=B4=A3=E5=8F=8D=E8=BD=AC=E5=AF=B9?= =?UTF-8?q?=E9=BD=90=E7=AB=AF=E7=82=B9=E5=8F=AF=E4=BA=92=E6=8D=A2=E7=BA=A6?= =?UTF-8?q?=E5=AE=9A=E2=80=94=E2=80=94=E6=8F=90=E5=8F=96=E6=8F=90=E7=A4=BA?= =?UTF-8?q?=E8=AF=8Dbase=5Furl=E6=94=B9=E4=B8=BB=E6=9C=BA=E6=A0=B9(?= =?UTF-8?q?=E7=A6=81=E8=B7=AF=E5=BE=84=E6=AE=B5)+chat=5Fpath=E6=94=B9?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E7=9B=B8=E5=AF=B9=E8=B7=AF=E5=BE=84(?= =?UTF-8?q?=E5=90=AB=E7=89=88=E6=9C=AC=E6=AE=B5);apply=E7=AB=AF=E7=82=B9?= =?UTF-8?q?=E5=BD=92=E4=B8=80=E4=B8=BB=E6=9C=BA=E6=A0=B9+=E5=90=8C(?= =?UTF-8?q?=E4=B8=BB=E6=9C=BA=E6=A0=B9,region)=E5=8E=BB=E9=87=8D+timeout?= =?UTF-8?q?=E5=8F=96max+=E5=88=A0=E8=A3=B8=E5=9F=9F=E6=8B=A6=E6=88=AA?= =?UTF-8?q?=E4=B8=8E=E5=8D=8F=E8=AE=AE=E8=A1=A5=E7=BB=91=E8=87=AA=E6=A3=80?= =?UTF-8?q?;=5Fgen=5Ftemplates/=5Fensure=5Fstep=5Fprofile=20path=E7=BB=8F?= =?UTF-8?q?=5Fnorm=5Frelative=5Fpath=E5=BD=92=E4=B8=80(=E5=85=A8URL?= =?UTF-8?q?=E5=89=A5scheme+host,=E4=BF=9D=E7=95=99=E7=89=88=E6=9C=AC?= =?UTF-8?q?=E6=AE=B5,=E4=B8=8D=E5=86=8D=E5=89=A5base=5Furl=E5=89=8D?= =?UTF-8?q?=E7=BC=80)=E3=80=82=E9=85=8D=E5=A5=97pipeline-llm=203ea03be?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_platform/platform_ability.py | 194 +++++++++++++++----------- 1 file changed, 114 insertions(+), 80 deletions(-) diff --git a/pipeline_platform/platform_ability.py b/pipeline_platform/platform_ability.py index bd024de..ab70936 100644 --- a/pipeline_platform/platform_ability.py +++ b/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", - "protocol": "openai_compat | dashscope_sync | dashscope_async | custom（能走 /chat/completions 的填 openai_compat；DashScope 原生接口按交互形态二选一，见下方协议判定规则）", - "endpoints": [{"base_url": "...", "region": "domestic|international", "timeout": 60}], - "chat_path": "对话接口路径（如 /chat/completions）", + "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 原生接口按交互形态二选一，见下方协议判定规则）", + "endpoints": [{"base_url": "主机根，如 https://dashscope.aliyuncs.com", "region": "domestic|international", "timeout": 60}], + "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 实测教训）： - # 同一供应商可同时有 openai_compat(compatible-mode/v1) 与原生异步(api/v1) - # 两类端点，账号绑定/运行时选择必须按协议匹配，否则原生模型被拼到 - # compatible-mode 前缀下 → 404。另拦截「裸域退化」：提取 LLM 偶发把 - # base_url 提成 https://host（丢了 /api/v1），与已有带路径端点同 host - # 时该裸域无意义，跳过不入目录。 + # 1. 供应商：按名称复用。端点 = 纯连接点（2026-09-12 用户定夺：模型和端点 + # 不能强绑定——多端点=冗余/区域/多key，每个端点都必须支持全部模型）。 + # base_url 一律归一为主机根（scheme+域名）：提取 LLM 若给了带路径段的 + # URL（旧约定残留），确定性剥掉路径段；路径段归 profile.path 承载 + # （完整相对路径，含版本段）。同 (主机根, 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("/"): - same_host = any(_urlparse((e.get("base_url") or "")).netloc == pu.netloc - 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 + if not pu.netloc: + skipped_eps.append("%s（非法 URL：无主机名）" % bu) continue - merged.append({"base_url": bu, "region": ep.get("region") or "domestic", - "timeout": int(_f(ep.get("timeout"), 60)), - "protocol": protocol}) - added_eps.append(bu) + root = pu.scheme + "://" + pu.netloc + if pu.path.strip("/"): + skipped_eps.append("%s 的路径段 /%s 不入端点（归 profile.path 承载）" + % (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 "") - vendor_endpoints = merged + ("，说明：%s" % "；".join(skipped_eps)) if skipped_eps else "") + vendor_endpoints = merged_out else: vendor_id = getID() - eps_norm = [{"base_url": (ep.get("base_url") or "").rstrip("/"), - "region": ep.get("region") or "domestic", - "timeout": int(_f(ep.get("timeout"), 60)), - "protocol": protocol} for ep in endpoints - if (ep.get("base_url") or "").strip()] + eps_norm = [] + for ep in endpoints: + bu = (ep.get("base_url") or "").rstrip("/") + if not bu: + 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 - # 端点时，原生协议模型运行时会被选到错误端点 → 404。自动补绑本协议 - # 端点到该供应商全部活跃账号（不删已有绑定），如实报告。 + # 4. 账号端点绑定自检已删（2026-09-12 端点可互换重构）：端点是纯连接点、 + # 对所有模型等价，账号 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 归一化（对齐运行时契约，实测踩过三个坑）： - - 任务号占位符用**单花括号** {task_id}（运行时 str.replace 消费， - Jinja 双花括号不会被替换）；文档给 {{task_id}} 也归一为单花括号 - - 剥掉 base_url 已有的路径前缀（提取给 /api/v1/tasks/{task_id}， - 运行时 url=base_url+path，base_url 已含 /api/v1 → 不归一会双前缀 404） - - 缺占位符时按 DashScope 惯例补 /tasks/{task_id} + path 归一化（2026-09-12 端点可互换约定，见 _norm_relative_path）： + - 保留从主机根起的完整相对路径（含 /api/v1 版本段），不再剥前缀 + - 任务号占位符用**单花括号** {task_id}（运行时 str.replace 消费） + - 缺占位符时按 DashScope 惯例补 /api/v1/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 归一化 - from urllib.parse import urlparse - bu_path = urlparse((spec.get("base_url") or "").strip()).path.rstrip("/") - if bu_path and path.startswith(bu_path + "/"): - path = path[len(bu_path):] + # path 归一化（2026-09-12 端点可互换约定）： + # - 保留从主机根起的完整相对路径（含 /api/v1 等版本段），不再剥前缀—— + # 端点 base_url 已是纯主机根，运行时 url = 主机根 + path，剥了反而 404 + # - 全 URL → 剥掉 scheme+host（_norm_relative_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):