707 lines
33 KiB
Python
707 lines
33 KiB
Python
"""
|
||
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
|
||
|
||
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")
|
||
|
||
|
||
# ────────────────────── 权限门禁（代码层） ──────────────────────
|
||
|
||
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)[^>]*>.*?</\1>", " ", 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(("<!doctype", "<html")):
|
||
text = _html_to_text(raw)
|
||
else:
|
||
text = raw
|
||
if len(text) > _MAX_DOC_CHARS:
|
||
text = text[:_MAX_DOC_CHARS] + "\n[文档过长已截断，共 %d 字符]" % len(text)
|
||
return text
|
||
|
||
|
||
# ────────────────────── 工具 3：提取配置规格（LLM） ──────────────────────
|
||
|
||
_EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型 API 文档，提取配置规格。
|
||
|
||
只输出一个 JSON 对象（不要 markdown 代码块），字段：
|
||
{
|
||
"vendor_name": "供应商名称",
|
||
"base_url": "API 基础地址（如 https://dashscope.aliyuncs.com/compatible-mode/v1）",
|
||
"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_fields": ["请求体字段名列表"],
|
||
"response_format": "响应格式说明（content 字段路径 + usage 字段路径）",
|
||
"async_steps": [{"purpose": "query|download", "path": "该步骤接口路径", "method": "GET|POST", "request_fields": ["请求字段名列表"], "response_format": "该步骤响应格式说明"}],
|
||
"models": [{"vendor_model_id": "供应商侧模型ID", "capability": "t2t|t2i|i2t|t2v|i2v|embedding|rerank|tts|asr",
|
||
"sync_mode": "sync|async（提交后需轮询查询结果的填 async）",
|
||
"description": "一句话说明"}],
|
||
"doc_notes": "文档中影响配置的关键注意点"
|
||
}
|
||
|
||
异步模型规则：
|
||
- 文档描述「先提交任务、再轮询查询结果」的模型，sync_mode 填 async，并提取 async_steps：
|
||
至少一条 purpose=query（查询任务状态/结果）；若文档另有独立下载/取文件接口，再加一条 purpose=download。
|
||
- 同步模型 async_steps 填空数组 []。
|
||
|
||
定价规则：
|
||
- 价格按文档原文记录，并保留文档标注的单位（在 description 中写明，如「元/百万tokens」）
|
||
- 禁止自行换算、推导或统一单位——单位是否吻合由上线测试阶段的记账核对验证
|
||
- 文档没写价格的模型：价格字段填 null（禁止编造价格），在 description 注明
|
||
- cost_* 未知时与 price_* 相同（无供应商折扣时成本=售价）
|
||
|
||
文档内容：
|
||
"""
|
||
|
||
|
||
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]
|
||
return json.dumps(spec, ensure_ascii=False)
|
||
|
||
|
||
# ────────────────────── 工具 4：写入配置（幂等） ──────────────────────
|
||
|
||
def _f(v, default=0.0):
|
||
try:
|
||
return float(v)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
async def _h_apply_llm_config(sor, params, ctx):
|
||
"""按提取的规格写库：供应商（复用/新建）+ 适配模板 + 模型（含定价）。
|
||
|
||
幂等：供应商按名称复用；模型按 (vendor, vendor_model_id) 复用——
|
||
已存在则更新定价，不重复创建。
|
||
"""
|
||
err = await _require_owner(sor, ctx)
|
||
if err:
|
||
return err
|
||
try:
|
||
spec = json.loads(params.get("spec") or "{}")
|
||
except Exception:
|
||
return "spec 不是合法 JSON"
|
||
vendor_name = (spec.get("vendor_name") or "").strip()
|
||
if not vendor_name:
|
||
return "spec 缺少 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-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. 供应商：按名称复用
|
||
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 = []
|
||
for ep in endpoints:
|
||
bu = (ep.get("base_url") or "").rstrip("/")
|
||
if bu and bu not in [(e.get("base_url") or "").rstrip("/") for e in merged]:
|
||
merged.append({"base_url": bu, "region": ep.get("region") or "domestic",
|
||
"timeout": int(_f(ep.get("timeout"), 60))})
|
||
added_eps.append(bu)
|
||
# 2026-09-05：供应商表不再有 protocol——请求形态由模型挂的适配模板决定
|
||
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 个）" % (vendor_id, len(added_eps))
|
||
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))} for ep in endpoints]
|
||
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)
|
||
|
||
# 2. 适配模板：按 (协议×能力) 复用，缺则新建
|
||
profile_ids = {}
|
||
caps = sorted(set((m.get("capability") or "t2t") for m in models))
|
||
for cap in caps:
|
||
recs = await sor.sqlExe(
|
||
"SELECT id FROM llm_api_profile WHERE protocol=${p}$ AND capability=${c}$ "
|
||
"AND status='active' ORDER BY created_at DESC LIMIT 1",
|
||
{"p": protocol, "c": cap})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if recs:
|
||
profile_ids[cap] = getattr(recs[0], "id", "")
|
||
continue
|
||
req_tpl = _default_templates(protocol, cap, spec)
|
||
pid = getID()
|
||
await sor.C("llm_api_profile", {
|
||
"id": pid, "name": "%s-%s-自动配置" % (protocol, cap),
|
||
"protocol": protocol, "capability": cap,
|
||
"path": req_tpl["path"],
|
||
"headers": json.dumps(req_tpl["headers"], ensure_ascii=False),
|
||
"request_template": req_tpl["req"], "response_template": req_tpl["resp"],
|
||
"param_schema": json.dumps([{"name": "prompt", "label": "提示词",
|
||
"uitype": "textarea", "required": True}],
|
||
ensure_ascii=False),
|
||
"status": "active"})
|
||
profile_ids[cap] = pid
|
||
|
||
# 3. 模型：按 (vendor, vendor_model_id) 幂等
|
||
created, updated, skipped = [], [], []
|
||
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()
|
||
# 异步模型：后续步骤模板链（提交后顺序执行 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 ""
|
||
recs = await sor.sqlExe(
|
||
"SELECT id FROM llm_model WHERE vendor_id=${v}$ AND vendor_model_id=${m}$",
|
||
{"v": vendor_id, "m": vmid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if recs:
|
||
mid = getattr(recs[0], "id", "")
|
||
await sor.sqlExe(
|
||
"UPDATE llm_model SET description=${d}$, "
|
||
"sync_mode=${sm}$, query_profile_ids=${q}$, updated_at=NOW() "
|
||
"WHERE id=${i}$",
|
||
{"d": desc, "sm": sync_mode, "q": query_ids_json, "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)))
|
||
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,
|
||
"runtime_note": rt_note,
|
||
"media_audit": audit_note,
|
||
}, ensure_ascii=False)
|
||
|
||
|
||
# 生成类能力：出参是文件（图/视频/音频/3D），response 必须 downloadfile2url 落地
|
||
_MEDIA_CAPS = ('t2i', 'i2v', 't2v', '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',
|
||
'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 ""
|
||
|
||
|
||
def _default_templates(protocol: str, capability: str, spec: dict) -> dict:
|
||
"""生成适配模板默认值（path/headers 独立列 + data/response 模板）。
|
||
|
||
OpenAI 兼容协议用平台标准模板；其他协议按文档格式生成骨架供人工微调。
|
||
返回 {"path", "headers", "req", "resp"}。
|
||
"""
|
||
chat_path = (spec.get("chat_path") or "/chat/completions").strip()
|
||
headers = {"Authorization": "Bearer {{api_key}}",
|
||
"Content-Type": "application/json"}
|
||
if protocol == "openai_compat":
|
||
req = json.dumps({
|
||
"data": {"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}
|
||
# 非标准协议：骨架模板（字段路径来自文档提取），标注需人工核对。
|
||
# 生成类能力（_MEDIA_CAPS）骨架直接带媒体转换约定占位——按用户铁律：
|
||
# 上行 b64media2url（上传转公网URL）、下行 downloadfile2url（生成物落地）。
|
||
req = json.dumps({
|
||
"data": {"__from_doc__": spec.get("request_fields") or []},
|
||
"__note__": "非 openai_compat 协议骨架模板，需按文档人工核对",
|
||
}, ensure_ascii=False)
|
||
resp_body = {"content": spec.get("response_format", "") or "待按文档填写",
|
||
"__note__": "需人工核对"}
|
||
if capability in _MEDIA_CAPS:
|
||
# 生成类：骨架给出落地写法提示，人工只需把 <产物url字段> 换成真实路径
|
||
resp_body = {
|
||
"status": "SUCCEEDED",
|
||
"video": "{{ downloadfile2url(request, output.video_url) }}",
|
||
"__note__": "生成类能力：产物 URL 必须用 downloadfile2url 落地为本地"
|
||
"持久 URL（上游仅 24 小时有效）；上传媒体在 request 模板用 "
|
||
"b64media2url(request, xxx_file)；产物字段名按文档核对",
|
||
}
|
||
req = json.dumps({
|
||
"data": {"__from_doc__": spec.get("request_fields") or [],
|
||
"__upload__": "{{ b64media2url(request, params.xxx_file) }}"},
|
||
"__note__": "上传媒体一律 xxx_file 命名 + b64media2url 转公网 URL",
|
||
}, ensure_ascii=False)
|
||
resp = json.dumps(resp_body, ensure_ascii=False)
|
||
return {"path": chat_path, "headers": headers, "req": req, "resp": resp}
|
||
|
||
|
||
async def _ensure_step_profile(sor, protocol, cap, vmid, step, spec):
|
||
"""异步模型的后续步骤模板（query/download）：按名幂等创建/复用 profile。
|
||
|
||
step: {"purpose": "query|download", "path", "method", "request_fields", "response_format"}
|
||
返回 profile id。
|
||
"""
|
||
purpose = (step.get("purpose") or "query").strip() or "query"
|
||
pname = "%s-%s" % (vmid, purpose)
|
||
recs = await sor.sqlExe(
|
||
"SELECT id FROM llm_api_profile WHERE name=${n}$", {"n": pname})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if recs:
|
||
return getattr(recs[0], "id", "")
|
||
headers = {"Authorization": "Bearer {{api_key}}",
|
||
"Content-Type": "application/json"}
|
||
method = (step.get("method") or "POST").strip().upper() or "POST"
|
||
req = json.dumps({
|
||
"method": method,
|
||
"data": {"__from_doc__": step.get("request_fields") or [],
|
||
"__task_id__": "{{task_id}}"},
|
||
"__note__": "异步%s步骤骨架，需按文档人工核对" % purpose,
|
||
}, ensure_ascii=False)
|
||
resp = json.dumps({
|
||
"content": step.get("response_format", "") or "待按文档填写",
|
||
"__note__": "需人工核对",
|
||
}, ensure_ascii=False)
|
||
pid = getID()
|
||
await sor.C("llm_api_profile", {
|
||
"id": pid, "name": pname, "protocol": protocol, "capability": cap,
|
||
"path": (step.get("path") or "").strip(),
|
||
"headers": json.dumps(headers, ensure_ascii=False),
|
||
"request_template": req, "response_template": resp,
|
||
"param_schema": "", "status": "active"})
|
||
return pid
|
||
|
||
|
||
# ────────────────────── 工具 5：模块/应用信息查询 ──────────────────────
|
||
|
||
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="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型（含定价）。幂等——已有模型只更新定价。仅owner组织角色可用。",
|
||
parameters={"spec": "extract_llm_api_spec 返回的 JSON 规格"},
|
||
category="platform",
|
||
requires_confirmation=True,
|
||
),
|
||
ToolDefinition(
|
||
name="platform_modules",
|
||
description="列出平台已装载的业务模块清单。用户问「平台有哪些模块」时调用。仅owner组织角色可用。",
|
||
parameters={},
|
||
category="platform",
|
||
),
|
||
]
|
||
|
||
PLATFORM_PROMPT = """
|
||
你是产线平台的内部运维 agent，服务对象是 owner 组织的角色。
|
||
|
||
## 模型自动配置工作流（用户给你文档 URL 要求配置模型时）
|
||
1. **先 `load_skill` 加载 `model-onboarding`**（配置→测试→上线三步标准流程）和
|
||
`auto-api-pricing-config`（模板库），按技能严格执行
|
||
2. `fetch_model_doc` 抓取官方文档页面
|
||
3. `extract_llm_api_spec` 通读文档提取配置规格——**定价只允许来自文档原文**，文档没写的价格一律 null，禁止编造
|
||
4. 把提取结果摘要给用户确认（尤其定价和端点），用户确认后
|
||
5. `apply_llm_config` 写入（幂等：已有模型只更新定价）
|
||
6. 进入测试/上线阶段时，严格按 model-onboarding 技能的判据执行（无 apikey 先补录，测试必查调用成功+记账四项）
|
||
|
||
## 硬规则
|
||
- **定价按文档原文记录**，单位照文档所写如实标注（写入 description），禁止自行换算、推导或统一单位；单位是否吻合由测试阶段的记账核对验证
|
||
- cost_*（供应商成本）未知时与 price_* 相同
|
||
- 非 openai_compat 协议的模型：模板会存档但运行时暂不可调用，必须如实告知
|
||
- 文档抓取失败/内容不足时如实说明，不要凭记忆编造 API 格式
|
||
- 所有工具仅 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,
|
||
"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()
|