"""
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 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")
# ────────────────────── 会话级规格缓存（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
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 基础地址——必须取文档示例请求 URL 中除接口路径外的完整前缀（含 /api/v1 或 /compatible-mode/v1 等路径段，如 https://dashscope.aliyuncs.com/api/v1）。禁止只填裸域（如 https://dashscope.aliyuncs.com）——裸域拼接口路径必 404",
"protocol": "openai_compat | dashscope_async | custom（能走 /chat/completions 的填 openai_compat）",
"endpoints": [{"base_url": "...", "region": "domestic|international", "timeout": 60}],
"chat_path": "对话接口路径（如 /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|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（计价因子：视频按秒=duration，按次=flat，token计价=对应token因子）",
"unit_price": 0.45,
"unit": "秒|次|百万（文档标注的计价单位名）",
"dimensions": {"resolution": "480P（影响价格档位的维度，按文档原文值；禁止放 model）"},
"doc_quote": "文档原文定价句（逐字照抄，标明出处用）"}]
},
"doc_url": "从文档文本末尾的 [出处URL] 行逐字照抄（配置出处，必填）",
"doc_notes": "文档中影响配置的关键注意点"
}
异步模型规则：
- 文档描述「先提交任务、再轮询查询结果」的模型，sync_mode 填 async，并提取 async_steps：
至少一条 purpose=query（查询任务状态/结果）；若文档另有独立下载/取文件接口，再加一条 purpose=download。
- 任务查询接口通常是**供应商级共用**的（如 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。
- 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 教训）：提取完当场校验描述模态声明 vs
# capability，矛盾信息直接回给助手（用户纠正时传 overrides.model_capability 重 apply）
cap_alerts = []
for _m in (spec.get("models") or []):
_fix, _warn = _capability_guard(_m.get("capability", ""), _m.get("description", ""))
if _warn:
cap_alerts.append("%s: %s" % (_m.get("vendor_model_id", "?"), _warn))
if cap_alerts:
out["__capability_alert__"] = ("能力分类与文档描述矛盾（按示例误判的典型信号），"
"apply 时会自动纠正，也可用 overrides.model_capability "
"人工指定：" + "；".join(cap_alerts))
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 = ('参考生', '全能参考', '参考图', '参考视频', '参考音频', '四模态', '多模态参考')
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)]
input_kinds = [k for k in kinds if k != 'video']
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
async def _h_apply_llm_config(sor, params, ctx):
"""按提取的规格写库：供应商（复用/新建）+ 适配模板 + 模型。
幂等：供应商按名称复用；模型按 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 二次教训）：必须在 caps 计算/模板
# 生成之前跑——上一版放在模型循环里，纠正 t2v→r2v 时模板已按 t2v 生成，
# profile_ids 无 r2v 键，模型更新的 IF 空值保护把旧 t2v 模板原样保留
# （实测：template_notes 有纠正记录、库里 cap 仍是 t2v）。
cap_notes = []
for m in models:
_fix, _warn = _capability_guard(
m.get("capability") or "", m.get("description") or "")
if _fix:
m["capability"] = _fix
if _warn:
cap_notes.append("%s: %s" % (m.get("vendor_model_id") or "?", _warn))
# 能力分类校验（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-05 404 实测教训）：
# 同一供应商可同时有 openai_compat(compatible-mode/v1) 与原生异步(api/v1)
# 两类端点，账号绑定/运行时选择必须按协议匹配，否则原生模型被拼到
# compatible-mode 前缀下 → 404。另拦截「裸域退化」：提取 LLM 偶发把
# base_url 提成 https://host（丢了 /api/v1），与已有带路径端点同 host
# 时该裸域无意义，跳过不入目录。
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 = []
merged = list(eps_old)
added_eps = []
skipped_eps = []
for ep in endpoints:
bu = (ep.get("base_url") or "").rstrip("/")
if not bu:
continue
# 裸域退化拦截（新增与回填打标两条路径都要拦，2026-09-05 e2e 实测：
# 提取 LLM 偶发把 base_url 提成裸域，回填打标会让裸域变「协议匹配」
# 端点被运行时选中 → 拼出 https://host/services/... → 404）
pu = _urlparse(bu)
if not pu.path.strip("/"):
same_host = any(_urlparse((e.get("base_url") or "")).netloc == pu.netloc
and _urlparse((e.get("base_url") or "")).path.strip("/")
for e in merged)
if same_host:
skipped_eps.append("%s（裸域退化：同 host 已有带路径端点）" % bu)
continue
# 同 base_url 已存在：只补协议标签（旧端点无 protocol 时回填）
hit = None
for e in merged:
if (e.get("base_url") or "").rstrip("/") == bu:
hit = e
break
if hit is not None:
if protocol and not hit.get("protocol"):
hit["protocol"] = protocol
continue
merged.append({"base_url": bu, "region": ep.get("region") or "domestic",
"timeout": int(_f(ep.get("timeout"), 60)),
"protocol": protocol})
added_eps.append(bu)
await sor.sqlExe(
"UPDATE llm_vendor SET endpoints=${e}$, updated_at=NOW() WHERE id=${i}$",
{"e": json.dumps(merged, ensure_ascii=False), "i": vendor_id})
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
else:
vendor_id = getID()
eps_norm = [{"base_url": (ep.get("base_url") or "").rstrip("/"),
"region": ep.get("region") or "domestic",
"timeout": int(_f(ep.get("timeout"), 60)),
"protocol": protocol} for ep in endpoints
if (ep.get("base_url") or "").strip()]
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. 适配模板：按 (协议×能力) 复用；旧骨架模板（硬编码 xxx_file/__from_doc__，
# 2026-09-05 实测 t2v 因它渲染崩）视为不可用，按文档示例自愈重建
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:
# 名称过滤必须加（2026-09-05 e2e 实测 bug）：查询/下载步骤模板同库同协议
# 同能力，不加过滤会被「最新优先」误选为提交模板 → 提交打到查询 path → 404
# 两种命名都要排（2026-09-06 i2v 实测漏网）：
# 共享命名 {protocol}-{purpose}-{path} → 含中缀 -query-
# legacy 命名 {vmid}-{purpose} → 以 -query 结尾（无后置短横）
# 只排中缀时 legacy 查询模板会被当成提交模板挂到模型上（schema 为空
# 过不了体检门禁也不报错——空 schema 无 hard 违规，静默误挂）
recs = await sor.sqlExe(
"SELECT id, request_template, response_template, param_schema "
"FROM llm_api_profile "
"WHERE protocol=${p}$ AND capability=${c}$ AND status='active' "
"AND name NOT LIKE ${l1}$ AND name NOT LIKE ${l2}$ "
"AND name NOT LIKE ${l3}$ AND name NOT LIKE ${l4}$ "
"ORDER BY created_at DESC LIMIT 1",
{"p": protocol, "c": cap,
"l1": "%-query-%", "l2": "%-download-%",
"l3": "%-query", "l4": "%-download"})
await sor.sqlExe("COMMIT", {})
if recs:
pid_old = getattr(recs[0], "id", "")
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))
if not sv['hard'] and not any(mk in old_tpl for mk in _SKELETON_MARKERS):
profile_ids[cap] = pid_old
continue
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'])))
else:
tpl_notes.append("旧骨架模板 %s（%s）含硬编码占位符，已弃用并按文档示例重建"
% (pid_old, cap))
tpl = _gen_templates(protocol, cap, spec)
dry_err = _dry_render_check(tpl["req"], cap, tpl["biz_params"], tpl["media_params"])
if dry_err:
# 宁可不建，不埋雷：渲染不了的模板落库=运行时必崩，如实报错由助手修正规格重跑
tpl_errors.append("%s: %s" % (cap, dry_err))
continue
pid = getID()
await sor.C("llm_api_profile", {
"id": pid, "name": "%s-%s-自动配置" % (protocol, cap),
"protocol": protocol, "capability": cap,
"path": tpl["path"],
"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-05 404 教训）：账号只绑 compatible-mode
# 端点时，原生协议模型运行时会被选到错误端点 → 404。自动补绑本协议
# 端点到该供应商全部活跃账号（不删已有绑定），如实报告。
account_notes = []
proto_idx = [i for i, e in enumerate(vendor_endpoints or [])
if (e.get("protocol") or "") == protocol]
if proto_idx:
recs = await sor.sqlExe(
"SELECT id, name, endpoint_ids FROM llm_account "
"WHERE vendor_id=${v}$ AND status='active'", {"v": vendor_id})
await sor.sqlExe("COMMIT", {})
for acc in recs or []:
try:
bound = [int(b) for b in json.loads(getattr(acc, "endpoint_ids", "") or "[]")]
except Exception:
bound = []
if any(i in proto_idx for i in bound):
continue # 已绑本协议端点，可用
newb = sorted(set(bound) | set(proto_idx))
await sor.sqlExe(
"UPDATE llm_account SET endpoint_ids=${e}$, updated_at=NOW() WHERE id=${i}$",
{"e": json.dumps(newb), "i": getattr(acc, "id", "")})
await sor.sqlExe("COMMIT", {})
account_notes.append("账号「%s」原绑定端点 %s 不含协议 %s 的端点（运行时会 404），"
"已自动补绑为 %s" % (getattr(acc, "name", ""), bound,
protocol, newb))
audit_note = await _audit_media_convention(sor, list(profile_ids.values()))
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,
"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:
issues.append("「%s」request 模板含媒体上传参数但缺 b64media2url——"
"上传文件必须转本地公网 URL 再传上游" % name)
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', '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_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 _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 = {}
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)
return _put(expr)
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:
out[k] = _put('{{ params.%s|default(%s)|tojson }}'
% (k, json.dumps(v, ensure_ascii=False)))
return out
body = walk(tree)
s = json.dumps(body, ensure_ascii=False)
for k, expr in tokens.items():
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 = ''
if not field:
field = _MEDIA_OUT_FIELD.get(outkey, 'result_url')
note = ('产物字段名「%s」按供应商惯例兜底（文档未提供响应示例或示例中无 URL 字段），'
'如上游字段不同需修正' % field)
resp = json.dumps({
'status': 'SUCCEEDED',
outkey: '{{ downloadfile2url(request, %s) }}' % field,
'usage': '{{ json.dumps(usage) }}',
'task_id': '{{ task_id }}',
}, ensure_ascii=False)
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 提交必须带异步开关头（文档示例遗漏也不至于提交即失败）
if str(spec.get('protocol') or '').strip() == 'dashscope_async':
headers.setdefault('X-DashScope-Async', 'enable')
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}。
"""
chat_path = (spec.get("chat_path") or "/chat/completions").strip()
headers = _headers_from_spec(spec)
notes = []
if protocol == "openai_compat":
req = json.dumps({
"model": "{{model}}", "messages": "{{messages}}",
"temperature": "{{temperature}}", "stream": False,
}, ensure_ascii=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 字段路径生成，"
"业务参数无默认值（调用时必须显式传入），建议人工核对")
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:
pos = m.end()
req = req[:pos] + '"media": %s, ' % expr + 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 家族契约）——"
"建议对照供应商 API 参考页人工核对")
except Exception as _e:
notes.append("r2v 媒体合成失败（%s），示例未见上传字段，请人工核对" % str(_e)[:80])
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}
# 旧骨架模板标记：含这些标记的 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 _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'), 'required': True,
# 三数组契约：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)):
ns = {'model': 'dry-run-model', 'prompt': '干跑校验', 'messages': [],
'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 归一化（对齐运行时契约，实测踩过三个坑）：
- 任务号占位符用**单花括号** {task_id}（运行时 str.replace 消费，
Jinja 双花括号不会被替换）；文档给 {{task_id}} 也归一为单花括号
- 剥掉 base_url 已有的路径前缀（提取给 /api/v1/tasks/{task_id}，
运行时 url=base_url+path，base_url 已含 /api/v1 → 不归一会双前缀 404）
- 缺占位符时按 DashScope 惯例补 /tasks/{task_id}
返回 profile id。
"""
purpose = (step.get("purpose") or "query").strip() or "query"
path = (step.get("path") or "").strip()
method = (step.get("method") or ("GET" if purpose == "query" else "POST")).strip().upper()
# path 归一化
from urllib.parse import urlparse
bu_path = urlparse((spec.get("base_url") or "").strip()).path.rstrip("/")
if bu_path and path.startswith(bu_path + "/"):
path = path[len(bu_path):]
path = path.replace("{{task_id}}", "{task_id}")
if "{task_id}" not in path and purpose == "query":
path = (path.rstrip("/") + "/{task_id}") if path else "/tasks/{task_id}"
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)
if method == "GET":
req = json.dumps({"method": method}, ensure_ascii=False)
else:
req = json.dumps({"method": method,
"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,
"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',
}
def _build_pricing_yaml(currency_items, dimensions_all, factor):
"""生成定价 YAML（2026-09-05 用户定夺的新模式）：
- 不放 model 过滤：一个定价方案只服务一个模型；定价相同的多个模型共享
同一 ppid（不是往定价里加 model 维度）
- 不用 filters 子结构：维度直接平铺在定价项（引擎对非保留键做 AND 匹配）
- fields 需定义每个维度（role: filter），否则引擎报「在fields中没有定义」
"""
fields = {
'price_factors': {'type': 'string', 'role': 'factor', 'label': '计价因子'},
'unit_prices': {'type': 'float', 'role': 'factor', 'label': '单位定价'},
'unit': {'type': 'string', 'role': 'factor', 'label': '计价单位'},
}
if factor not in fields:
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': factor, '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))
if len(factors) != 1:
results.append({"model": vmid, "ok": False,
"error": "同一模型混用多个计价因子 %s——请拆分为多个定价方案" % factors})
continue
factor = factors[0]
dims_all = set()
for it in mitems:
dims_all.update((it.get("dimensions") or {}).keys())
yaml_str = _build_pricing_yaml(mitems, dims_all, factor)
# 引擎试算护栏：渲染一遍确认合法 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；"
"模型未映射产品→产品管理导入产线模型；"
"定价无匹配档位→核对usages维度值与定价YAML是否一致")
elif st == "created":
# m0011 设计：accounting_status 只对 status='ok' 行有意义——
# failed 调用行/recharge 充值行挂 created 是被出账循环天然排除（不计费），
# 不是「待记账」（2026-09-06 实测：助手把 21 条 failed 行误报成待记账）
call_st = row.get("status", "")
if call_st == "recharge":
item["verdict"] = "充值对账流水（非计费用量，不参与出账）"
elif call_st != "ok":
item["verdict"] = "调用失败行（不计费，出账循环天然排除）"
item["reason"] = (row.get("note") or "")[:200]
else:
item["verdict"] = "待记账（出账循环60秒一轮，稍后复查；持续不变查记账worker进程）"
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-dist", "分销"),
("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)
# ────────────────────── 工具定义 ──────────────────────
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",
),
]
PLATFORM_PROMPT = """
你是产线平台的内部运维 agent，服务对象是 owner 组织的角色。
## 模型自动配置全链工作流（用户给文档 URL 要求配置模型时，一口气独立完成）
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 处置（未映射产品→告知用户在产品管理导入产线模型）
9. 汇报：配置/定价/测试/记账四段结果 + 出处 URL + 遗留事项（如模板需人工核对项）。
用户发现问题后按其说明修改，再重新走对应环节（全链幂等，重跑安全）。
## 交互原则（用户定，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,
}
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()