""" pipeline-service: platform_ability — 平台内部 agent 能力包（pipeline_id=platform_general） 产线平台自身的运维/管理 agent：拥有平台应用和各模块的技能， 核心能力是「通读模型 API 文档 → 自动生成模型治理配置（供应商/适配模板/模型/定价）」。 权限模型（代码层硬门禁，不依赖 prompt）： 所有工具 handler 入口校验调用者持有 owner 组织的角色（orgtypeid='owner'，含通配 'owner.*'）。 内部 agent 只服务 owner 组织的角色——其他组织的用户即使打开页面也调不动任何工具。 配置生成链路（用户给定 API 文档 → 完成配置）： ① fetch_model_doc：抓取文档页面（SSRF 防护：仅 http(s) 公网域名） ② extract_llm_api_spec：LLM 从文档提取端点/协议/请求响应格式/定价 ③ apply_llm_config：写库——复用 pipeline-llm 模块的 CRUD 函数 （llm_vendor.endpoints + llm_api_profile 模板 + llm_model 含四价） """ import asyncio import hashlib import json import logging import re import time import yaml from appPublic.uniqueID import getID from pipeline_core import ( ToolDefinition, PipelineAbility, register_ability, ) logger = logging.getLogger("pipeline.platform_ability") PLATFORM_PIPELINE_ID = "platform_general" # 组织门禁：内部 agent 仅 owner 组织的角色可操作（任一 owner.* 角色，含通配） OWNER_ORG = "owner" # 当前运行时调用链支持的协议（模式定义在模型的适配模板/执行器，非供应商）： # openai_compat —— 同步 chat/completions 一次往返 # dashscope_async —— 异步执行器：提交任务→轮询 query_profile_ids→取结果（2026-09-05） # 其他协议的适配模板会存入 llm_api_profile 备查，但运行时暂不渲染。 RUNTIME_PROTOCOLS = ("openai_compat", "dashscope_async", "dashscope_sync") # ────────────────────── 会话级规格缓存（extract 锚定，2026-09-05）────────────────────── # 根因：apply_llm_config 要求 LLM 把 extract 返回的大 JSON 规格逐字复制进参数， # 模型「转述」大 JSON 必然编造结构（实测三轮全编造 vendor{}/api_profile{} 等契约外字段）。 # 对策：extract 成功后把规格锚定到「会话级缓存」，apply/apply_model_pricing 用 # use_last_extract=true 直接取用 + 少量覆盖项（vendor_name 等），LLM 永不搬运大 JSON。 # # 会话级隔离（用户硬要求）：缓存键 = pipeline_id:user_id:session_id，三段缺一不可， # 防跨会话/跨用户/跨产线串数据。存 Redis db4（治理命名空间，gateway 已在用）， # 跨 worker 进程不丢；TTL 兜底过期。Redis 不可用时降级进程内字典（仍按同键隔离）。 _SPEC_CACHE_TTL = 7200 # 规格缓存 2 小时（一次配置会话足够） _SPEC_CACHE_PREFIX = "llm_spec" _spec_cache_local = {} # Redis 不可用时的进程内降级（同样按隔离键） def _spec_cache_key(ctx): """会话级隔离键：pipeline_id:user_id:session_id。任一段缺失返回 ''（拒绝缓存）。""" pl = (ctx.get("pipeline_id") or "").strip() uid = (ctx.get("user_id") or "").strip() sid = (ctx.get("session_id") or "").strip() if not (pl and uid and sid): return '' return "%s:%s:%s:%s" % (_SPEC_CACHE_PREFIX, pl, uid, sid) def _spec_redis(): """复用治理命名空间 Redis db4（与 gateway 限流同库，会话态用 db3 不冲突）。""" try: import redis as _r try: from appPublic.jsonConfig import getConfig url = (getConfig().website or {}).get('session_redis', {}).get('url', '') except Exception: url = '' if not url: url = 'redis://127.0.0.1:6379/3' base = url.rsplit('/', 1)[0] return _r.Redis.from_url(base + '/4', socket_timeout=2) except Exception: return None def _spec_save(ctx, spec): """锚定规格到会话缓存。返回 True/False。无隔离键则不缓存（apply 会要求显式传 spec）。""" key = _spec_cache_key(ctx) if not key: return False try: payload = json.dumps(spec, ensure_ascii=False) except Exception: return False r = _spec_redis() if r is not None: try: r.set(key, payload, ex=_SPEC_CACHE_TTL) return True except Exception: pass _spec_cache_local[key] = (payload, time.time() + _SPEC_CACHE_TTL) return True def _spec_load(ctx): """取本会话锚定的规格。无则返回 None。严格会话级隔离（键含 session_id）。""" key = _spec_cache_key(ctx) if not key: return None r = _spec_redis() raw = None if r is not None: try: v = r.get(key) raw = v.decode('utf-8') if isinstance(v, (bytes, bytearray)) else v except Exception: raw = None if raw is None: ent = _spec_cache_local.get(key) if ent and ent[1] > time.time(): raw = ent[0] elif ent: _spec_cache_local.pop(key, None) if not raw: return None try: return json.loads(raw) except Exception: return None def _spec_overlay(base, overrides): """覆盖项合并：仅允许少量标量覆盖（vendor_name 等），不做大 JSON 搬运。 overrides 只认白名单键，防止 LLM 借覆盖项塞回编造结构。 model_capability（2026-09-06 wan3.0 教训）：{vendor_model_id: capability} 按模型纠正能力——用户提醒「这是参考生视频」后助手必须有工具路径把纠正 落进规格，否则只能重跑 extract，同样提示词同样误判（提醒无效的死循环）。 纠正值仍要过能力字典校验（下游 unknown 拒绝落库），不会被塞私货。 """ allowed = {'vendor_name', 'base_url', 'protocol', 'doc_url', 'doc_notes'} out = dict(base or {}) for k in allowed: if k in (overrides or {}) and str(overrides.get(k) or '').strip(): out[k] = str(overrides[k]).strip() mc = (overrides or {}).get('model_capability') if isinstance(mc, str): try: mc = json.loads(mc) except Exception: mc = None if isinstance(mc, dict) and mc: fixed = [] models = list(out.get('models') or []) for m in models: vmid = m.get('vendor_model_id') or m.get('name') or '' newcap = str(mc.get(vmid) or '').strip() if vmid and newcap and newcap != m.get('capability'): fixed.append('%s: %s→%s' % (vmid, m.get('capability'), newcap)) m['capability'] = newcap # 人工纠正登记（2026-09-08）：apply 阶段的能力守卫必须跳过 # 这些模型——人工指令是权威，守卫再纠正=把用户的纠正改回去 # （qwen-image-plus 实测死循环的机制根源）。 hc = dict(out.get('__human_cap_overrides__') or {}) hc[vmid] = newcap out['__human_cap_overrides__'] = hc out['models'] = models if fixed: out['doc_notes'] = ((out.get('doc_notes') or '') + ' [能力已按人工纠正: %s]' % '；'.join(fixed)).strip() return out # ────────────────────── 权限门禁（代码层） ────────────────────── async def _require_owner(sor, ctx) -> str: """校验当前用户持有 owner 组织的角色。返回 ''=通过，否则错误信息。 判定：userrole→role JOIN 后，存在任一 orgtypeid='owner' 的角色（含通配 'owner.*'）。 """ uid = ctx.get("user_id", "") or "" if not uid: return "无法识别当前用户身份（未登录），内部 agent 仅 owner 组织角色可用" recs = await sor.sqlExe( "SELECT r.orgtypeid, r.name FROM userrole ur JOIN role r ON ur.roleid=r.id " "WHERE ur.userid=${u}$", {"u": uid}) await sor.sqlExe("COMMIT", {}) roles = [] for r in (recs or []): o = getattr(r, "orgtypeid", "") or "" n = getattr(r, "name", "") or "" if o and n: roles.append("%s.%s" % (o, n)) if any(fn.split(".", 1)[0] == OWNER_ORG for fn in roles): return "" return "权限不足：内部 agent 工具仅 %s 组织的角色可操作，当前角色 %s" % ( OWNER_ORG, sorted(set(roles)) or "无") def _row(r): """sqlor row → dict（容忍属性访问）。""" if hasattr(r, "__dict__"): return {k: v for k, v in r.__dict__.items() if not k.startswith("_")} return {} # ────────────────────── 工具 1：模型治理状态 ────────────────────── async def _h_platform_llm_status(sor, params, ctx): """供应商/账号/模型/用量概览（owner 组织诊断配置用）。""" err = await _require_owner(sor, ctx) if err: return err out = {} for tbl in ("llm_vendor", "llm_account", "llm_model"): recs = await sor.sqlExe( "SELECT status, COUNT(*) AS c FROM " + tbl + " GROUP BY status", {}) await sor.sqlExe("COMMIT", {}) out[tbl] = {getattr(r, "status", ""): int(getattr(r, "c", 0)) for r in (recs or [])} accs = await sor.sqlExe( "SELECT name, balance, status FROM llm_account ORDER BY balance DESC LIMIT 20", {}) await sor.sqlExe("COMMIT", {}) out["accounts"] = [ {"name": getattr(r, "name", ""), "balance": float(getattr(r, "balance", 0) or 0), "status": getattr(r, "status", "")} for r in (accs or [])] models = await sor.sqlExe( "SELECT name, vendor_model_id, capability, status " "FROM llm_model ORDER BY created_at DESC LIMIT 30", {}) await sor.sqlExe("COMMIT", {}) out["models"] = [ {"name": getattr(r, "name", ""), "vendor_model_id": getattr(r, "vendor_model_id", ""), "capability": getattr(r, "capability", ""), "status": getattr(r, "status", "")} for r in (models or [])] return json.dumps(out, ensure_ascii=False) # ────────────────────── 工具 2：抓取模型 API 文档 ────────────────────── _MAX_DOC_CHARS = 60000 _MAX_REDIRECTS = 5 def _validate_doc_url(url: str) -> str: """SSRF 防护：仅允许 http(s) + 公网域名。返回 ''=通过，否则错误。""" if not re.match(r"^https?://", url or ""): return "文档 URL 必须是 http/https 链接" host = re.sub(r"^https?://", "", url).split("/")[0].split(":")[0].lower() if not re.match(r"^[a-z0-9][a-z0-9.-]+\.[a-z]{2,}$", host): return "文档 URL 域名非法（不接受 IP/内网地址）" if host in ("localhost",) or host.endswith((".local", ".internal", ".localhost")): return "文档 URL 不允许指向内网/本地地址" return "" def _html_to_text(html: str) -> str: """粗暴去标签提取文本（文档页面用，不追求完美排版）。""" txt = re.sub(r"(?is)<(script|style|noscript)[^>]*>.*?", " ", html or "") txt = re.sub(r"(?is)<[^>]+>", " ", txt) txt = re.sub(r" ", " ", txt) txt = re.sub(r"<", "<", txt) txt = re.sub(r">", ">", txt) txt = re.sub(r"&", "&", txt) txt = re.sub(r""", '"', txt) txt = re.sub(r"[ \t]+", " ", txt) txt = re.sub(r"\n\s*\n+", "\n", txt) return txt.strip() def _is_private_ip(ip: str) -> bool: """DNS 解析后的二次防线：解析到私有/环回/链路本地地址一律拒绝（DNS rebinding 防护）。""" import ipaddress try: a = ipaddress.ip_address(ip) except ValueError: return True # 解析不出就当私有（拒绝） return (a.is_private or a.is_loopback or a.is_link_local or a.is_reserved or a.is_multicast or a.is_unspecified) async def _fetch_url_safe(url: str, max_redirects: int = _MAX_REDIRECTS): """带重定向逐跳校验的抓取（每一跳都过域名+DNS双重校验）。 返回 (text, content_type)；失败抛 ValueError。 """ import aiohttp current = url for _ in range(max_redirects + 1): verr = _validate_doc_url(current) if verr: raise ValueError(verr) host = re.sub(r"^https?://", "", current).split("/")[0].split(":")[0].lower() # DNS 解析校验（同步阻塞短调用，可接受；防 DNS rebinding 指向内网） try: infos = await asyncio.get_event_loop().getaddrinfo(host, None) except Exception as e: raise ValueError("文档域名无法解析：%s" % str(e)[:120]) for info in infos: ip = str(info[4][0]) if _is_private_ip(ip): raise ValueError("文档域名解析到内网地址（%s），拒绝访问" % ip) timeout = aiohttp.ClientTimeout(total=30) async with aiohttp.ClientSession(timeout=timeout) as sess: async with sess.get(current, headers={"User-Agent": "Mozilla/5.0"}, allow_redirects=False, ssl=False) as resp: if resp.status in (301, 302, 303, 307, 308): loc = resp.headers.get("Location", "") if not loc: raise ValueError("重定向缺少 Location") if loc.startswith("/"): scheme = "https" if current.startswith("https") else "http" loc = "%s://%s%s" % (scheme, host, loc) current = loc continue if resp.status != 200: raise ValueError("抓取失败：HTTP %d" % resp.status) ctype = resp.headers.get("Content-Type", "") raw = await resp.text(errors="replace") return raw, ctype raise ValueError("重定向次数超限（>%d）" % max_redirects) async def _h_fetch_model_doc(sor, params, ctx): """抓取模型 API 文档页面 → 纯文本（供 LLM 提取配置规格）。""" err = await _require_owner(sor, ctx) if err: return err url = (params.get("url") or "").strip() try: raw, ctype = await _fetch_url_safe(url) except ValueError as e: return str(e) except Exception as e: return "抓取失败：%s" % str(e)[:200] if "html" in ctype.lower() or raw.lstrip()[:15].lower().startswith((" _MAX_DOC_CHARS: text = text[:_MAX_DOC_CHARS] + "\n[文档过长已截断，共 %d 字符]" % len(text) # 出处行：提取规格照抄进 doc_url，落库写进模型/定价描述（2026-09-05 用户规则） text = text + "\n\n[出处URL] " + url return text # ────────────────────── 工具 3：提取配置规格（LLM） ────────────────────── _EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型 API 文档，提取配置规格。 只输出一个 JSON 对象（不要 markdown 代码块），字段： { "vendor_name": "供应商名称", "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": ["请求体字段名列表"], "request_example": {"说明": "文档调用示例(cURL/代码)中的完整请求体 JSON，逐字照抄结构与各字段示例值，不要修改；文档无示例填 null"}, "field_enums": {"说明": "业务参数的可选值枚举（从文档参数说明表逐字照抄，如 {\"resolution\": [\"480P\",\"720P\",\"1080P\"], \"duration\": [3,5,10]}）。文档没列可选值的参数不要编造；无则 null"}, "response_format": "响应格式说明（content 字段路径 + usage 字段路径）", "response_example": {"说明": "文档中的响应示例 JSON（或结果字段说明），逐字照抄；无则 null"}, "async_steps": [{"purpose": "query|download", "path": "该步骤接口路径", "method": "GET|POST", "request_fields": ["请求字段名列表"], "response_format": "该步骤响应格式说明"}], "models": [{"vendor_model_id": "供应商侧模型ID", "capability": "t2t|t2i|i2t|m2t|t2v|i2v|r2v|i2i|embedding|rerank|tts|asr|mm-embedding|mm-rerank", "sync_mode": "sync|async（提交后需轮询查询结果的填 async）", "description": "一句话说明"}], "pricing": { "currency": "CNY|USD（文档标注的币种）", "items": [{"vendor_model_id": "对应模型ID", "factor": "duration|flat|prompt_tokens|completion_tokens|uncache_tokens|cached_tokens（计价因子：视频按秒=duration，按次=flat；token计价：文档只给一个输入价=prompt_tokens；文档区分缓存命中价=非缓存输入用uncache_tokens+缓存命中用cached_tokens+输出completion_tokens——禁止用prompt_tokens配缓存价，会与uncache/cached重复计价）", "unit_price": 0.45, "unit": "秒|次|百万（文档标注的计价单位名）", "dimensions": {"resolution": "480P（影响价格档位的维度，按文档原文值；禁止放 model）"}, "doc_quote": "文档原文定价句（逐字照抄，标明出处用）"}] }, "doc_url": "从文档文本末尾的 [出处URL] 行逐字照抄（配置出处，必填）", "doc_notes": "文档中影响配置的关键注意点" } 协议判定规则（2026-09-08 qwen-image-plus 实测教训：交互形态有三种，同步生成此前被误判成异步）： - **dashscope_sync（同步生成）**：调用示例是 `MultiModalConversation.call(stream=False)` 或一次 HTTP 请求直接返回产物（图像/语音），无 task_id、无轮询查询接口 → sync_mode=sync、async_steps=[]、**不带 X-DashScope-Async 头**。 - **dashscope_async（异步任务）**：文档描述「先提交任务拿 task_id、再轮询查询结果」，或示例带 `X-DashScope-Async: enable` 头 → sync_mode=async，并提取 async_steps： 至少一条 purpose=query（查询任务状态/结果）；若文档另有独立下载/取文件接口，再加一条 purpose=download。 - 判据优先级：**调用示例的交互形态 > 供应商惯例**。同一供应商（如 DashScope）可同时有同步生成模型（图像/语音）和异步任务模型（长视频）——按各模型自己的文档判，不要按供应商一刀切。 - 任务查询接口通常是**供应商级共用**的（如 DashScope 全系异步模型都是 GET /tasks/{task_id}），path 照文档逐字抄；同供应商多个模型会复用同一份查询模板，不要因模型而异。 - 同步模型 async_steps 填空数组 []。 能力判定铁律（2026-09-06 wan3.0-video-prime 误判教训——概述明写"四模态全能参考"、 输入=音频/图像/文本/视频，却因调用示例只有 prompt 被判成 t2v）： - capability 必须以文档**概述正文 + 输入/输出模态声明**为准，禁止以调用示例为准—— 示例常只演示最简形态（纯文本 prompt），照示例判会把多媒体参考模型误判成纯文本模型。 - 判定表（输入模态 → 输出模态 ⇒ 能力）： 多种媒体输入或正文含「参考/reference/全能参考/多图/多模态输入」且输出为视频 ⇒ r2v（即使示例只有文本）； 图像+文本 → 视频 ⇒ i2v；仅文本 → 视频 ⇒ t2v；仅文本 → 图像 ⇒ t2i；图像 → 图像 ⇒ i2i； 视频 → 视频 ⇒ v2v 类按字典就近；文本 → 音频 ⇒ tts；音频 → 文本 ⇒ asr； 仅图像(+文本) → 文本 ⇒ i2t；多种媒体输入(图+视频/音频等任意组合)(+文本) → 文本 ⇒ m2t；仅文本 → 文本 ⇒ t2t。 - description 里必须保留正文的模态声明原文关键词（如"支持四模态全能参考"），供下游一致性校验。 定价规则： - pricing.items 每条必须带 doc_quote（文档原文定价句逐字照抄）——定价只允许来自文档原文 - 价格/单位按文档原文记录，禁止自行换算、推导或统一单位 - 同一模型多档价格（如不同分辨率）拆成多条 item，各带自己的 dimensions 与 doc_quote - dimensions 禁止放 model——一个定价方案只服务一个模型；定价完全相同的多个模型 共享同一个定价方案（同一 ppid），不是往定价里加 model 过滤 - 文档没写价格：items 填空数组 []（禁止编造价格） 文档内容： """ async def _h_extract_llm_api_spec(sor, params, ctx): """LLM 通读文档文本 → 结构化配置规格（JSON）。""" err = await _require_owner(sor, ctx) if err: return err doc_text = (params.get("doc_text") or "").strip() if len(doc_text) < 100: return "文档文本太短（<100字符），无法提取配置——先用 fetch_model_doc 抓取" try: from pipeline_service.llm_bridge import llm_call_msgs # purpose='utility'：辅助任务走辅助模型链（机构策略配置则优先，便宜快） # timeout=300：长文档提取慢，端点默认 60 秒实测跑不完（2026-09-04 根因） raw = await llm_call_msgs( [{"role": "system", "content": _EXTRACT_PROMPT}, {"role": "user", "content": doc_text[:48000]}], temperature=0.1, org_id="0", user_id=ctx.get("user_id", ""), purpose="utility", timeout=300) except Exception as e: return "LLM 提取失败：%s" % str(e)[:200] txt = (raw or "").strip() if txt.startswith("```"): txt = txt.split("\n", 1)[1] if "\n" in txt else txt txt = txt.rsplit("```", 1)[0] try: spec = json.loads(txt) except Exception: m = re.search(r"\{.*\}", txt, re.S) if not m: return "LLM 输出不是合法 JSON：%s" % txt[:300] try: spec = json.loads(m.group(0)) except Exception: return "LLM 输出不是合法 JSON：%s" % txt[:300] # 会话级锚定（2026-09-05）：规格存会话缓存（键含 session_id，跨会话隔离）， # 后续 apply_llm_config / apply_model_pricing 用 use_last_extract=true 取用—— # LLM 不再搬运/转述大 JSON（转述必编造结构，实测三轮全编造）。 anchored = _spec_save(ctx, spec) out = dict(spec) if anchored: out["__anchored__"] = ("规格已锚定到本会话缓存。下一步调用 apply_llm_config 与 " "apply_model_pricing 时传 {\"use_last_extract\": true}，" "需改动只传覆盖项（overrides 白名单：vendor_name/base_url/" "protocol/doc_url/doc_notes/model_capability）。" "禁止把本规格复制进 spec 参数——复制即编造。") # 能力一致性告警（2026-09-06 wan3.0 教训；2026-09-08 升级：词匹配只当触发器， # 裁决交 LLM——qwen-image-plus 实测词匹配把输出词当输入证据、t2i 被反复改成 i2i） cap_alerts = [] _cap_changed = False for _m in (spec.get("models") or []): _fix, _warn = _capability_guard(_m.get("capability", ""), _m.get("description", "")) if _warn or _fix: # 词匹配报可疑 → LLM 语义裁决（裁决失败则只告警不纠正） _jfix, _jnote = await _llm_capability_judge( _m.get("capability", ""), _m.get("description", ""), _m.get("vendor_model_id", "")) if _jfix: _m["capability"] = _jfix _cap_changed = True cap_alerts.append("%s: %s（已按 LLM 裁决更新提取规格）" % (_m.get("vendor_model_id", "?"), _jnote)) elif _jnote: cap_alerts.append("%s: %s" % (_m.get("vendor_model_id", "?"), _jnote)) elif _warn: cap_alerts.append("%s: %s（词匹配可疑，LLM 裁决维持原判）" % (_m.get("vendor_model_id", "?"), _warn)) # 裁决改了能力 → 必须重新锚定：_spec_save 在裁决前已存旧值，不重存则 # apply(use_last_extract=true) 取回的还是旧能力，LLM 裁决白做（实测根因）。 if _cap_changed: _spec_save(ctx, spec) if cap_alerts: out["__capability_alert__"] = ("能力分类存疑项（LLM 语义裁决结果）：" + "；".join(cap_alerts) + "。如需人工指定用 overrides.model_capability。") return json.dumps(out, ensure_ascii=False) # ────────────────────── 工具 4：写入配置（幂等） ────────────────────── def _f(v, default=0.0): try: return float(v) except (TypeError, ValueError): return default # 描述里的媒体输入声明词（能力一致性守卫用） _DESC_MEDIA_WORDS = { 'image': ('图像', '图片', '多图'), 'video': ('视频',), 'audio': ('音频', '声音'), } _DESC_REF_WORDS = ('参考生', '全能参考', '参考图', '参考视频', '参考音频', '四模态', '多模态参考') async def _llm_capability_judge(capability, description, vendor_model_id=''): """LLM 语义裁决能力分类（2026-09-08 用户定夺：语义判断必须 LLM 参与， 词匹配只能当廉价触发器，不能当裁决者）。 背景：缺陷2（qwen-image-plus）实测——关键词守卫把描述里「图像生成」的 「图像」（输出词）当输入证据，反复把 t2i 纠正成 i2i，人工 override 也被 改回去，死循环。词匹配分不清输入/输出语义，LLM 能。 分工：_capability_guard（词匹配）只报「可疑」；本函数用 utility 廉价模型 按能力字典语义裁决。LLM 失败时返回 (None, 原因)——不裁决就不纠正， 保持提取 LLM 的原判定（宁可保守也不让词匹配越权）。 """ if not (description or '').strip(): return None, '' prompt = ( '你是模型能力分类裁决器。模型 %s 的文档描述：\n%s\n\n' '当前判定的能力分类：%s\n\n' '能力分类字典（按输入→输出模态判定，描述里的输出媒体词不算输入证据）：\n' 't2t=文本→文本；i2t=图像→文本；m2t=多媒体→文本；t2i=文本→图像；' 'i2i=图像→图像；t2v=文本→视频；i2v=图像→视频；r2v=参考媒体(图/视频/音频组合)→视频；' 'tts=文本→语音；asr=语音→文本；embedding=文本→向量；rerank=重排序\n\n' '请判断当前分类是否与描述的输入/输出模态一致。只返回 JSON：' '{"consistent": true/false, "capability": "正确分类(不一致时给出)", ' '"reason": "一句话依据"}' % (vendor_model_id or '(未命名)', description[:600], capability or '(空)')) try: from pipeline_service.llm_bridge import llm_call_msgs raw = await llm_call_msgs( [{"role": "user", "content": prompt}], temperature=0, org_id="0", purpose="utility", timeout=60) except Exception as e: return None, 'LLM 裁决调用失败（保持原判定不纠正）: %s' % repr(e)[:120] txt = (raw or '').strip() m = re.search(r'\{[\s\S]*\}', txt) if not m: return None, 'LLM 裁决输出非 JSON（保持原判定不纠正）: %s' % txt[:120] try: obj = json.loads(m.group(0)) except Exception: return None, 'LLM 裁决 JSON 解析失败（保持原判定不纠正）' if not isinstance(obj, dict): return None, '' if obj.get('consistent') is True: return None, '' newcap = str(obj.get('capability') or '').strip().lower() reason = str(obj.get('reason') or '')[:150] if newcap and newcap != (capability or ''): return newcap, 'LLM 语义裁决 %s→%s：%s' % (capability, newcap, reason) return None, ('LLM 裁决认为能力分类存疑但未给出替代（%s），请人工核对' % reason) if reason else None def _capability_guard(capability, description): """描述模态声明 vs capability 一致性守卫（2026-09-06 wan3.0-video-prime 实测： 描述自带"支持四模态全能参考"、capability 却是 t2v——提取 LLM 照调用示例判能力， 示例只演示最简形态就误判；提示词加铁律后仍需代码级兜底，LLM 不保证遵守）。 保守策略：只在「纯文本能力（t2v/t2i/t2t）+ 描述含媒体输入/参考声明」这种 铁证矛盾时纠正；其他不一致只警告不改（避免误伤描述措辞不规范的正确配置）。 歧义处理："视频"二字在 t2v/i2v 描述里几乎必现（输出侧），不算输入声明； 输入声明判据 = 参考词（参考生/全能参考/四模态等）或图像/音频词或输入媒体≥2类。 返回 (corrected_cap 或 None, warning 或 None)。 """ desc = description or '' if not desc or capability not in ('t2v', 't2i', 't2t'): return None, None has_ref = any(w in desc for w in _DESC_REF_WORDS) kinds = [k for k, words in _DESC_MEDIA_WORDS.items() if any(w in desc for w in words)] # 输出侧媒体词对称排除（2026-09-07 qwen-image-plus 实测根因）：纯文本生成 # 能力的描述几乎必含其输出媒体词——t2v 必含「视频」、t2i 必含「图像/图片」， # 那是输出声明不是输入声明。此前只排 video（video 在 t2v/i2v 描述必现）， # 没排 image（image 在 t2i 描述必现）→「千问系列图像生成模型」被误判有 # 图像输入 → t2i 被反复纠正成 i2i，且 overrides.model_capability 压不住 # （守卫在 apply 前置阶段每次又改回去，用户纠正陷入死循环）。 _OUTPUT_KIND = {'t2v': 'video', 't2i': 'image'} out_kind = _OUTPUT_KIND.get(capability) input_kinds = [k for k in kinds if k != out_kind] if not (has_ref or input_kinds): return None, None if capability == 't2t': return None, ("描述含媒体输入声明（%s%s）但能力是 t2t——请人工核对能力分类" % ('/'.join(kinds), '，含参考声明' if has_ref else '')) out_word = '视频' if capability == 't2v' else '图像' expected = None if has_ref: expected = 'r2v' if capability == 't2v' else 'i2i' elif len(input_kinds) >= 2: expected = 'r2v' if capability == 't2v' else 'i2i' elif 'image' in input_kinds: expected = 'i2v' if capability == 't2v' else 'i2i' elif 'audio' in input_kinds: # 音频输入语义不定（r2v 音频参考 / asr / tts 反向），不敢自动纠正 return None, ("描述含音频输入声明但能力是 %s——无法确定目标能力，" "请人工核对（可用 overrides.model_capability 纠正）" % capability) if expected and expected != capability: return expected, ("能力自动纠正 %s→%s：描述声明媒体输入（%s%s），" "纯文本能力与之矛盾（提取照示例误判，守卫按正文模态纠正）" % (capability, expected, '/'.join(kinds), '，含参考声明' if has_ref else '')) return None, None # ────────────────────── 模型配套技能自动生成（2026-09-08 用户定夺） ────────────────────── def _model_skill_path(model_name): """模型配套技能路径：skills/models/{model}/SKILL.md（全局模板目录）。 skills/models/ 不在 SkillLoader 扫描范围（global/pipelines/orgs/users）， 不进 agent 技能目录层——它是 invoke_model 完备性裁决的数据契约， 由 platform_model_tools.load_model_skill 直读。 """ import os from pipeline_core.skill_pack import get_skills_base safe = "".join(ch for ch in (model_name or "") if ch.isalnum() or ch in "-_.") if not safe: raise ValueError("模型名不能用于目录: %r" % model_name) return os.path.join(get_skills_base(), "models", safe, "SKILL.md") _CAP_MODAL_TEXT = { "t2t": ("文本", "文本"), "i2t": ("图像", "文本"), "m2t": ("多媒体", "文本"), "t2i": ("文本", "图像"), "i2i": ("图像", "图像"), "t2v": ("文本", "视频"), "i2v": ("图像", "视频"), "r2v": ("参考媒体（图像/视频/音频组合）", "视频"), "tts": ("文本", "语音"), "asr": ("语音", "文本"), "embedding": ("文本", "向量"), "rerank": ("候选集", "排序"), } async def _write_model_skill(sor, vmid, m, spec, profile_ids): """生成/更新模型配套技能（幂等：重 apply 覆盖旧版）。 内容全部来自本次落库的确定数据（不编造）： - 能力模态（输入→输出，能力字典语义） - 必备输入（param_schema 里 required=true 的字段 + 能力模态决定的媒体输入） - 可选参数（required=false，带默认值） - 调用契约（三数组媒体格式 + 产物形态） - 出处（doc_url / vendor / sync_mode） invoke_model 的完备性裁决 LLM 按本技能对照用户已给内容判断缺什么。 """ import os cap = (m.get("capability") or "t2t").strip().lower() inp, outp = _CAP_MODAL_TEXT.get(cap, ("文本", "产物")) desc = (m.get("description") or "").strip() sync_mode = "async" if str(m.get("sync_mode") or "").strip() == "async" else "sync" doc_url = (spec.get("doc_url") or "").strip() # param_schema（本次生成/复用的模板）→ 必备/可选参数表 required_rows, optional_rows = [], [] pid = profile_ids.get(cap, "") if pid: recs = await sor.sqlExe( "SELECT param_schema FROM llm_api_profile WHERE id=${i}$ LIMIT 1", {"i": pid}) await sor.sqlExe("COMMIT", {}) if recs: try: schema = json.loads(getattr(recs[0], "param_schema", "") or "[]") except Exception: schema = [] for f in schema if isinstance(schema, list) else []: if not isinstance(f, dict): continue nm = f.get("name") or "" label = f.get("label") or nm dv = f.get("defaultvalue") if f.get("required"): required_rows.append("- %s（%s）" % (nm, label)) else: optional_rows.append("- %s：%s%s" % ( nm, label, ("，默认 %s" % dv) if dv not in (None, "") else "")) # 能力模态决定的媒体输入（媒体三数组契约） media_required = [] if cap in ("i2v", "i2i"): media_required.append("- image_files：输入图片（公网 URL 或 base64，字符串或数组）——必备") elif cap == "asr": media_required.append("- audio_files：输入音频（公网 URL 或 base64）——必备") elif cap == "r2v": media_required.append("- image_files / audio_files / video_files：至少一类参考素材——必备") elif cap in ("t2v",): media_required.append("- image_files：（可选）提供参考图则接近 i2v 形态") lines = [ "---", "name: model-%s" % vmid, "description: 模型 %s（%s）的输入契约——必备输入/可选参数/媒体格式。invoke_model 完备性裁决依据。" % (vmid, cap), "capability: %s" % cap, "sync_mode: %s" % sync_mode, "---" , "", "# 模型 %s 输入契约" % vmid, "", "能力：%s（%s → %s）%s" % (cap, inp, outp, "，异步任务（提交→轮询）" if sync_mode == "async" else "，同步返回"), ] if desc: lines += ["", "描述：" + desc[:300]] lines += ["", "## 必备输入（缺任一即不可调用，须先向用户索取）"] if media_required: lines += media_required if required_rows: lines += required_rows if not media_required and not required_rows: lines.append("- task：任务描述/提示词（%s内容）" % inp) lines += ["", "## 可选参数（有默认值或缺省可接受，缺失**不算**不完备）"] lines += (optional_rows or ["- （无）"]) lines += [ "", "## 调用契约", "- 提示词经 task 传入（组包为 messages 末条文本）。", "- 媒体输入一律三数组：image_files / audio_files / video_files（值为公网 URL 或 base64 的字符串或数组）。", "- 产物：%s（返回本地持久 URL）。" % outp, ] if doc_url: lines.append("- 文档出处：" + doc_url) lines += [ "", "## 完备性裁决规则（judge_completeness 按此执行）", "- 对照「必备输入」逐项检查用户已提供的内容（含任务描述里隐含的 URL/素材）。", "- 必备项缺失 → 判不完备，追问话术说清缺什么、要什么格式。", "- 可选参数缺失不拦。判断不了放行（宁可放行不误拦）。", "", "（本技能由模型上线 apply_llm_config 自动生成，重 apply 时覆盖更新；数据源=落库的 param_schema+能力字典，人工可修订。）", ] path = _model_skill_path(vmid) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write("\n".join(lines)) logger.info("模型配套技能已写入: %s", path) # pipeline_llm 产品类别根名（首次自动建；已存在则复用，防类别树分裂） _LLM_PRODUCT_ROOT_NAME = "大模型消费" _LLM_ROOT_CAT_ID = "7S4za5hOEJqBGXb2f-NmK" # 现有根类别 id（实测），新建时兜底用它 async def _sync_model_products(sor, org_id="0"): """模型上线后自动产品映射（2026-09-08 用户需求：添加模型时如有产品模块自动映射）。 根治记账 failed「模型未映射到产品」：apply 落库模型后，调产品模块的 import_categories_and_products(resource_module='pipeline_llm') 增量导入 所有 active 模型为产品（幂等：已映射的按 resource_ref_id+org_id 跳过， 顺带自愈其他漏映射模型）。 产品模块未装（env 无 import_categories_and_products）→ 返回跳过说明， 不阻断 apply（模型注册本身已成功，产品映射是下游增值步骤）。 返回 (ok, note)。 """ from ahserver.serverenv import ServerEnv env = ServerEnv() import_fn = getattr(env, "import_categories_and_products", None) if not callable(import_fn): return True, "产品模块（product_management）未加载，跳过自动产品映射（模型已注册成功，可后续在产品管理手动导入产线模型）" # 定父类别：复用已有 pipeline_llm 根类别（防类别树分裂）；没有则新建根 parent_cat = _LLM_ROOT_CAT_ID try: recs = await sor.sqlExe( "SELECT id FROM product_category WHERE resource_module='pipeline_llm' " "AND parent_id IS NULL AND org_id=${o}$ LIMIT 1", {"o": org_id}) await sor.sqlExe("COMMIT", {}) if recs: parent_cat = getattr(recs[0], "id", "") or _LLM_ROOT_CAT_ID else: # 查名字兜底（parent_id 可能存 '0' 而非 NULL） recs2 = await sor.sqlExe( "SELECT id FROM product_category WHERE name=${n}$ AND org_id=${o}$ LIMIT 1", {"n": _LLM_PRODUCT_ROOT_NAME, "o": org_id}) await sor.sqlExe("COMMIT", {}) if recs2: parent_cat = getattr(recs2[0], "id", "") or _LLM_ROOT_CAT_ID except Exception as e: logger.warning("_sync_model_products 父类别解析失败（用默认根）: %s", e) try: result = await import_fn( resource_module="pipeline_llm", org_id=org_id, parent_category_id=parent_cat, user_id="system") except Exception as e: return False, "自动产品映射调用失败（模型已注册，请手动在产品管理导入产线模型）：%s" % repr(e)[:150] if not result or not result.get("success"): err = (result or {}).get("error", "导入无返回") if isinstance(result, dict) else "导入无返回" return False, "自动产品映射未成功（模型已注册，请手动导入）：%s" % str(err)[:150] return True, "自动产品映射完成：%s" % str((result or {}).get("message", ""))[:150] async def _h_apply_llm_config(sor, params, ctx): """按提取的规格写库：供应商（复用/新建）+ 适配模板 + 模型。 幂等：供应商按名称复用；模板按 (协议×能力×形态指纹) 复用——同形态复用/收编， 异形态各建各的（见 _tpl_fingerprint）；模型按 name/vendor_model_id 复用 （name 全局唯一键）——已存在则更新元数据；供应商归属不一致时不改挂、 如实报 vendor_conflicts。 规格来源（2026-09-05 防编造改造）： use_last_extract=true（推荐）→ 取本会话 extract 锚定的规格， overrides 白名单覆盖（vendor_name 等少量标量）； 否则用 params.spec（LLM 手传，历史路径）——结构不符时报错列出期望键。 """ err = await _require_owner(sor, ctx) if err: return err use_last = str(params.get("use_last_extract") or "").lower() in ("true", "1", "yes") overrides = params.get("overrides") or {} if isinstance(overrides, str): try: overrides = json.loads(overrides) except Exception: overrides = {} spec = None if use_last: spec = _spec_load(ctx) if spec is None: return ("本会话没有锚定的提取规格——先调 extract_llm_api_spec（成功后规格自动" "锚定本会话），再传 {\"use_last_extract\": true}。" "注意会话级隔离：其他会话提取的规格本会话不可见，需在本会话重新提取。") spec = _spec_overlay(spec, overrides) else: try: spec = json.loads(params.get("spec") or "{}") except Exception: return "spec 不是合法 JSON" # 结构校验（修B：报错给出路，不让模型盲改重试） if not isinstance(spec, dict) or "vendor_name" not in spec or "models" not in spec: got = sorted(spec.keys())[:12] if isinstance(spec, dict) else "(非对象)" return ("spec 结构不符——不要自己编写/改写规格结构。期望顶层键：" "vendor_name / base_url / endpoints / protocol / chat_path / models / " "pricing / doc_url / doc_notes（extract_llm_api_spec 的原样输出）。" "你传入的顶层键：%s。正确做法：调 extract_llm_api_spec 后传 " "{\"use_last_extract\": true}（规格已锚定本会话，禁止复制转述），" "需改动加 overrides（仅 vendor_name/base_url/protocol/doc_url/doc_notes）。" % got) vendor_name = (spec.get("vendor_name") or "").strip() if not vendor_name: return "spec 缺少 vendor_name（可用 overrides.vendor_name 指定）" endpoints = spec.get("endpoints") or [] if not endpoints and spec.get("base_url"): endpoints = [{"base_url": spec["base_url"], "region": "domestic", "timeout": 60}] if not endpoints: return "spec 缺少 endpoints/base_url" protocol = (spec.get("protocol") or "openai_compat").strip() or "openai_compat" models = spec.get("models") or [] if not models: return "spec.models 为空——文档里没有可配置的模型？" # 能力一致性守卫前置（2026-09-06 wan3.0 二次教训；2026-09-08 用户定夺升级）： # 必须在 caps 计算/模板生成之前跑——纠正能力后模板按新能力生成，否则 # profile_ids 无该能力键、模型更新的 IF 空值保护把旧模板原样保留。 # ⚠️ 语义判断必须 LLM 参与：词匹配只当廉价触发器（报可疑），裁决交 # _llm_capability_judge；**人工 override 的模型一律跳过守卫**——人工指令是 # 权威，守卫把用户纠正改回去正是 qwen-image-plus 死循环的机制根源。 cap_notes = [] human_caps = spec.get('__human_cap_overrides__') or {} for m in models: vmid = (m.get('vendor_model_id') or m.get('name') or '').strip() if vmid and vmid in human_caps: cap_notes.append("%s: 能力按人工指定 %s（守卫已跳过，不再自动纠正）" % (vmid, human_caps[vmid])) continue _fix, _warn = _capability_guard( m.get("capability") or "", m.get("description") or "") if not (_fix or _warn): continue _jfix, _jnote = await _llm_capability_judge( m.get("capability") or "", m.get("description") or "", vmid) if _jfix: m["capability"] = _jfix cap_notes.append("%s: %s" % (vmid or "?", _jnote)) elif _jnote: cap_notes.append("%s: %s" % (vmid or "?", _jnote)) else: cap_notes.append("%s: 词匹配报可疑但 LLM 裁决维持 %s（%s）" % (vmid or "?", m.get("capability") or "?", (_warn or _fix or '')[:80])) # 能力分类校验（2026-09-05 用户定夺的规则）：模型能力必须是字典已登记的 # 分类；不存在则拒绝落库——先加能力分类（appcodes_kv llm_capability， # 含种子/提示词/端点注释四处同步），再配模型。防 LLM 静默塞进近似分类。 recs = await sor.sqlExe( "SELECT k FROM appcodes_kv WHERE parentid='llm_capability'", {}) await sor.sqlExe("COMMIT", {}) known_caps = set(getattr(r, "k", "") for r in (recs or [])) unknown = sorted(set( (m.get("capability") or "t2t").strip() for m in models) - known_caps) if unknown: return ("能力分类未登记，拒绝落库：%s。请先在能力分类字典" "（appcodes_kv parentid=llm_capability，含种子数据/提取提示词/" "模型列表端点注释同步）中添加该分类，再重新执行本工具。" "现有分类：%s" % ("、".join(unknown), "、".join(sorted(known_caps)))) from appPublic.uniqueID import getID # 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}) await sor.sqlExe("COMMIT", {}) if recs: vendor_id = getattr(recs[0], "id", "") eps_old = [] try: eps_old = json.loads(getattr(recs[0], "endpoints", "") or "[]") except Exception: 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 pu = _urlparse(bu) if not pu.netloc: skipped_eps.append("%s（非法 URL：无主机名）" % bu) continue 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_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_out else: vendor_id = getID() 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), "description": spec.get("doc_notes", "") or "", "status": "active", "org_id": ctx.get("org_id", "") or "0"}) vendor_action = "新建供应商 %s（%s）" % (vendor_id, vendor_name) vendor_endpoints = eps_norm # 2. 适配模板：按 (协议×能力×形态指纹) 复用（2026-09-06 用户定夺）。 # 纯按能力复用的事故：happyhorse r2v 纯图 vs wan3.0 r2v 四模态，共享模板 # 静默丢媒体；一模型一模板则 wan3.1 同 API 时白白新增。指纹同→复用， # 指纹异→各建各的。旧骨架模板（硬编码 xxx_file/__from_doc__，2026-09-05 # 实测 t2v 因它渲染崩）视为不可用，按文档示例自愈重建。 # 旧命名 {protocol}-{cap}-自动配置（无指纹段）：形态一致则改名收编， # 不一致则保留（已挂模型继续可用），新形态另建。 profile_ids = {} tpl_errors = [] tpl_notes = [] tpl_notes.extend(cap_notes) # 能力守卫纠正/告警先进 notes（前置阶段产生） caps = sorted(set((m.get("capability") or "t2t") for m in models)) for cap in caps: # 先生成再比对：指纹来自生成结果（path/req/resp/schema 归一化 sha256[:12]） tpl = _gen_templates(protocol, cap, spec) dry_err = _dry_render_check(tpl["req"], cap, tpl["biz_params"], tpl["media_params"]) if dry_err: # 宁可不建，不埋雷：渲染不了的模板落库=运行时必崩，如实报错由助手修正规格重跑 # 模型已挂的旧模板不动（更新路径 IF 保护，profile_id 保持原值） tpl_errors.append("%s: %s" % (cap, dry_err)) continue fp = _tpl_fingerprint(tpl["path"], tpl["req"], tpl["resp"], tpl["param_schema"]) shared_name = "%s-%s-自动配置-%s" % (protocol, cap, fp) legacy_name = "%s-%s-自动配置" % (protocol, cap) recs = await sor.sqlExe( "SELECT id, name, path, request_template, response_template, param_schema " "FROM llm_api_profile " "WHERE name IN (${n1}$, ${n2}$) AND status='active' " "ORDER BY CASE WHEN name=${n1}$ THEN 0 ELSE 1 END LIMIT 1", {"n1": shared_name, "n2": legacy_name}) await sor.sqlExe("COMMIT", {}) if recs: pid_old = getattr(recs[0], "id", "") nm_old = getattr(recs[0], "name", "") old_tpl = (getattr(recs[0], "request_template", "") or "") + \ (getattr(recs[0], "response_template", "") or "") # schema 体检：存量脏 schema（未注册 uitype/default/重复字段/结构常量）判死重建， # 否则代码修了、用户前端还是坏表单（自愈门禁，见 _schema_violations） sv = _schema_violations(getattr(recs[0], "param_schema", "")) for s in sv['soft']: tpl_notes.append("模板 %s（%s）schema 提示：%s" % (pid_old, cap, s)) fp_old = _tpl_fingerprint(getattr(recs[0], "path", "") or "", getattr(recs[0], "request_template", "") or "", getattr(recs[0], "response_template", "") or "", getattr(recs[0], "param_schema", "") or "") broken = bool(sv['hard']) or any(mk in old_tpl for mk in _SKELETON_MARKERS) if not broken and fp_old == fp: if nm_old != shared_name: # 旧命名收编：形态一致，改名挂指纹（存量模型按 id 引用，不受影响） await sor.sqlExe( "UPDATE llm_api_profile SET name=${n}$ WHERE id=${i}$", {"n": shared_name, "i": pid_old}) await sor.sqlExe("COMMIT", {}) tpl_notes.append("模板 %s（%s）形态与本次生成一致，旧命名收编为 %s" % (pid_old, cap, shared_name)) profile_ids[cap] = pid_old continue if broken or nm_old == shared_name: # 坏模板（hard 违规/骨架标记）一律弃用——不论命名； # 指纹命名但内容漂移（人工改过库）的也弃用重建 await sor.sqlExe("UPDATE llm_api_profile SET status='deprecated' " "WHERE id=${i}$", {"i": pid_old}) await sor.sqlExe("COMMIT", {}) if sv['hard']: tpl_notes.append("存量模板 %s（%s）param_schema 不合规，已弃用重建：%s" % (pid_old, cap, "；".join(sv['hard']))) elif any(mk in old_tpl for mk in _SKELETON_MARKERS): tpl_notes.append("旧骨架模板 %s（%s）含硬编码占位符，已弃用并按文档示例重建" % (pid_old, cap)) else: tpl_notes.append("模板 %s（%s）内容与命名指纹不符（漂移），已弃用重建" % (pid_old, cap)) else: # 旧命名、健康、但形态不一致：保留（还挂着存量模型，弃用会断运行时）， # 新形态走下面新建 tpl_notes.append("旧命名模板 %s（%s）形态与本次生成不一致（指纹 %s≠%s），" "保留旧模板、按新形态另建" % (pid_old, cap, fp_old, fp)) pid = getID() await sor.C("llm_api_profile", { "id": pid, "name": shared_name, "protocol": protocol, "capability": cap, "path": tpl["path"], # method 独立列（2026-09-06 用户定夺）：提交类模板恒 POST "method": "POST", "headers": json.dumps(tpl["headers"], ensure_ascii=False), "request_template": tpl["req"], "response_template": tpl["resp"], "param_schema": json.dumps(tpl["param_schema"], ensure_ascii=False), "status": "active"}) profile_ids[cap] = pid for nt in tpl["notes"]: tpl_notes.append("%s: %s" % (cap, nt)) # 3. 模型：按 name/vendor_model_id 幂等（name 全局唯一键） doc_url = (spec.get("doc_url") or "").strip() created, updated, skipped = [], [], [] conflicts = [] # 供应商归属冲突（模型已挂别的供应商）——如实报告不静默迁移 for m in models: vmid = (m.get("vendor_model_id") or "").strip() if not vmid: skipped.append("(缺 vendor_model_id)") continue cap = m.get("capability") or "t2t" sync_mode = "async" if str(m.get("sync_mode") or "").strip() == "async" else "sync" desc = (m.get("description", "") or "").strip() # 出处标注（2026-09-05 用户规则）：模型注册表描述字段必须保存文档 URL if doc_url and ("[出处:" + doc_url + "]") not in desc: desc = (desc + " " if desc else "") + "[出处:%s]" % doc_url # 异步模型：后续步骤模板链（提交后顺序执行 query→[download]） query_ids = [] if sync_mode == "async": steps = spec.get("async_steps") or [] if not any((s.get("purpose") or "") == "query" for s in steps): desc = "⚠️异步模型但文档未提取到查询步骤（query_profile_ids 需人工补录）。" + desc for s in steps: qid = await _ensure_step_profile(sor, protocol, cap, vmid, s, spec) if qid: query_ids.append(qid) query_ids_json = json.dumps(query_ids) if query_ids else "" # 幂等键=name（uk_llm_model_name 全局唯一）。不能按 (vendor_id, vmid) 查： # 模型换挂供应商后（如迁到阿里云百炼），旧查询找不到→INSERT→撞唯一键崩（实测）。 recs = await sor.sqlExe( "SELECT id, vendor_id FROM llm_model WHERE name=${m}$ OR vendor_model_id=${m}$", {"m": vmid}) await sor.sqlExe("COMMIT", {}) if recs: mid = getattr(recs[0], "id", "") cur_vendor = getattr(recs[0], "vendor_id", "") or "" if cur_vendor and cur_vendor != vendor_id: # 供应商归属不一致：治理决策，不静默迁移——只更新元数据并如实报告 conflicts.append( "%s: 模型已存在且挂供应商 id=%s，与本次规格供应商（%s）不一致——" "未改供应商归属，仅更新描述/同步模式；如需迁移请明确指示" % (vmid, cur_vendor, vendor_name)) # profile_id 必须一并刷新（2026-09-05 实测 bug）：模型原挂模板可能已被 # 弃用重建（旧骨架自愈），不刷新则运行时仍指 deprecated 模板→必崩 # query_profile_ids 同理要 IF 保护（2026-09-06 实测 bug）：本次规格没带 # async_steps 时空串覆盖会把模型已挂好的查询模板清空 → 异步测试报 # 「未登记查询步骤模板」。空值=没提取到，不是「要清除」。 # capability 必须写回（2026-09-06 wan3.0 二次教训）：守卫纠正 t2v→r2v # 后模型行能力不更新=纠正只改了模板没改注册表，运行时仍按旧能力路由 new_pid = profile_ids.get(cap, "") await sor.sqlExe( "UPDATE llm_model SET description=${d}$, capability=${cap}$, " "sync_mode=${sm}$, query_profile_ids=IF(${q}$='', query_profile_ids, ${q}$), " "profile_id=IF(${p}$='', profile_id, ${p}$), updated_at=NOW() " "WHERE id=${i}$", {"d": desc, "cap": cap, "sm": sync_mode, "q": query_ids_json, "p": new_pid, "i": mid}) await sor.sqlExe("COMMIT", {}) updated.append(vmid) continue mid = getID() await sor.C("llm_model", { "id": mid, "vendor_id": vendor_id, "account_id": "", "name": vmid, "vendor_model_id": vmid, "capability": cap, "sync_mode": sync_mode, "profile_id": profile_ids.get(cap, ""), "query_profile_ids": query_ids_json, "ppid": "", "default_params": "{}", "status": "active", "description": desc, "org_id": "0"}) created.append(vmid) rt_note = "" if protocol not in RUNTIME_PROTOCOLS: rt_note = ("⚠️ 协议「%s」的适配模板已存档，但当前运行时调用链仅支持 %s——" "该供应商模型暂不可被产线直接调用" % (protocol, "/".join(RUNTIME_PROTOCOLS))) # 4. 账号端点绑定自检已删（2026-09-12 端点可互换重构）：端点是纯连接点、 # 对所有模型等价，账号 endpoint_ids 空=选用全部端点，不存在「协议不匹配 # 会 404」需要补绑的场景；请求形态由 profile.path（完整相对路径）承载。 account_notes = [] audit_note = await _audit_media_convention(sor, list(profile_ids.values())) # 5. 模型配套技能自动生成（2026-09-08 用户定夺：每个模型的加入需要有 # 配套 skill——记录该模型需要什么样的输入；invoke_model 完备性判断由 # LLM 按此技能对照用户已给内容裁决缺什么，不再靠硬编码能力表）。 # 数据全部来自本次落库结果（param_schema 的 required + 能力模态 + # 媒体契约），生成失败只记 note 不阻断（技能缺失时门禁降级为 # 「LLM 按能力语义判断 → 代码表兜底」，功能不失效）。 skill_notes = [] for m in models: vmid = (m.get("vendor_model_id") or "").strip() if not vmid: continue try: await _write_model_skill(sor, vmid, m, spec, profile_ids) skill_notes.append("%s: 配套技能已生成/更新" % vmid) except Exception as e: skill_notes.append("%s: 配套技能生成失败（%s）——完备性判断将按能力语义降级" % (vmid, str(e)[:80])) # 6. 自动产品映射（2026-09-08 用户需求：添加模型时如有产品模块自动做产品映射）。 # 根治记账 failed「模型未映射到产品」——不靠人工事后补。产品模块未装则 # 跳过不阻断（模型注册已成功）；映射幂等（已映射跳过+自愈其他漏映射模型）。 prod_ok, prod_note = await _sync_model_products(sor, org_id="0") return json.dumps({ "vendor": vendor_action, "profiles": profile_ids, "models_created": created, "models_updated": updated, "models_skipped": skipped, "vendor_conflicts": conflicts, "template_errors": tpl_errors, "template_notes": tpl_notes, "account_notes": account_notes, "skill_notes": skill_notes, "product_mapping": {"ok": prod_ok, "note": prod_note}, "runtime_note": rt_note, "media_audit": audit_note, }, ensure_ascii=False) # 生成类能力：出参是文件（图/视频/音频/3D），response 必须 downloadfile2url 落地 _MEDIA_CAPS = ('t2i', 'i2v', 't2v', 'r2v', 't2a', 'tts', 'i2i', 'v2v', '3d') async def _audit_media_convention(sor, profile_ids): """强制检查（2026-09-05 用户规则：每个 llm 配置都要检查媒体转换约定）： 生成类能力的适配模板—— request_template 含上传媒体参数时，必须用 {{b64media2url(request, xxx_file)}} 转本地公网 URL 再传上游； response_template 必须用 {{downloadfile2url(request, <产物url>)}} 把生成物落地为本地持久 URL（上游 URL 有效期短，视频仅 24 小时）。 返回检查结论（无问题为空串），随 apply 结果返回给内部 agent 转告。 """ if not profile_ids: return "" issues = [] for pid in profile_ids: recs = await sor.sqlExe( "SELECT name, capability, request_template, response_template " "FROM llm_api_profile WHERE id=${i}$", {"i": pid}) await sor.sqlExe("COMMIT", {}) if not recs: continue r = recs[0] cap = getattr(r, 'capability', '') or '' name = getattr(r, 'name', '') or pid if cap not in _MEDIA_CAPS: continue # 非生成类（t2t/embedding/rerank）无生成物 rt = getattr(r, 'request_template', '') or '' st = getattr(r, 'response_template', '') or '' if '__note__' in st and 'downloadfile2url' not in st: issues.append("「%s」response 模板是骨架——生成物必须用 " "downloadfile2url(request, <产物url>) 落地后再返回" % name) elif st and 'downloadfile2url' not in st: issues.append("「%s」response 模板缺 downloadfile2url——生成类能力" "（%s）的产物 URL 必须落地为本地持久 URL" % (name, cap)) # 上传参数线索：模板里出现 image/video/audio 文件参数但没有 b64media2url has_upload_ref = any(k in rt for k in ('image_file', 'video_file', 'audio_file', 'image_files', 'video_files', 'audio_files', 'first_frame', 'image_url', 'media')) if has_upload_ref and 'b64media2url' not in rt: # 三数组直通透传（image_files 等直接渲染进 content）是内部执行器的 # 合法路径：request=None 时 b64media2url 兜底缺 BASEURL 会拼出坏路径 # （2026-09-16 实测），data URL/公网 URL 直传上游已验证可用。 three_arr = any(k in rt for k in ('image_files', 'video_files', 'audio_files')) if not three_arr: issues.append("「%s」request 模板含媒体上传参数但缺 b64media2url——" "上传文件必须转本地公网 URL 再传上游" % name) # 媒体输入能力模板必须渲染输入媒体（2026-09-16 i2i 闭眼编造教训）： # i2i/i2v/r2v 等能力的 request_template 若不引用任何输入媒体变量， # 调用方传的 image_files 会在模板层静默丢失 → 模型无输入图纯编造。 if cap in _MEDIA_INPUT_CAPS and not has_upload_ref: issues.append("「%s」能力 %s 需要输入媒体，但 request 模板未渲染任何输入" "媒体变量（image_files/media 等）——输入图会在模板层丢失，" "模型将无图编造；请补 content 图像数组或对应媒体参数" % (name, cap)) if issues: return "⚠️ 媒体转换约定检查未通过：" + "；".join(issues) return "" # 能力 → 统一出参键（同类能力对外契约一致：视频出 video、图出 image） _MEDIA_OUT_KEY = { 't2v': 'video', 'i2v': 'video', 'v2v': 'video', 'r2v': 'video', 't2i': 'image', 'i2i': 'image', 'tts': 'audio', 't2a': 'audio', '3d': 'glb', } # 能力 → 上游产物字段名兜底（文档没给响应示例时用供应商惯例） _MEDIA_OUT_FIELD = { 'video': 'video_url', 'image': 'image_url', 'audio': 'audio_url', 'glb': 'model_url', } # 文档示例里表示「上传媒体」的结构线索 _MEDIA_TYPES = ('first_frame', 'last_frame', 'ref_image', 'ref_images', 'image', 'img', 'reference_image', 'reference_video', 'reference_audio', 'reference') _MEDIA_KEY_HINTS = ('img_url', 'image_url', 'first_frame_url', 'last_frame_url', 'ref_image_url', 'video_url', 'audio_url') # 需要「输入媒体」的能力（这些能力里出现 URL 叶子才判定为上传媒体） _MEDIA_INPUT_CAPS = ('i2v', 'i2i', 'v2v', 'i2t', 'm2t', '2i2v', 'r2v') def _media_param_name(typeval, keyname): """按文档媒体结构推断运行时统一媒体参数名。 2026-09-06 用户定夺契约（对齐 sage/llmage）：媒体统一三数组参数 image_files / audio_files / video_files——值是数组或单字符串， 模板 Jinja 动态判断两种形态；有些模型三种任意组合（r2v 最多 9 图）， 有些只支持图片，按文档示例动态组装，不固定槽位。 """ t = ('%s %s' % (typeval or '', keyname or '')).lower() if 'video' in t: return 'video_files' if 'audio' in t: return 'audio_files' return 'image_files' def _media_list_expr(param): """Jinja 表达式：params. 归一为列表（字符串→单元素；缺省→空列表）。""" return ("(params.{p} if params.{p} is not string and params.{p} " "else ([params.{p}] if params.{p} is string else []))").format(p=param) def _media_scalar_expr(param): """Jinja 表达式：params. 取单值（字符串原样；数组取首个；缺省→空串）。""" return ("(params.{p} if params.{p} is string " "else ((params.{p} or [''])[0]))").format(p=param) def _is_url_example(v): return isinstance(v, str) and v.strip().lower().startswith(('http://', 'https://')) def _example_to_nested(fields): """request_fields 点号路径 → 嵌套 dict（文档无请求示例时的兜底结构来源）。""" root = {} for f in fields or []: parts = [p for p in str(f).split('.') if p] if not parts: continue cur = root for p in parts[:-1]: nxt = cur.get(p) if not isinstance(nxt, dict): nxt = {} cur[p] = nxt cur = nxt cur[parts[-1]] = '' return root def _is_media_array(lst, capability): """判定「媒体数组」（2026-09-06 r2v 教训）：文档示例里 [{"type": "reference_image", "url": "https://..."}, ...] 这类变长数组 （r2v 最多 9 张参考图，有些模型图/视频/音频任意组合）。 判据：非空 dict 列表，每个元素都含 URL 叶子且 type/键名带媒体线索 （或能力属输入媒体类）。整组换 Jinja 动态组装，**禁止按示例元素个数 展开固定槽位**——那是上一版 r2v 模板渲染崩（image_file）的根因。 """ if not isinstance(lst, list) or not lst: return False for d in lst: if not isinstance(d, dict): return False tval = str(d.get('type') or '').lower() url_hit = any(_is_url_example(v) for v in d.values()) hinted = (tval in _MEDIA_TYPES or any(str(k).lower() in _MEDIA_KEY_HINTS for k in d) or capability in _MEDIA_INPUT_CAPS) if not (url_hit and hinted): return False return True def _media_any_expr(mps): """「任一媒体参数有值」Jinja 布尔表达式（字符串/数组两形态归一）。 2026-09-06 用户纠正后引入：全能参考模型（wan3.0 类 r2v）文档明示 「prompt 和 media 必填其一」「文生视频不传入任何媒体文件」——零媒体是 合法 t2v 形态且文档示例**不带 media 键**。无条件渲染会发出 "media": [] 空数组（API 未承诺接受，不能赌），故 media 键必须条件渲染：有媒体才 输出键，零媒体整个省略。真实示例路径与合成兜底路径共用本表达式。 """ return ' or '.join( '(params.%s if params.%s is not string and params.%s else ' '([params.%s] if params.%s is string else []))' % ((mp,) * 5) for mp in mps) def _media_array_expr(elements): """媒体数组示例 → Jinja 动态组装表达式（三数组契约 image_files/ audio_files/video_files，值为字符串或数组均可）。 按元素 type/键名归组到统一参数，每组渲染一个 for 循环（元素结构逐字 保留，仅 URL 叶子换 b64media2url）；混合类型多循环用 namespace 计数 合并逗号。返回 (expr, media_params)。 """ groups, seen = [], set() for d in elements: if not isinstance(d, dict): continue tval = d.get('type') if isinstance(d.get('type'), str) else '' urlkey = next((k for k, v in d.items() if _is_url_example(v)), '') if not urlkey: continue mp = _media_param_name(tval, urlkey) if mp in seen: continue seen.add(mp) parts = [] for k, v in d.items(): if k == urlkey: parts.append('"%s": {{ b64media2url(request, _f)|tojson }}' % k) else: parts.append('%s: %s' % (json.dumps(str(k), ensure_ascii=False), json.dumps(v, ensure_ascii=False))) groups.append((mp, '{%s}' % ', '.join(parts))) if not groups: return '', [] params = [g[0] for g in groups] if len(groups) == 1: mp, elem = groups[0] expr = ('[{%% for _f in %s %%}%s{%% if not loop.last %%},' '{%% endif %%}{%% endfor %%}]' % (_media_list_expr(mp), elem)) return expr, params segs = ['[{% set _c = namespace(n=0) %}'] for mp, elem in groups: segs.append('{%% for _f in %s %%}{%% if _c.n %%},{%% endif %%}' '{%% set _c.n = _c.n + 1 %%}%s{%% endfor %%}' % (_media_list_expr(mp), elem)) segs.append(']') return ''.join(segs), params def _dashscope_envelope(example): """DashScope SDK kwargs 形态 → 原生 HTTP 信封归一（2026-09-08 实测根因）。 文档常只给 Python SDK 示例（MultiModalConversation.call(messages=..., size=...)），提取 LLM 照抄成平铺 {model, messages, size, watermark...}； 但原生 HTTP 端点要求 {"model", "input": {"messages": [...]}, "parameters": {...}}——平铺体提交必 400 InvalidParameter （qwen-image-plus 实测「messages parameter invalid」）。 代码级确定性归一（不靠提取 LLM 自觉）：顶层有 messages 且无 input 信封 时，messages 进 input，除 model/input/parameters 外的业务键全进 parameters。 """ if not isinstance(example, dict) or not example: return example if 'input' in example or 'parameters' in example: return example # 已是信封形态（curl 示例），不动 msgs = example.get('messages') if not isinstance(msgs, list): return example # 非 SDK kwargs 形态，不动 out = {'model': example.get('model', '')} out['input'] = {'messages': msgs} params = {k: v for k, v in example.items() if k not in ('model', 'messages')} if params: out['parameters'] = params return out def _tpl_from_example(example, capability): """文档请求示例 → Jinja2 请求体模板（结构逐字保留，值换成运行时变量）。 规则（2026-09-06 三数组契约版，对齐 sage/llmage）： model 键 → {{ model|tojson }} prompt/text 键 → {{ prompt|tojson }} 媒体数组（r2v media 等变长数组）→ Jinja for 循环按 image_files/ audio_files/video_files 动态组装（见 _media_array_expr） 单值上传媒体 → {{ b64media2url(request, )|tojson }} （params.xxx_files 字符串/数组两形态兼容） 其他标量 → {{ params.|default(<文档示例值>)|tojson }} **示例里没有上传媒体就不生成媒体参数**——纯文生视频(t2v)曾因硬编码 params.xxx_file 骨架在运行时崩（'dict object' has no attribute 'xxx_file'）。 返回 (模板串, media_params, biz_params, used_example)。 """ tokens = {} cond_tokens = {} media_params, biz_params = [], [] used_example = isinstance(example, dict) and bool(example) tree = example if used_example else _example_to_nested(None) def _put(expr): k = '__TPL%d__' % len(tokens) tokens[k] = expr return k def walk(node, parent_key='', sib_type=None): if isinstance(node, list): if _is_media_array(node, capability): expr, mps = _media_array_expr(node) if expr: for mp in mps: if mp not in media_params: media_params.append(mp) tk = _put(expr) # media 键条件渲染（2026-09-06 用户纠正）：零媒体=合法 t2v # 形态且文档示例不带 media 键，无条件渲染发出 "media": [] # 空数组（API 未承诺接受，不能赌）。记录 (token→键名/媒体参数) # 供替换阶段整键包 Jinja 条件。仅 dict 下的数组键可条件化。 if parent_key: cond_tokens[tk] = (parent_key, mps) return tk return [walk(v, parent_key, sib_type) for v in node] if not isinstance(node, dict): return node tval = node.get('type') if isinstance(node.get('type'), str) else sib_type out = {} for k, v in node.items(): if isinstance(v, (dict, list)): out[k] = walk(v, k, tval) continue kl = str(k).lower() if kl == 'model' and isinstance(v, str): out[k] = _put('{{ model|tojson }}') continue if kl in ('prompt', 'text') and isinstance(v, str): out[k] = _put('{{ prompt|tojson }}') continue is_media = _is_url_example(v) and ( str(tval or '').lower() in _MEDIA_TYPES or kl in _MEDIA_KEY_HINTS or capability in _MEDIA_INPUT_CAPS) if is_media: mp = _media_param_name(tval, k) if mp not in media_params: media_params.append(mp) out[k] = _put('{{ b64media2url(request, %s)|tojson }}' % _media_scalar_expr(mp)) continue biz_params.append({'name': k, 'example': v}) if v == '' or v is None: out[k] = _put('{{ params.%s|tojson }}' % k) # 无示例值：调用必传 else: # 数值强转（2026-09-06 真实调用实抓）：wan3.0 零媒体 t2v 提交 # 报 InvalidParameter "Input should be a valid integer: # parameters.duration"——params 来自表单/agent 时常为字符串 "5"， # 直接 tojson 出 "5"（带引号）被 API 拒。文档示例是 int/float 的 # 参数按示例类型强转，字符串数字也能落到正确 JSON 类型。 # bool 是 int 子类须先排除（bool 不加 |int，避免 UI 真值语义被改写）。 coerce = '' if isinstance(v, bool): pass elif isinstance(v, int): coerce = '|int' elif isinstance(v, float): coerce = '|float' out[k] = _put('{{ (params.%s|default(%s))%s|tojson }}' % (k, json.dumps(v, ensure_ascii=False), coerce)) return out body = walk(tree) s = json.dumps(body, ensure_ascii=False) for k, expr in tokens.items(): if k in cond_tokens: # media 键条件渲染：零媒体时整键省略（对齐文档 t2v 形态）。 # json.dumps 分隔符固定 ", "，按键在对象中的位置分三模式吃逗号： # 首/中键（后随 ", "）→ 条件段自带尾逗号 # 末键（前缀 ", "） → 条件段吃掉前逗号 # 独键 → 直接包 keyname, mps = cond_tokens[k] anyexpr = _media_any_expr(mps) seg_body = '"%s": %s' % (keyname, expr) pat_first = '"%s": "%s", ' % (keyname, k) pat_last = ', "%s": "%s"' % (keyname, k) pat_only = '"%s": "%s"' % (keyname, k) if pat_first in s: s = s.replace(pat_first, ' {%% if %s %%}%s, {%% endif %%}' % (anyexpr, seg_body)) elif pat_last in s: s = s.replace(pat_last, '{%% if %s %%}, %s{%% endif %%}' % (anyexpr, seg_body)) elif pat_only in s: s = s.replace(pat_only, ' {%% if %s %%}%s{%% endif %%}' % (anyexpr, seg_body)) continue s = s.replace('"%s"' % k, expr) return s, media_params, biz_params, used_example def _collect_url_fields(node, prefix=''): """递归收集响应示例里的 URL 字段点号路径（如 output.video_url）。""" hits = [] if isinstance(node, dict): for k, v in node.items(): p = (prefix + '.' + str(k)) if prefix else str(k) if isinstance(v, (dict, list)): hits.extend(_collect_url_fields(v, p)) elif _is_url_example(v): hits.append(p) elif isinstance(node, list): for i, v in enumerate(node[:3]): hits.extend(_collect_url_fields(v, prefix)) return hits def _resp_tpl_for(capability, spec): """生成类响应模板：产物 URL 经 downloadfile2url 落地（上游 URL 有效期短）。 产物字段路径优先取文档响应示例里的 URL 字段（递归，支持 output.video_url 这类嵌套——运行时 ns 有 output/usage/task_id），缺则用供应商惯例兜底， 并在 notes 里如实说明来源（不假装是文档确证）。 """ outkey = _MEDIA_OUT_KEY.get(capability, 'video') field = '' rex = spec.get('response_example') if isinstance(rex, dict): hits = _collect_url_fields(rex) # 优先 *_url 结尾且与产物类型匹配的字段 want = ('video', 'image', 'audio', 'glb', 'model') for h in hits: leaf = h.split('.')[-1].lower() if leaf.endswith('_url') and any(w in leaf for w in want): field = h break if not field and hits: field = hits[0] note = '' proto = str(spec.get('protocol') or '').strip() if proto == 'dashscope_sync': # DashScope 同步生成实测响应形态（2026-09-08 qwen-image-plus 真图验证）： # output.choices[0].message.content[0].{image|video|audio} # 同步路径渲染命名空间无 task_id/status（StrictUndefined 会崩）， # 模板只引用 output/usage/request 三个必有变量。 # 优先级：文档响应示例给出的 output.* 产物路径 > 实测兜底形态。 if field and field.startswith('output.'): path_expr = field note = '' # 文档确证，无需兜底声明 else: seg_key = {'image': 'image', 'video': 'video', 'audio': 'audio'}.get(outkey, outkey) path_expr = 'output.choices[0].message.content[0].%s' % seg_key if not field: note = ('产物路径按 DashScope 同步生成实测形态生成（%s），' '文档未提供响应示例，如上游形态不同需修正' % path_expr) resp = ('{"%s": "{{ downloadfile2url(request, %s) }}", ' '"usage": {{ json.dumps(usage) }}}' % (outkey, path_expr)) return resp, note if not field: field = _MEDIA_OUT_FIELD.get(outkey, 'result_url') note = ('产物字段名「%s」按供应商惯例兜底（文档未提供响应示例或示例中无 URL 字段），' '如上游字段不同需修正' % field) # usage 值不加包裹引号（2026-09-06 实测 bug）：`"usage": "{{ json.dumps(usage) }}"` # 在 usage 非空时内层双引号破坏外层 JSON 字符串值 → json.loads 断在 char 161， # 渲染整体回退、生成物不落地。渲染为嵌套 JSON 对象才是合法形态，手工拼接： resp = ('{"status": "SUCCEEDED", "%s": "{{ downloadfile2url(request, %s) }}", ' '"usage": {{ json.dumps(usage) }}, "task_id": "{{ task_id }}"}' % (outkey, field)) return resp, note def _headers_from_spec(spec): """请求头：认证头统一 api_key 变量，其余按文档示例逐字带上 （如 DashScope 异步必需的 X-DashScope-Async: enable）。""" headers = {'Authorization': 'Bea' + 'rer {{api_key}}', 'Content-Type': 'application/json'} rh = spec.get('request_headers') if isinstance(rh, dict): for k, v in rh.items(): if str(k).lower() in ('authorization', 'content-type'): continue if isinstance(v, str) and v.strip(): headers[k] = v.strip() # 协议兜底：dashscope_async 提交必须带异步开关头（文档示例遗漏也不至于提交即失败）； # dashscope_sync 是同步生成（MultiModalConversation stream=False 类，即时返回产物）， # **绝不能带异步头**——带了上游按异步受理，不支持异步的密钥直接 403 # （2026-09-08 qwen-image-plus 实测：'current user api does not support asynchronous calls'）。 if str(spec.get('protocol') or '').strip() == 'dashscope_async': headers.setdefault('X-DashScope-Async', 'enable') elif str(spec.get('protocol') or '').strip() == 'dashscope_sync': headers.pop('X-DashScope-Async', None) return headers 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 = _norm_relative_path( spec.get("chat_path") or "/chat/completions") headers = _headers_from_spec(spec) notes = [] if protocol == "openai_compat": # 2026-09-06 修复（豆包 pro 配置实抓）：旧模板用 json.dumps 包占位符—— # 1. "messages": "{{messages}}" 带引号，渲染成 Python repr 字符串而非 # JSON 数组，上游必拒； # 2. "{{temperature}}" 运行时命名空间无此变量（inference._build_upstream_body # 的 ns 是 model/messages/extra/params/api_key/org_id），StrictUndefined # 干跑必崩 → 落库被拦，profiles 恒空，运行时侥幸走 _BUILTIN_REQUEST_TMPL。 # 正确形态对齐 pipeline-llm 内置模板：tojson 无引号 + extra 白名单透传循环 # （temperature/top_p/max_tokens 等调用方传了才发，不硬编码业务参数）。 req = ('{"model": {{ model|tojson }}, "messages": {{ messages|tojson }}' '{% for k, v in extra.items() %}, "{{ k }}": {{ v|tojson }}' '{% endfor %}, "stream": false}') resp = json.dumps({ "content": "choices[0].message.content", "usage": {"prompt_tokens": "usage.prompt_tokens", "completion_tokens": "usage.completion_tokens"}, }, ensure_ascii=False) return {"path": chat_path, "headers": headers, "req": req, "resp": resp, "param_schema": _build_param_schema(capability, [], []), "media_params": [], "biz_params": [], "notes": notes} example = spec.get("request_example") if not (isinstance(example, dict) and example): example = _example_to_nested(spec.get("request_fields")) notes.append("文档未提供完整请求示例，模板按 request_fields 字段路径生成，" "业务参数无默认值（调用时必须显式传入），建议人工核对") # DashScope SDK kwargs → 原生 HTTP 信封归一（2026-09-08 qwen-image-plus 实测： # 平铺体提交 400 InvalidParameter，必须 input.messages + parameters 信封） if protocol in ('dashscope_sync', 'dashscope_async'): example = _dashscope_envelope(example) req, media_params, biz_params, _used = _tpl_from_example(example, capability) if capability in _MEDIA_CAPS: resp, rnote = _resp_tpl_for(capability, spec) if rnote: notes.append(rnote) if not media_params and capability in _MEDIA_INPUT_CAPS: if capability == 'r2v' and (example or {}).get('input') is not None: # 页面示例是纯文本最简形态（SPA 懒加载抓不到完整示例，wan3.0-video-prime # 实测：概述明写"四模态全能参考"、示例只有 input.prompt）——r2v 必有 # 参考媒体，DashScope 家族契约是 input.media 数组 [{type,url}]， # 类型集 image→reference_image / video→reference_video / audio→reference_audio。 # 复用 _media_array_expr（e2e 验证过的 namespace 混合类型写法）按契约 # 合成三数组，比只报警更有用：用户填了参考媒体就能发出去。 # 如实注明合成来源，不假装是文档确证。 try: import re as _re synth = [{'type': 'reference_image', 'url': 'https://synth.invalid/ref.jpg'}, {'type': 'reference_video', 'url': 'https://synth.invalid/ref.mp4'}, {'type': 'reference_audio', 'url': 'https://synth.invalid/ref.mp3'}] expr, mps = _media_array_expr(synth) m = _re.search(r'("input"\s*:\s*\{)', req) if expr and m: # 条件渲染（2026-09-06 用户纠正后补）：wan3.0 类全能参考模型 # 文档明示「prompt 和 media 必填其一」「文生视频不传入任何媒体 # 文件」——零媒体是合法 t2v 形态，且文档 t2v 示例**不带 media # 键**；无条件渲染会发出 "media": [] 空数组（API 未承诺接受， # 不能赌）。有媒体才渲染 media 键，零媒体整个省略。 anyexpr = _media_any_expr(mps) # 前导空格不可省：段以 {% 开头，紧跟 input 的 { 会拼成 # "{{%" —— Jinja 词法把 {{ 当变量表达式开始 → unexpected '%' # （2026-09-06 单测实抓）。JSON 空白无副作用。 media_seg = (' {%% if %s %%}"media": %s, {%% endif %%}' % (anyexpr, expr)) pos = m.end() req = req[:pos] + media_seg + req[pos:] for mp in mps: if mp not in media_params: media_params.append(mp) notes.append("示例未见参考媒体（页面示例为纯文本最简形态），" "r2v 契约合成 input.media 数组（reference_image/" "video/audio 三类，DashScope 家族契约），零媒体时" "省略 media 键（对齐文档 t2v 形态）——" "建议对照供应商 API 参考页人工核对") except Exception as _e: notes.append("r2v 媒体合成失败（%s），示例未见上传字段，请人工核对" % str(_e)[:80]) elif capability == 'i2i' and protocol in ('dashscope_sync', 'dashscope_async'): # i2i 契约合成（2026-09-16 实测教训：qwen-image-3.0-pro 自动配置出的 # 模板 content 只有 text 项，image_files 在模板层丢失 → 模型"闭眼 i2i" # 纯编造画面，白皮书配图 QA 逐字核对实锤）。DashScope 图像编辑家族契约： # input.messages[].content = [{"image": url}, ..., {"text": prompt}]。 # 页面示例常只给纯文本最简形态 → 在 content 数组头部条件渲染 image 项 # （零媒体时整个省略，对齐 t2i 合法形态），与 r2v 合成同款思路。 import re as _re2 m2 = _re2.search(r'("content"\s*:\s*\[)', req) if m2 and 'image_files' not in req: # 单 % ：本串不经 % 格式化（r2v precedent 的 %% 是因为它走 # '...%s...' % (...) 格式化）；写 %% 会把字面 {%% 留进模板 # 致 Jinja "tag name expected"（2026-09-16 单测实抓）。 seg = ('{% for img in (params.image_files or []) %}' '{"image": {{ img|tojson }}},{% endfor %}') pos2 = m2.end() req = req[:pos2] + seg + req[pos2:] if 'image_files' not in media_params: media_params.append('image_files') notes.append("示例未见输入图（页面示例为纯文本最简形态），按 " "DashScope 图像编辑家族契约合成 content 图像数组" "（image_files 三数组，零媒体时省略）——" "建议对照供应商 API 参考页人工核对") else: notes.append("能力 %s 通常需要输入媒体，但文档示例未见上传字段——" "模板未生成媒体参数，请核对文档" % capability) else: resp = json.dumps({"content": "{{ text }}"}, ensure_ascii=False) # field_enums：文档参数说明表里的可选值（提取层逐字照抄），驱动 code 下拉 enums = spec.get("field_enums") if isinstance(spec.get("field_enums"), dict) else None schema = _build_param_schema(capability, media_params, biz_params, enums) return {"path": chat_path, "headers": headers, "req": req, "resp": resp, "param_schema": schema, "media_params": media_params, "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) 等。 # 新三数组契约表达式是 params.image_files if ...，不会误伤。 'params.image_file)', 'params.video_file)', 'params.audio_file)') def _tpl_fingerprint(path: str, req: str, resp: str, schema) -> str: """模板形态指纹（2026-09-06 用户定夺）：提交模板复用键=协议×能力×指纹。 为什么不能只按能力复用（实测事故）：happyhorse r2v 纯图参考、wan3.0 r2v 四模态参考（图/视频/音频）——按能力共享一份模板，后配模型的媒体模态被 静默丢弃。也不能一模型一模板：wan3.1 若 API 与 wan3.0 完全一致应零新增。 指纹=归一化 (path, request_template, response_template, param_schema) 的 sha256 前 12 位：形态同→复用，形态异→各建各的，无需人工设计签名维度。 归一化规则：path 去空白；req/resp/schema JSON 往返（loads→dumps sorted 紧凑分隔符）消除键序/空白差异；解析失败退化为去空白原文（保守不误判同形）。 headers/method 不入指纹：headers 按协议统一，提交模板 method 恒 POST。 """ def _canon(s): s = (s or "").strip() if not s: return "" try: return json.dumps(json.loads(s), sort_keys=True, ensure_ascii=False, separators=(",", ":")) except Exception: return re.sub(r"\s+", "", s) if not isinstance(schema, str): schema = json.dumps(schema, sort_keys=True, ensure_ascii=False) blob = "|".join([(path or "").strip(), _canon(req), _canon(resp), _canon(schema)]) return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:12] def _schema_violations(raw): """存量 param_schema 合规体检（2026-09-06 用户指正后补的自愈门禁）。 为什么必须有它：apply 的模板复用门禁原先只看 request_template 里的骨架标记， 完全不看 param_schema —— 于是「模板正常但 schema 是 textarea/number/default」的 存量 profile 会被永久复用（实测 happyhorse r2v/t2v/i2v 三个 active profile 全中招， r2v 还带 3 个重复 type 字段）。代码修好了，用户在前端看到的仍是坏表单： Input.create 遇未注册 uitype 返回 null 只打一行 debug（input.js:1294），字段静默消失。 所以复用前必须体检 schema，违规一律弃用重建，让存量数据跟着代码自愈。 分级（避免误杀仿权威样例的 profile）： hard = 会导致字段消失/撞名覆盖/默认值不生效 → 判死重建 （未注册 uitype、用 default、同名字段重复、结构常量入表单） soft = 功能受限但不崩 → 只提示，不判死。典型：媒体数组缺 multiple:true （input.js:426 决定值是数组还是 files[0]；权威样例 minimax_h3_setup.sql:13 的 image_files 就没带 multiple，模板靠 for 循环兼容 str，故不能判死） 返回 {'hard': [...], 'soft': [...]}。 """ out = {'hard': [], 'soft': []} raw = (raw or "").strip() if not raw: return out try: schema = json.loads(raw) except Exception as exc: out['hard'].append("param_schema 不是合法 JSON（%s）" % exc) return out if not isinstance(schema, list): out['hard'].append("param_schema 顶层应为数组") return out seen = set() for f in schema: if not isinstance(f, dict): out['hard'].append("schema 元素非对象: %r" % (f,)) continue nm = f.get("name") ut = f.get("uitype") if ut not in _BRICKS_UITYPES: out['hard'].append("字段 %s 的 uitype %r 未注册" % (nm, ut)) if "default" in f: out['hard'].append("字段 %s 用了 default（应为 defaultvalue）" % nm) # 同名字段重复：前端 dom_element.id 撞名，后者覆盖前者 if nm in seen: out['hard'].append("字段 %s 重复出现" % nm) seen.add(nm) # 结构常量不该出现在表单（值已固化在模板里） if nm in _STRUCT_CONST_FIELDS: out['hard'].append("结构常量字段 %s 不该让用户填" % nm) # 媒体数组参数缺 multiple：UI 只能选 1 个（soft，权威样例也这样） if nm in _MEDIA_UITYPE and not f.get("multiple"): out['soft'].append("媒体数组字段 %s 缺 multiple:true（UI 只能选 1 个）" % nm) return out def _biz_field_uitype(name, example, enums): """业务参数 → bricks uitype（2026-09-06 用户指正：只准用注册过的类型）。 合法全集（bricks/input.js Input.register）：str/hide/tel/date/int/float/check/ checkbox/email/file/image/code/text/password/audio/video/audiorecorder/ audiotext/search/group。**textarea / number 均未注册**，用了前端渲染不出来。 规则：有枚举 → code（UiCode=select，data:[{value,text}]）； 数值 → int/float；布尔 → check；其余 → text（UiText 自动增高多行）。 返回 (uitype, extra_opts)。 """ opts = {} vals = None if isinstance(enums, dict): for k in enums: if str(k).lower() == str(name).lower(): v = enums[k] if isinstance(v, (list, tuple)) and v: vals = list(v) break if vals: opts['data'] = [{'value': v, 'text': ('%s' % v)} for v in vals] return 'code', opts if isinstance(example, bool): return 'check', opts if isinstance(example, int): return 'int', opts if isinstance(example, float): return 'float', opts # 文档没给示例值也没枚举：一律 text（不是未注册的 textarea/number） return 'text', opts # 媒体参数 → bricks 专用上传控件（UiImage/UiAudio/UiVideo 均继承 UiFile） _MEDIA_UITYPE = {'image_files': 'image', 'audio_files': 'audio', 'video_files': 'video'} # 媒体参数的中文标签（人话，不做机械拼接） _MEDIA_LABEL = {'image_files': '参考图片', 'audio_files': '参考音频', 'video_files': '参考视频'} # bricks 已注册的 uitype 全集（input.js 末尾 Input.register 逐行抄录，2026-09-06）。 # 白名单硬校验：写进 param_schema 的类型若不在其中，前端 Input.create 返回 null # 只打一行 debug 日志（input.js:1294），字段静默消失——比报错更难查。 _BRICKS_UITYPES = frozenset([ 'str', 'hide', 'tel', 'date', 'int', 'float', 'check', 'checkbox', 'email', 'file', 'image', 'code', 'text', 'password', 'audio', 'video', 'audiorecorder', 'audiotext', 'search', 'group', ]) # 文档结构常量：值在模板里已逐字固化，不该让用户填（r2v 的 media[].type 等） _STRUCT_CONST_FIELDS = frozenset(['type', 'role', 'format', 'mime_type']) def _validate_uitypes(schema): """param_schema uitype 白名单校验（防未注册类型静默丢字段）。 另校验默认值字段名：只准 defaultvalue——input.js 从不读 default， 写成 default 的默认值前端不生效（2026-09-06 实测踩坑）。 违规直接抛异常，让 apply 当场失败并报进 template_errors，不落库坏 schema。 """ for f in schema: ut = f.get('uitype') if ut not in _BRICKS_UITYPES: raise ValueError('param_schema 非法 uitype %r（字段 %s）——bricks 已注册类型: %s' % (ut, f.get('name'), '/'.join(sorted(_BRICKS_UITYPES)))) if 'default' in f: raise ValueError('param_schema 字段 %s 用了 default——bricks 只认 defaultvalue' % f.get('name')) return True def _build_param_schema(capability, media_params, biz_params, enums=None): """生成 bricks 合规的参数表单 schema。 合规要点（2026-09-06，对照 bricks/input.js + uapi/sql/minimax_h3_setup.sql 权威样例）： - uitype 只取注册过的值（禁 textarea/number） - 默认值字段名必须是 defaultvalue（input.js 只读它，default 无人消费） - 数组媒体参数带 multiple:true（input.js:426 决定值是数组还是单文件； r2v 最多 9 张参考图，缺它 UI 只能选 1 张） - 媒体 uitype 用专用控件 image/audio/video（非泛用 file），带预览+相机 - 文档结构常量（如 media 元素的 type: "reference_image"）不进表单—— 模板里已逐字固化，让用户填只会出错（旧版曾生成 3 个重复 type 字段） - 同名参数去重（数组多元素曾各自产出一条同名业务参数） """ schema = [{'name': 'prompt', 'label': '提示词', 'uitype': 'text', 'required': capability not in ('embedding', 'rerank')}] seen = {'prompt'} for mp in media_params: if mp in seen: continue seen.add(mp) schema.append({'name': mp, 'label': _MEDIA_LABEL.get(mp, mp), 'uitype': _MEDIA_UITYPE.get(mp, 'file'), # 媒体一律非必填（2026-09-06 用户纠正）：全能参考模型(wan3.0) # 三类媒体可自由组合、且允许全不传——全不传即纯文生视频(t2v)。 # 文档明示「prompt 和 media 必填其一」「文生视频仅通过 prompt， # 不传入任何媒体文件」。required=True 会把合法的 t2v / 部分组合 # 模式全堵死。prompt 保持必填，作「必填其一」的锚。 # 纯参考模型(无 t2v 模式)若空媒体提交，API 会拒、test_model_call # 当场暴露，非静默失败——故 UI 层一律非必填是安全默认。 'required': False, # 三数组契约：UI 允许多选，值成数组；运行时兼容字符串/数组两形态 'multiple': True}) for b in biz_params: name = b['name'] # 结构常量不进表单：媒体元素的 type 等已在模板里逐字固化 if name in seen or str(name).lower() in _STRUCT_CONST_FIELDS: continue seen.add(name) uitype, extra = _biz_field_uitype(name, b.get('example'), enums) f = {'name': name, 'label': b.get('label') or name, 'uitype': uitype, 'required': False} f.update(extra) if b.get('example') not in (None, ''): f['defaultvalue'] = b['example'] schema.append(f) _validate_uitypes(schema) return schema def _dry_render_check(req_tpl, capability, biz_params, media_params): """落库前干跑渲染（StrictUndefined）：模板引用了运行时拿不到的变量当场报错。 运行时命名空间见 pipeline-llm inference._build_async_body： model / prompt / messages / params(业务参数) / api_key / org_id + request / json / b64media2url / downloadfile2url。 旧缺陷正是漏了这步：模板硬编码 params.xxx_file，直到 test_model_call 才崩。 """ try: from jinja2 import Environment, StrictUndefined except Exception as e: return "干跑校验跳过（jinja2 不可用：%s）" % str(e)[:60] def _stub(request, value, *a, **k): return str(value) # StrictUndefined 传入未定义值时在此抛错 params_arr = {} for b in biz_params or []: # 干跑模拟「业务参数全部提供」场景：无示例值的参数给占位串 # （模板 {{ params.x|tojson }} 无默认——运行时必须显式传，属预期契约） params_arr[b['name']] = b['example'] if b.get('example') not in ('', None) else 'dry' params_str = dict(params_arr) for mp in media_params or []: # 三数组契约两形态都要干跑（2026-09-06）：数组走 for 循环路径， # 字符串走 `is string` 归一路径——模板只兼容其一会在运行时崩 params_arr[mp] = ['https://dry-run.invalid/sample.bin'] params_str[mp] = 'https://dry-run.invalid/sample.bin' errors = [] for tag, params in (('数组形态', params_arr), ('字符串形态', params_str), # 零媒体形态（2026-09-06）：全能参考模型(r2v)文档明示 # 「prompt 和 media 必填其一」，零媒体=合法 t2v 模式且 # 文档示例不带 media 键——模板条件渲染的 False 分支 # 必须干跑出合法 JSON（省略 media 键，非空数组）。 # 只测空数组 []（运行时真实「未选媒体」场景，schema 字段 # 存在值为空）；不测缺键——运行时用默认 Undefined 容错， # StrictUndefined 下缺键会误报，二者行为不一致。 ('零媒体形态', dict(params_arr, **{mp: [] for mp in media_params or []}))): ns = {'model': 'dry-run-model', 'prompt': '干跑校验', 'messages': [], # extra：运行时白名单透传参数（inference._build_upstream_body 注入， # openai_compat 模板的 for 循环消费）——干跑缺它会误报 UndefinedError 'extra': {'temperature': 0.7}, 'params': params, 'api_key': 'sk-dry-run', 'org_id': '0', 'request': None, 'json': json, 'b64media2url': _stub, 'downloadfile2url': _stub} try: env = Environment(undefined=StrictUndefined) out = env.from_string(req_tpl).render(**ns) json.loads(out) except Exception as e: errors.append("%s渲染失败：%s: %s" % (tag, type(e).__name__, str(e)[:120])) if errors: return ("请求模板干跑渲染失败（落库前拦截）：%s——模板引用了运行时不存在的" "变量（业务参数可用：%s）" % ("；".join(errors), sorted(params_arr.keys()) or '无')) return "" async def _ensure_step_profile(sor, protocol, cap, vmid, step, spec): """异步后续步骤的适配模板（query/download）。 2026-09-05 改造（用户：同类任务查询接口供应商级相同，可复用）： 幂等键=名称「{protocol}-{purpose}-{path}」（去掉模型维度）——同供应商同协议 下多个模型共享同一份查询模板；旧的 {vmid}-{purpose} 命名兼容复用， 但若旧模板仍是骨架（__from_doc__）同样弃用重建。 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 归一化（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 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): recs = await sor.sqlExe( "SELECT id, request_template, response_template FROM llm_api_profile " "WHERE name=${n}$ AND status='active' LIMIT 1", {"n": cand}) await sor.sqlExe("COMMIT", {}) if not recs: continue old_tpl = (getattr(recs[0], "request_template", "") or "") + \ (getattr(recs[0], "response_template", "") or "") if not any(mk in old_tpl for mk in _SKELETON_MARKERS): return getattr(recs[0], "id", "") pid_old = getattr(recs[0], "id", "") await sor.sqlExe("UPDATE llm_api_profile SET status='deprecated' " "WHERE id=${i}$", {"i": pid_old}) await sor.sqlExe("COMMIT", {}) # 旧骨架弃用——继续往下重建（不 return） headers = _headers_from_spec(spec) # method 独立列（2026-09-06 用户定夺）：不再塞 request_template JSON if method == "GET": req = "" else: req = json.dumps({"data": {"task_id": "{task_id}"}}, ensure_ascii=False) # 查询步骤的响应解析在运行时由提交模板的 response_template 接管 # （inference._async_inference 轮询后直接用提交模板渲染），这里存档即可 resp = json.dumps({ "note": "查询步骤模板：运行时响应解析走模型提交模板的 response_template", "response_format": step.get("response_format", "") or "", }, ensure_ascii=False) pid = getID() await sor.C("llm_api_profile", { "id": pid, "name": shared_name, "protocol": protocol, "capability": cap, "path": path, "method": method, "headers": json.dumps(headers, ensure_ascii=False), "request_template": req, "response_template": resp, "param_schema": "", "status": "active"}) return pid # ────────────────────── 工具 5：定价自动导入 ────────────────────── _FACTOR_LABELS = { 'duration': '时长', 'flat': '按次', 'prompt_tokens': '输入tokens', 'completion_tokens': '输出tokens', 'uncache_tokens': '非缓存输入tokens', 'cached_tokens': '缓存tokens', } # 缓存拆分因子（derived 自 usage 原始字段，定价引擎自动计算，见 pricing-data-format 技能） _DERIVED_FACTOR_SPECS = { 'uncache_tokens': { 'type': 'int', 'role': 'factor', 'label': '非缓存输入Token', 'derived': 'prompt_tokens - prompt_tokens_details.cached_tokens'}, 'cached_tokens': { 'type': 'int', 'role': 'factor', 'label': '缓存Token', 'derived': 'prompt_tokens_details.cached_tokens'}, } # token 因子族：一个 token 模型的定价天然多因子（输入/输出/缓存同在一个方案）。 # 旧校验「同模型仅允许单一因子」与生产事实矛盾（qwen3.8-max 三因子同方案正常出账）， # token 模型按此族合并为一个定价方案（2026-09-06 修复）。 _TOKEN_FACTORS = ('prompt_tokens', 'completion_tokens', 'uncache_tokens', 'cached_tokens') # 同一因子多条价格且无维度区分 → 引擎会重复计价，必须拆方案（真混用的判据） def _check_factor_dup(mitems, factors): seen = set() for it in mitems: f = it.get("factor") or "flat" dims = json.dumps(it.get("dimensions") or {}, ensure_ascii=False, sort_keys=True) k = (f, dims) if k in seen: return ("因子 %s 出现多条相同维度的价格（引擎会重复计价）——请核对文档，" "同因子多价必须用维度区分档位" % f) seen.add(k) return None def _build_pricing_yaml(currency_items, dimensions_all, factors): """生成定价 YAML（2026-09-05 用户定夺的新模式；2026-09-06 支持多因子）： - 不放 model 过滤：一个定价方案只服务一个模型；定价相同的多个模型共享 同一 ppid（不是往定价里加 model 维度） - 不用 filters 子结构：维度直接平铺在定价项（引擎对非保留键做 AND 匹配） - fields 需定义每个维度（role: filter），否则引擎报「在fields中没有定义」 - factors 为列表：token 模型天然多因子（输入/输出/缓存同在一个方案， 生产实证 qwen3.8-max），每条 pricing 用各自的 factor；derived 因子 （uncache_tokens/cached_tokens）必须在 fields 定义 derived 表达式， 引擎计算衍生字段依赖此定义（pricing-data-format 技能 §2） """ fields = { 'price_factors': {'type': 'string', 'role': 'factor', 'label': '计价因子'}, 'unit_prices': {'type': 'float', 'role': 'factor', 'label': '单位定价'}, 'unit': {'type': 'string', 'role': 'factor', 'label': '计价单位'}, } for factor in factors: if factor in fields: continue if factor in _DERIVED_FACTOR_SPECS: fields[factor] = dict(_DERIVED_FACTOR_SPECS[factor]) else: fields[factor] = {'type': 'float', 'role': 'factor', 'label': _FACTOR_LABELS.get(factor, factor)} for dim in sorted(dimensions_all): fields[dim] = {'type': 'string', 'role': 'filter', 'label': dim} pricings = [] for it in currency_items: item = {'price_factors': it.get('factor') or 'flat', 'unit_prices': it['unit_price'], 'unit': it['unit']} for dk, dv in sorted((it.get('dimensions') or {}).items()): item[dk] = dv pricings.append(item) units = sorted(set(it['unit'] for it in currency_items)) unit_values = {u: 1 for u in units} if '百万' in unit_values: unit_values['百万'] = 1000000 if '千' in unit_values: unit_values['千'] = 1000 doc = {'unit_values': unit_values, 'fields': fields, 'pricings': pricings} return yaml.dump(doc, allow_unicode=True, sort_keys=False) async def _h_apply_model_pricing(sor, params, ctx): """定价自动导入：按提取规格建定价方案并挂模型 ppid（幂等）。 幂等键：模型已有 ppid 且方案描述含同一出处 URL → 更新时序（拉链）； 否则新建方案。价格只来自文档原文（提取层强制 doc_quote）。 规格来源（2026-09-05 防编造改造）：use_last_extract=true → 取本会话 extract 锚定规格（含 pricing/doc_url），禁止 LLM 复制转述大 JSON。 """ err = await _require_owner(sor, ctx) if err: return err use_last = str(params.get("use_last_extract") or "").lower() in ("true", "1", "yes") if use_last: spec = _spec_load(ctx) if spec is None: return ("本会话没有锚定的提取规格——先调 extract_llm_api_spec（成功后规格自动" "锚定本会话，含 pricing 与 doc_url），再传 {\"use_last_extract\": true}。" "会话级隔离：其他会话的规格本会话不可见。") else: try: spec = json.loads(params.get("spec") or "{}") except Exception: return "spec 不是合法 JSON" if not isinstance(spec, dict) or "pricing" not in spec: return ("spec 结构不符——不要自己编写规格。正确做法：extract_llm_api_spec 后传 " "{\"use_last_extract\": true}（pricing 与 doc_url 已锚定本会话）。") pricing = spec.get("pricing") or {} items = pricing.get("items") or [] if not items: return "spec.pricing.items 为空——文档没提取到价格（禁止编造），如需定价请先核对文档" doc_url = (spec.get("doc_url") or "").strip() currency = (pricing.get("currency") or "CNY").strip() or "CNY" # 校验：每条必须有 doc_quote（出处）；dimensions 禁止 model for it in items: if not (it.get("doc_quote") or "").strip(): return "定价条目缺 doc_quote（文档原文定价句）——定价只允许来自文档原文，拒绝落库" if "model" in (it.get("dimensions") or {}): return "定价 dimensions 含 model——一个定价方案只服务一个模型，禁止 model 维度；定价相同的模型共享同一 ppid" # 按 vendor_model_id 分组（一个模型一个定价方案） by_model = {} for it in items: vmid = (it.get("vendor_model_id") or "").strip() if not vmid: return "定价条目缺 vendor_model_id" by_model.setdefault(vmid, []).append(it) results = [] for vmid, mitems in by_model.items(): # 找模型 recs = await sor.sqlExe( "SELECT id, name, ppid FROM llm_model WHERE vendor_model_id=${m}$ " "AND status='active' LIMIT 1", {"m": vmid}) await sor.sqlExe("COMMIT", {}) if not recs: results.append({"model": vmid, "ok": False, "error": "模型未注册（先 apply_llm_config）"}) continue model_id = getattr(recs[0], "id", "") old_ppid = getattr(recs[0], "ppid", "") or "" factors = sorted(set((it.get("factor") or "flat") for it in mitems)) # 因子族校验（2026-09-06 修复）：token 模型天然多因子（输入/输出/缓存 # 同在一个方案，生产实证 qwen3.8-max ppid=TNJnx_Nw1HGDOtDIFji-w 三因子 # 正常出账）——旧「单因子」铁律把所有 token 定价判死，工具校验与生产 # 事实矛盾。真混用的判据改为两条： # 1. token 族与非 token 族（duration/flat 等）不可混在一个方案 # 2. 同一因子多条价格且维度相同 → 引擎重复计价 non_token = sorted(set(factors) - set(_TOKEN_FACTORS)) token_hit = sorted(set(factors) & set(_TOKEN_FACTORS)) if non_token and token_hit: results.append({"model": vmid, "ok": False, "error": "token 因子 %s 与按量因子 %s 混在一个模型——" "计费单位不同，请拆分为多个定价方案" % (token_hit, non_token)}) continue if len(non_token) > 1: results.append({"model": vmid, "ok": False, "error": "同一模型混用多个按量计价因子 %s——请拆分为多个定价方案" % non_token}) continue # 缓存拆分与整体输入价互斥：prompt_tokens 计全部输入，uncache_tokens 只计 # 非缓存部分——同方案并存会对缓存命中 tokens 重复计价（2026-09-06） if 'prompt_tokens' in factors and ('uncache_tokens' in factors or 'cached_tokens' in factors): results.append({"model": vmid, "ok": False, "error": "prompt_tokens 与 uncache_tokens/cached_tokens 同方案——" "缓存命中部分会被重复计价。文档区分缓存价时只用 " "uncache_tokens+cached_tokens+completion_tokens；" "不区分时只用 prompt_tokens+completion_tokens"}) continue dup_err = _check_factor_dup(mitems, factors) if dup_err: results.append({"model": vmid, "ok": False, "error": dup_err}) continue factor = ",".join(factors) # 方案命名/回报展示用 dims_all = set() for it in mitems: dims_all.update((it.get("dimensions") or {}).keys()) yaml_str = _build_pricing_yaml(mitems, dims_all, factors) # 引擎试算护栏：渲染一遍确认合法 YAML + fields 完整（不落库先验证） try: parsed = yaml.safe_load(yaml_str) assert parsed.get('pricings') and parsed.get('fields') except Exception as e: results.append({"model": vmid, "ok": False, "error": "定价 YAML 生成校验失败：%s" % str(e)[:150]}) continue desc_quotes = "；".join((it.get("doc_quote") or "")[:80] for it in mitems)[:500] pp_desc = "定价出处：%s | 文档原文：%s" % (doc_url or "(未提供)", desc_quotes) now = time.strftime('%Y-%m-%d') if old_ppid: ppid = old_ppid # 幂等核心（2026-09-05 用户纠正：时序应只有一条有效行，重复执行不得堆历史）： # 先取当前生效行比对内容—— # 内容相同 → 跳过（不产生新时序行） # 内容不同且生效行今天才启用 → 原地 UPDATE（同日替换无历史区间可保留， # 拉链会产生 enabled==expired 的零宽死行，纯垃圾） # 内容不同且生效行是历史日期 → 正常拉链（关旧行+插新行） recs2 = await sor.sqlExe( "SELECT id, pricing_data, enabled_date FROM pricing_program_timing " "WHERE ppid=${p}$ AND expired_date='9999-12-31' " "ORDER BY enabled_date DESC LIMIT 1", {"p": ppid}) await sor.sqlExe("COMMIT", {}) cur = recs2[0] if recs2 else None same = False if cur is not None: try: same = (yaml.safe_load(getattr(cur, 'pricing_data', '') or '{}') == yaml.safe_load(yaml_str)) except Exception: same = (getattr(cur, 'pricing_data', '') or '') == yaml_str await sor.sqlExe( "UPDATE pricing_program SET description=${d}$ WHERE id=${i}$", {"d": pp_desc[:1000], "i": ppid}) await sor.sqlExe("COMMIT", {}) if same: results.append({"model": vmid, "ok": True, "ppid": ppid, "action": "无变化（幂等跳过，未产生新时序行）", "items": len(mitems), "factor": factor, "dimensions": sorted(dims_all)}) continue if cur is not None and str(getattr(cur, 'enabled_date', ''))[:10] == now: await sor.sqlExe( "UPDATE pricing_program_timing SET pricing_data=${y}$, name=${n}$ " "WHERE id=${i}$", {"y": yaml_str, "n": "%s %s计价" % (vmid, factor), "i": getattr(cur, 'id', '')}) await sor.sqlExe("COMMIT", {}) action = "原地更新（同日替换，不产生历史行）" results.append({"model": vmid, "ok": True, "ppid": ppid, "action": action, "items": len(mitems), "factor": factor, "dimensions": sorted(dims_all)}) continue # 历史生效行 → 拉链 await sor.sqlExe( "UPDATE pricing_program_timing SET expired_date=${d}$ " "WHERE ppid=${p}$ AND expired_date='9999-12-31'", {"d": now, "p": ppid}) action = "更新（拉链：旧行截至今日，新时序生效 %s）" % now else: ppid = getID() await sor.C("pricing_program", { "id": ppid, "name": "%s 定价" % vmid, "ownerid": "0", "providerid": "", "pricing_belong": "", "description": pp_desc[:1000], "currency": currency}) action = "新建" await sor.C("pricing_program_timing", { "id": getID(), "ppid": ppid, "name": "%s %s计价" % (vmid, factor), "pricing_data": yaml_str, "enabled_date": now, "expired_date": "9999-12-31"}) await sor.sqlExe("COMMIT", {}) # 挂模型 await sor.sqlExe("UPDATE llm_model SET ppid=${p}$ WHERE id=${i}$", {"p": ppid, "i": model_id}) await sor.sqlExe("COMMIT", {}) results.append({"model": vmid, "ok": True, "ppid": ppid, "action": action, "items": len(mitems), "factor": factor, "dimensions": sorted(dims_all)}) ok_n = sum(1 for r in results if r.get("ok")) return json.dumps({ "summary": "定价导入：%d/%d 个模型成功" % (ok_n, len(results)), "results": results, "convention": "一个定价方案只服务一个模型（无 model 维度/无 filters，维度平铺）；" "定价完全相同的模型共享同一 ppid", }, ensure_ascii=False) # ────────────────────── 工具 6：模型测试（真实调用） ────────────────────── async def _h_test_model_call(sor, params, ctx): """真实调用测试：走完整治理链（门禁→上游→结算），返回调用结果+流水ID。 参数：model_name 必填；prompt 文本提示（t2t）；业务参数放 params JSON （媒体统一三数组契约 image_files/audio_files/video_files——字符串或数组 均可；另 resolution/duration 等）；timeout 等待秒数（异步模型默认 600）。 前置：供应商账号已配 api_key（无 key 会报可行动错误，不静默）。 """ err = await _require_owner(sor, ctx) if err: return err model_name = (params.get("model_name") or "").strip() if not model_name: return "缺 model_name" prompt = (params.get("prompt") or "回复两个字：正常").strip() try: extra = json.loads(params.get("params") or "{}") except Exception: return "params 不是合法 JSON" timeout_s = int(_f(params.get("timeout"), 0)) or 600 try: from pipeline_llm.inference import chat_inference except Exception as e: return "推理模块未加载：%s" % str(e)[:100] payload = {"model": model_name, "messages": [{"role": "user", "content": prompt}]} payload.update(extra) if timeout_s: payload["_timeout"] = timeout_s task_ref = "agent-test:%s:%d" % (model_name[:20], int(time.time())) t0 = time.time() try: data = await chat_inference(ctx.get("org_id", "") or "0", ctx.get("user_id", ""), payload, model_name=model_name, task_ref=task_ref) except Exception as e: return json.dumps({ "ok": False, "model": model_name, "task_ref": task_ref, "elapsed_sec": round(time.time() - t0, 1), "error": str(e)[:400], "hint": "常见原因：账号无api_key/余额不足/端点超时/模板配置缺失——按错误消息处置", }, ensure_ascii=False) elapsed = round(time.time() - t0, 1) content = "" try: content = data["choices"][0]["message"]["content"] except Exception: content = str(data)[:200] out = { "ok": True, "model": model_name, "task_ref": task_ref, "elapsed_sec": elapsed, "content_head": str(content)[:200], "usage": data.get("usage") or {}, } if data.get("media"): out["media"] = data["media"] # 生成物本地持久 URL（已落地） if data.get("task_id"): out["task_id"] = data["task_id"] return json.dumps(out, ensure_ascii=False) # ────────────────────── 工具 7：记账检查 ────────────────────── async def _h_check_model_accounting(sor, params, ctx): """记账正确性检查（三态：created/accounted/failed）。 按 task_ref（测试调用返回的）或 model_name 查最近流水： 1. 流水存在性 + accounting_status 2. usages 计价因子完整性 3. accounted → charge 金额与定价引擎重算对比（独立复算，不信记账侧） 4. failed → 给出 note 原因与处置建议 5. created → 出账循环 60 秒一轮，提示等待或查 worker 进程 """ err = await _require_owner(sor, ctx) if err: return err task_ref = (params.get("task_ref") or "").strip() model_name = (params.get("model_name") or "").strip() if not task_ref and not model_name: return "需要 task_ref（测试调用返回值）或 model_name 之一" if task_ref: recs = await sor.sqlExe( "SELECT id, model_id, status, accounting_status, charge, cost, usages, " "ppid, note, created_at FROM llm_usage WHERE task_ref=${t}$ " "ORDER BY created_at DESC LIMIT 3", {"t": task_ref}) else: recs = await sor.sqlExe( "SELECT u.id, u.model_id, u.status, u.accounting_status, u.charge, u.cost, " "u.usages, u.ppid, u.note, u.created_at FROM llm_usage u " "JOIN llm_model m ON m.id=u.model_id " "WHERE m.name=${n}$ OR m.vendor_model_id=${n}$ " "ORDER BY u.created_at DESC LIMIT 3", {"n": model_name}) await sor.sqlExe("COMMIT", {}) if not recs: return "未找到流水（task_ref=%s model=%s）——调用可能没发生或没结算" % ( task_ref, model_name) out_rows = [] for r in recs: row = dict(r) st = row.get("accounting_status", "") item = { "usage_id": row.get("id", ""), "call_status": row.get("status", ""), "accounting_status": st, "charge": float(row.get("charge") or 0), "usages": row.get("usages", "") or "", "ppid": row.get("ppid", "") or "", "created_at": str(row.get("created_at", "")), } if st == "failed": item["verdict"] = "记账失败" item["reason"] = (row.get("note") or "")[:200] item["hint"] = ("常见处置：模型未挂ppid→apply_model_pricing；" "模型未映射产品→apply_llm_config 已自动映射（2026-09-08），" "仍报未映射说明自动映射失败（看 apply 返回 product_mapping.note，" "多为产品模块未加载），重跑一次 apply 或告知用户；" "定价无匹配档位→核对usages维度值与定价YAML是否一致") elif st == "created": # 2026-09-06 用户定夺：只有 status='SUCCEEDED' 的行才写 accounting_status， # 非 SUCCEEDED 行为 NULL（不再挂 created 冒充待记账）—— # 走到这里必然是 SUCCEEDED 成功调用的真待记账行 item["verdict"] = "待记账（出账循环60秒一轮，稍后复查；持续不变查记账worker进程）" elif not st: # accounting_status=NULL：无记账意义（调用失败行/充值对账行） call_st = row.get("status", "") if call_st == "recharge": item["verdict"] = "充值对账流水（非计费用量，accounting_status=NULL）" else: item["verdict"] = "调用未成功行（status=%s，不计费，accounting_status=NULL）" % call_st item["reason"] = (row.get("note") or "")[:200] elif st == "accounted": # 独立复算：用 usages 因子过定价引擎，对比 charge item["verdict"] = "已记账" try: usage_data = json.loads(row.get("usages") or "{}") except Exception: usage_data = {} ppid = row.get("ppid") or "" if ppid and usage_data: try: from ahserver.serverenv import ServerEnv env = ServerEnv() fn = getattr(env, "buffered_charging", None) if fn: prices = await fn(ppid, usage_data) expect = round(sum(float(getattr(p, "amount", 0) or 0) for p in (prices or [])), 6) item["recomputed_amount"] = expect item["amount_match"] = abs(expect - float(row.get("charge") or 0)) < 0.01 if not item["amount_match"]: item["verdict"] = "已记账但金额不符（复算 %.4f ≠ charge %.4f）" % ( expect, float(row.get("charge") or 0)) except Exception as e: item["recompute_error"] = str(e)[:150] else: item["verdict"] = "已记账（缺ppid或usages，无法独立复算）" out_rows.append(item) return json.dumps({"rows": out_rows}, ensure_ascii=False) # ────────────────────── 工具 8：模块/应用信息查询 ────────────────────── async def _h_platform_modules(sor, params, ctx): """列出平台已装载的业务模块（内部 agent 了解平台构成用）。""" err = await _require_owner(sor, ctx) if err: return err mods = [] for name, title in [ ("pipeline_core", "产线核心（会话/技能/项目）"), ("pipeline_service", "执行引擎（任务链/角色/治理）"), ("pipeline-llm", "模型治理（供应商/账号/模型/定价/记账）"), ("pipeline-bidding", "投标产线"), ("pipeline-opportunity", "商机产线"), ("pipeline-sdlc", "开发产线前端"), ("pipeline-ops", "运维"), ("pipeline-task", "任务中心"), ]: try: __import__(name.replace("-", "_")) loaded = True except Exception: loaded = False mods.append({"module": name, "title": title, "loaded": loaded}) return json.dumps(mods, ensure_ascii=False) # ────────────────────── 工具 9：只读数据库查询 ────────────────────── # 根因（2026-09-07 记账失败排查实录）：agent 想核实库内数据（产品映射/记账流水）时， # 工具集里没有 DB 查询动作，只能拿 run_command 手搓连库脚本——撞上沙箱四重摩擦： # ① bwrap 的 /tmp 是会话私有 tmpfs，脚本不跨命令持久（写完下一条就 No such file） # ② /d/pipeline 只读 bind，应用目录写不进 # ③ run_command 输出双重截断（stdout[:2000] + tool_result[:500]），无截断提示 # ④ 框架 API 逐个踩坑（initEnv 签名/sqlor % 格式化/SHOW 语句返回 int） # 十余轮试错查一条数据。补只读查询工具根治：SELECT 白名单 + ${k}$ 强制参数化 + # 行数/字符上限显式截断 + SELECT 后 COMMIT（防 MDL 锁，已知坑）。 _DB_QUERY_MAX_ROWS = 200 _DB_QUERY_MAX_CHARS = 8000 _DB_QUERY_MAX_TABLES = 10 def _validate_readonly_sql(sql: str) -> str: """只读校验：仅单条 SELECT。返回 ''=通过，否则可行动的错误信息。""" s = re.sub(r"--[^\n]*", " ", sql or "") # 去行注释（防注释夹带） s = re.sub(r"/\*.*?\*/", " ", s, flags=re.S) # 去块注释 s = " ".join(s.split()).strip() if not s: return "SQL 为空" if ";" in s.rstrip(";"): return "只允许单条语句（禁止分号分隔多条）" head = s.lstrip("(").lstrip().split(" ", 1)[0].lower() if head != "select": return ("只允许 SELECT 查询（收到 %s）。SHOW/DESC 不支持——查表结构用 tables 参数" "（返回列定义），或 SELECT information_schema.COLUMNS" % head) low = s.lower() for bad in ("into outfile", "into dumpfile", "load_file(", "for update", "lock in share mode"): if bad in low: return "禁止 %s 子句（只读工具）" % bad # 裸 % 会被 aiomysql 当格式占位符 → "not enough arguments for format string" # （内部助手实录踩坑）。LIKE 通配符必须走 ${k}$ 参数化由 params 传值。 if "%" in s: return ("SQL 里禁止直接写 %（会被数据库驱动误当格式占位符报 " "not enough arguments）。用 ${key}$ 占位符 + params 传值：" "如 LIKE ${k}$ 配 params {\"k\": \"%xxx%\"}") return "" async def _h_db_query(sor, params, ctx): """只读数据库查询（SELECT 白名单 / tables 表结构捷径）。 - sql + params：单条 SELECT，占位符 ${k}$ 参数化，禁止裸 % - tables：逗号分隔表名，直接返回各表列定义（免手写 information_schema SQL） - 行数上限 200、输出上限 8000 字符，超限显式告知（禁止静默截断） - SELECT 后 COMMIT 释放 MDL 锁（已知坑） """ err = await _require_owner(sor, ctx) if err: return err sql = (params.get("sql") or "").strip() tables = (params.get("tables") or "").strip() ns = params.get("params") or {} if isinstance(ns, str): try: ns = json.loads(ns) except (json.JSONDecodeError, ValueError): return "params 必须是 JSON 对象，如 {\"k\": \"%xxx%\"}" if not isinstance(ns, dict): return "params 必须是 JSON 对象（键对应 SQL 里的 ${键}$ 占位符）" # 防御：sqlor.sqlExe 见 ns 含 'page' 键会走分页路径返回 DictObject(total,rows) # 而非行列表——占位符命名撞 'page' 时静默变形，这里剔除。 ns.pop("page", None) # ── 分支 1：tables 表结构捷径 ── if tables and not sql: tlist = [t.strip() for t in tables.split(",") if t.strip()][:_DB_QUERY_MAX_TABLES] out = {} try: for t in tlist: # 注意：sqlor 把返回列名统一转小写（sor.py _get_data）， # 故用 AS 别名取小写键，不能 getattr(r, "COLUMN_NAME")。 recs = await sor.sqlExe( "SELECT COLUMN_NAME AS col_name, COLUMN_TYPE AS col_type, " "IS_NULLABLE AS col_null, COLUMN_KEY AS col_key, " "COLUMN_DEFAULT AS col_default FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=${t}$ " "ORDER BY ORDINAL_POSITION", {"t": t}) await sor.sqlExe("COMMIT", {}) out[t] = [{"name": getattr(r, "col_name", ""), "type": getattr(r, "col_type", ""), "null": getattr(r, "col_null", ""), "key": getattr(r, "col_key", ""), "default": getattr(r, "col_default", None)} for r in (recs or [])] except Exception as e: return "查表结构失败: %s" % str(e)[:300] empty = [t for t in tlist if not out.get(t)] note = ("表不存在或无列: %s" % ",".join(empty)) if empty else "" text = json.dumps({"tables": out}, ensure_ascii=False, default=str) if note: text += "\n[⚠️ %s]" % note return text if not sql: return "需要 sql（SELECT 语句）或 tables（逗号分隔表名，查列定义）之一" # ── 分支 2：SELECT 查询 ── verr = _validate_readonly_sql(sql) if verr: return verr try: recs = await sor.sqlExe(sql, ns) await sor.sqlExe("COMMIT", {}) except Exception as e: msg = str(e)[:300] hint = "" if "Unknown column" in msg or "1054" in msg: hint = "（列名不存在：先用 tables 参数查该表的列定义再改 SQL）" elif "Unknown table" in msg or "1146" in msg: hint = "（表不存在：核对表名，如 llm_usage/product/llm_model）" elif "not enough arguments" in msg: hint = "（SQL 含裸 % 或占位符缺值：LIKE 等用 ${k}$ + params 传值）" return "SQL 执行失败: %s%s" % (msg, hint) # sqlor 返回 DictObject（dict 子类，__setattr__ 写入键而非 __dict__）—— # 必须 dict(r) 转换，不能用 r.__dict__（永远为空）。同 check_model_accounting 先例。 rows = [dict(r) for r in (recs or [])] total = len(rows) note = "" if total > _DB_QUERY_MAX_ROWS: rows = rows[:_DB_QUERY_MAX_ROWS] note = ("⚠️ 结果共 %d 行，只返回前 %d 行——请加 WHERE/LIMIT 收窄" % (total, _DB_QUERY_MAX_ROWS)) text = json.dumps({"total": total, "rows": rows}, ensure_ascii=False, default=str) while rows and len(text) > _DB_QUERY_MAX_CHARS: rows.pop() text = json.dumps({"total": total, "rows": rows}, ensure_ascii=False, default=str) note = ("⚠️ 输出超 %d 字符上限，已截断到前 %d 行（共 %d 行）——请加 WHERE/LIMIT 收窄" % (_DB_QUERY_MAX_CHARS, len(rows), total)) if note: text += "\n[%s]" % note return text # ────────────────────── 工具定义 ────────────────────── PLATFORM_TOOLS = [ ToolDefinition( name="platform_llm_status", description="查看模型治理状态：供应商/账号余额/模型定价/各状态分布。用户问「模型配置状态/账号余额」时调用。仅owner组织角色可用。", parameters={}, category="platform", ), ToolDefinition( name="fetch_model_doc", description="抓取大模型供应商的 API 文档页面（返回纯文本）。配置新模型前先用此抓取官方文档。仅owner组织角色可用。", parameters={"url": "文档页面 URL（必须是公网 http/https）"}, category="platform", ), ToolDefinition( name="extract_llm_api_spec", description="通读文档文本，LLM 提取 API 配置规格（端点/协议/请求响应格式/定价）。配合 fetch_model_doc 使用。仅owner组织角色可用。", parameters={"doc_text": "fetch_model_doc 返回的文档文本"}, category="platform", ), ToolDefinition( name="apply_llm_config", description="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型。幂等。推荐 use_last_extract=true 取本会话锚定规格（禁止把规格复制进 spec——复制即编造）；需改动加 overrides（vendor_name/base_url/protocol/doc_url/doc_notes/model_capability——用户纠正能力分类时传 model_capability 而非重提取）。仅owner组织角色可用。", parameters={ "use_last_extract": "true=用本会话 extract_llm_api_spec 锚定的规格（推荐，会话级隔离）", "overrides": "覆盖项 JSON（白名单：vendor_name/base_url/protocol/doc_url/doc_notes/model_capability。model_capability={模型ID:能力} 如 {\\\"wan3.0-video-prime\\\": \\\"r2v\\\"}——用户纠正能力分类时传它+use_last_extract 重新 apply，不要重提取；模板会按新能力重建）", "spec": "（仅无锚定时的兜底）extract 返回的 JSON 规格原文", }, category="platform", # 交互原则（2026-09-05 用户定）：配置链工具不设确认门——缺信息才问， # 否则一口气做完给测试结果，用户发现问题再按说明修改。 ), ToolDefinition( name="apply_model_pricing", description="定价自动导入：建定价方案（pricing_program+时序YAML）并挂模型 ppid。幂等（内容无变化跳过；同日原地更新；历史行才拉链——时序表始终只一条有效行）。一个定价方案只服务一个模型；定价相同的模型共享同一 ppid。推荐 use_last_extract=true 取本会话锚定规格（含 pricing/doc_url），禁止复制转述。仅owner组织角色可用。", parameters={ "use_last_extract": "true=用本会话锚定规格的 pricing+doc_url（推荐）", "spec": "（仅无锚定时的兜底）extract 返回的 JSON 规格原文", }, category="platform", ), ToolDefinition( name="test_model_call", description="模型真实调用测试：走完整治理链（门禁→上游→结算）。返回 ok/content/usage/media（生成物本地URL）/task_ref。前置：供应商账号已配 api_key。异步模型（视频等）等待至任务完成（默认600秒）。仅owner组织角色可用。", parameters={ "model_name": "模型名（llm_model.name 或 vendor_model_id）", "prompt": "文本提示词（t2t 默认「回复两个字：正常」；生成类填生成描述）", "params": "业务参数 JSON 字符串（生成类模型用；媒体统一三数组契约 " "image_files/audio_files/video_files，字符串或数组均可，如 " "{\\\"image_files\\\": [\\\"https://...\\\", \\\"https://...\\\"], " "\\\"resolution\\\": \\\"480P\\\", \\\"duration\\\": 5}）", "timeout": "等待秒数（默认600，上限900）", }, category="platform", ), ToolDefinition( name="check_model_accounting", description="记账正确性检查（三态 created/accounted/failed）：查最近流水的记账状态/usages因子/charge金额，accounted 时用定价引擎独立复算金额对比。failed 给原因与处置建议。仅owner组织角色可用。", parameters={ "task_ref": "test_model_call 返回的 task_ref（优先）", "model_name": "或按模型名查最近流水", }, category="platform", ), ToolDefinition( name="platform_modules", description="列出平台已装载的业务模块清单。用户问「平台有哪些模块」时调用。仅owner组织角色可用。", parameters={}, category="platform", ), ToolDefinition( name="db_query", description="只读数据库查询（排查/核实库内数据的唯一正道——禁止用 run_command 手搓连库脚本）。" "两种用法：①sql+params 执行单条 SELECT（占位符 ${k}$ 参数化，SQL 内禁写裸 %，" "LIKE 通配符放 params 传值如 {\"k\":\"%xxx%\"}）；②tables=逗号分隔表名，直接返回列定义" "（免查 information_schema，SQL 报列名/表名错误时先用它核对结构）。" "行数上限200、输出上限8000字符，超限会显式提示收窄。仅owner组织角色可用。", parameters={ "sql": "单条 SELECT 语句（占位符 ${k}$；禁裸 %、禁多语句、禁 INTO OUTFILE）", "params": "占位符取值 JSON 对象，如 {\"k\": \"%happyhorse%\", \"id\": \"abc123\"}", "tables": "（与 sql 二选一）逗号分隔表名，返回各表列定义，如 llm_usage,product", }, category="platform", ), ] PLATFORM_PROMPT = """ 你是产线平台的内部运维 agent，服务对象是 owner 组织的角色。 ## 模型自动配置全链工作流（用户给文档 URL 或正文要求配置模型时，一口气独立完成） 0. **输入形态判定**：用户给的是 URL → 走 fetch_model_doc；用户直接粘贴了文档正文 （无 URL 或 URL 抓不到）→ **跳过 fetch_model_doc，把正文原样作为 doc_text 传给 extract_llm_api_spec**（禁止凭记忆改写正文）。用户单独给了定价链接/正文时： 定价链接可再 fetch_model_doc 抓取，定价正文直接并入 doc_text 一起提取。 1. **先 `load_skill` 加载 `model-auto-config`**（含媒体转换铁律/能力分类先行/定价新模式），严格执行 2. `fetch_model_doc` 抓取官方文档（返回文本末尾带 [出处URL] 行） 3. `extract_llm_api_spec` 通读文档提取规格——**成功后规格自动锚定到本会话缓存** （会话级隔离），返回里有 __anchored__ 提示 4. **不要停下来找用户确认**——直接进入下一步落库。只有文档里确实缺失、 且无法从上下文合理推断的信息（如 api_key、能力分类需新增）才向用户提问。 5. `apply_llm_config` 传 **{"use_last_extract": true}**（用户要求改名等只加 overrides 白名单覆盖项，如 {"overrides": {"vendor_name": "阿里云百炼"}}）。 **绝对禁止把规格 JSON 复制/转述进 spec 参数——转述必编造结构（历史教训： 三轮全编造 vendor{}/api_profile{} 等不存在的字段）。** 6. `apply_model_pricing` 同样传 **{"use_last_extract": true}**（定价+出处已锚定） 7. `test_model_call` 真实调用测试（前置：账号已配 api_key，没配则如实告知用户去补， 不要伪造结果）——生成类模型传 params（媒体三数组 image_files/audio_files/ video_files + resolution/duration 等） 8. `check_model_accounting` 检查记账（传上一步的 task_ref）： - accounted 且 amount_match=true → 全链通过，汇报金额 - created → 等 60~120 秒再查一次（出账循环 60 秒一轮） - failed → 按 reason/hint 处置；需要核实库内数据（产品映射/定价挂接/流水明细） 一律用 `db_query` 只读查询，**禁止用 run_command 手搓连库脚本**（沙箱 /tmp 不跨 命令持久、平台目录只读、输出双截断——手搓必陷多轮试错）。 产品映射由 apply_llm_config 自动完成（2026-09-08 起）——若仍报「未映射到产品」， 回看 apply 返回的 product_mapping.note（多为产品模块未加载），重跑一次 apply_llm_config（幂等）触发自动映射，仍失败才如实告知用户。 9. 汇报：配置/定价/测试/记账四段结果 + 出处 URL + 遗留事项（如模板需人工核对项）。 用户发现问题后按其说明修改，再重新走对应环节（全链幂等，重跑安全）。 ## 上线检查工作流（用户选定模型要求「上线检查」时，按模型配置后的测试过程处理） 1. `test_model_call` 对指定模型真实调用（完整治理链：门禁→上游→结算）。 生成类模型（t2i/t2v/i2v/r2v 等）缺必要业务参数（媒体输入/分辨率等）时**先问用户**， 禁止编造参数或跳过测试；t2t 用默认提示词即可。 2. `check_model_accounting` 传上一步 task_ref 检查记账： accounted 且 amount_match → 记账通过；created → 等 60~120 秒复查一次再下结论； failed → 按 reason/hint 报告原因与处置建议，需核实库内数据用 `db_query` （如 `SELECT ... FROM product WHERE resource_ref_id=${id}$` 查产品映射是否存在）， 禁止 run_command 手搓连库脚本。 3. `platform_llm_status` 复核该模型/定价/供应商当前状态（active/ppid 挂接等）。 4. **最后明确告诉用户检查结果：通过 / 不通过** + 各环节明细（调用/记账/状态） + 遗留事项。结论必须如实——任何一环失败都是不通过，禁止粉饰。 ## 交互原则（用户定，2026-09-05） - **只有缺失信息需要用户补充时才交互**（api_key、业务参数、需新增能力分类等） - 否则从抓取到记账检查**一口气做完**，最后给出完整测试结果 - **不要每步找用户确认**——落库/定价/测试都直接执行（全部幂等，改起来安全） - 用户发现问题 → 按用户说明修改重跑，这才是正确的纠错循环 ## 工具报错自救规则 - 报「spec 结构不符/缺少 xxx」→ **不要改写结构重试**，改用 use_last_extract=true； 本会话没锚定就重新走 fetch→extract（锚定是会话级的，跨会话不可见） - 报「能力分类未登记」→ 如实转告用户需先加能力分类（这是人工评审门，不要绕过） - 报「media_audit/runtime_note/需人工核对」警告 → 原样转告用户，不假装完成 ## 硬规则 - **定价只允许来自文档原文**（doc_quote 强制），禁止编造/换算/推导 - **出处必留**：模型注册表描述字段与定价方案描述字段都必须保存文档 URL - 能力分类不存在 → 先加分类（字典/种子/提示词/端点注释四处同步）再配模型，禁止塞近似分类 - 能力分类铁律：按文档正文的输入/输出模态声明判定，禁止照调用示例判（示例常只演示 最简形态——wan3.0 实测「四模态全能参考」被示例判成 t2v）。extract 返回 __capability_alert__ 说明描述与能力矛盾，apply 会自动纠正；用户口头纠正能力时， 传 overrides={"model_capability": {"模型ID": "能力"}} + use_last_extract 重新 apply， 禁止重提取（同样的提示词只会同样误判） - 生成类模型媒体铁律：上行统一三数组契约 image_files/audio_files/video_files （字符串或数组均可）经 b64media2url 转公网 URL；下行产物用 downloadfile2url(request, url) 落地；apply 返回的 media_audit 警告必须如实转告 - **模板渲染报错（如 'dict object' has no attribute 'xxx_file'）是机制缺陷， 不是要用户去页面手改模板**：如实报告错误原文 + 你判断的根因，由维护者修生成器； 禁止提出「请你在模型治理页面把模板改成…」这类人工兜底方案，禁止把模板 JSON 抄给用户 - 非标协议骨架模板需人工核对的项（__note__）必须如实转告用户，不假装完成 - 文档抓取失败/内容不足时如实说明，不要凭记忆编造 API 格式 - 测试没有 api_key 就停在配置阶段如实汇报，禁止跳过测试谎称完成 - 所有工具仅 owner 组织角色可用，权限报错时如实转告用户 ## 平台知识 平台模块清单用 `platform_modules` 查询；技能库中「平台」相关技能（skills_library/all/ 与 pipelines/platform_general/）包含各模块的开发规范与踩坑记录，需要时用 load_skill 加载。 """ PLATFORM_HANDLERS = { "platform_llm_status": _h_platform_llm_status, "fetch_model_doc": _h_fetch_model_doc, "extract_llm_api_spec": _h_extract_llm_api_spec, "apply_llm_config": _h_apply_llm_config, "apply_model_pricing": _h_apply_model_pricing, "test_model_call": _h_test_model_call, "check_model_accounting": _h_check_model_accounting, "platform_modules": _h_platform_modules, "db_query": _h_db_query, } def register_platform_ability(): """注册平台内部 agent 能力包（幂等）。""" ability = PipelineAbility( pipeline_id=PLATFORM_PIPELINE_ID, name="平台内部 agent", tools=PLATFORM_TOOLS, system_prompt=PLATFORM_PROMPT, handlers=PLATFORM_HANDLERS, roles=[], menus=[ {"label": "📊 模型治理", "icon": "", "url": "/pipeline-llm", "type": "tab"}, ], ) register_ability(ability) return ability register_platform_ability()