diff --git a/pipeline_platform/platform_ability.py b/pipeline_platform/platform_ability.py index 2d70c69..561b4ad 100644 --- a/pipeline_platform/platform_ability.py +++ b/pipeline_platform/platform_ability.py @@ -19,6 +19,9 @@ import asyncio import json import logging import re +import time + +import yaml from appPublic.uniqueID import getID @@ -208,6 +211,8 @@ async def _h_fetch_model_doc(sor, params, ctx): text = raw if len(text) > _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 @@ -229,6 +234,16 @@ _EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型 "models": [{"vendor_model_id": "供应商侧模型ID", "capability": "t2t|t2i|i2t|t2v|i2v|embedding|rerank|tts|asr", "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": "文档中影响配置的关键注意点" } @@ -238,10 +253,12 @@ _EXTRACT_PROMPT = """你是大模型 API 配置专家。通读下面这份模型 - 同步模型 async_steps 填空数组 []。 定价规则： -- 价格按文档原文记录，并保留文档标注的单位（在 description 中写明，如「元/百万tokens」） -- 禁止自行换算、推导或统一单位——单位是否吻合由上线测试阶段的记账核对验证 -- 文档没写价格的模型：价格字段填 null（禁止编造价格），在 description 注明 -- cost_* 未知时与 price_* 相同（无供应商折扣时成本=售价） +- pricing.items 每条必须带 doc_quote（文档原文定价句逐字照抄）——定价只允许来自文档原文 +- 价格/单位按文档原文记录，禁止自行换算、推导或统一单位 +- 同一模型多档价格（如不同分辨率）拆成多条 item，各带自己的 dimensions 与 doc_quote +- dimensions 禁止放 model——一个定价方案只服务一个模型；定价完全相同的多个模型 + 共享同一个定价方案（同一 ppid），不是往定价里加 model 过滤 +- 文档没写价格：items 填空数组 []（禁止编造价格） 文档内容： """ @@ -399,6 +416,7 @@ async def _h_apply_llm_config(sor, params, ctx): profile_ids[cap] = pid # 3. 模型：按 (vendor, vendor_model_id) 幂等 + doc_url = (spec.get("doc_url") or "").strip() created, updated, skipped = [], [], [] for m in models: vmid = (m.get("vendor_model_id") or "").strip() @@ -408,6 +426,9 @@ async def _h_apply_llm_config(sor, params, ctx): 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": @@ -588,7 +609,305 @@ async def _ensure_step_profile(sor, protocol, cap, vmid, step, spec): return pid -# ────────────────────── 工具 5：模块/应用信息查询 ────────────────────── +# ────────────────────── 工具 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）。 + """ + err = await _require_owner(sor, ctx) + if err: + return err + try: + spec = json.loads(params.get("spec") or "{}") + except Exception: + return "spec 不是合法 JSON" + 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: + # 幂等更新：新时序行（拉链——旧行 expired_date 截到今天） + ppid = old_ppid + await sor.sqlExe( + "UPDATE pricing_program_timing SET expired_date=${d}$ " + "WHERE ppid=${p}$ AND expired_date='9999-12-31'", + {"d": now, "p": ppid}) + await sor.sqlExe( + "UPDATE pricing_program SET description=${d}$ WHERE id=${i}$", + {"d": pp_desc[:1000], "i": 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_file/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": + 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 了解平台构成用）。""" @@ -639,11 +958,39 @@ PLATFORM_TOOLS = [ ), ToolDefinition( name="apply_llm_config", - description="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型（含定价）。幂等——已有模型只更新定价。仅owner组织角色可用。", + description="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型。幂等——已有模型只更新。定价不在此工具（用 apply_model_pricing）。仅owner组织角色可用。", parameters={"spec": "extract_llm_api_spec 返回的 JSON 规格"}, category="platform", requires_confirmation=True, ), + ToolDefinition( + name="apply_model_pricing", + description="定价自动导入：按提取规格的 pricing 建定价方案（pricing_program+时序YAML）并挂模型 ppid。幂等（已有 ppid 拉链更新时序）。一个定价方案只服务一个模型；定价相同的模型共享同一 ppid。价格必须来自文档原文（每条带 doc_quote 否则拒绝）。仅owner组织角色可用。", + parameters={"spec": "extract_llm_api_spec 返回的 JSON 规格（含 pricing 与 doc_url）"}, + category="platform", + requires_confirmation=True, + ), + 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_file\": \"https://...\", \"resolution\": \"480P\", \"duration\": 5}）", + "timeout": "等待秒数（默认600，上限900）", + }, + category="platform", + requires_confirmation=True, + ), + 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组织角色可用。", @@ -655,20 +1002,32 @@ PLATFORM_TOOLS = [ 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 先补录，测试必查调用成功+记账四项） +## 模型自动配置全链工作流（用户给文档 URL 要求配置模型时，按序独立完成） +1. **先 `load_skill` 加载 `model-auto-config`**（含媒体转换铁律/能力分类先行/定价新模式），严格执行 +2. `fetch_model_doc` 抓取官方文档（返回文本末尾带 [出处URL] 行） +3. `extract_llm_api_spec` 通读文档提取规格：端点/协议/能力/sync_mode/async_steps， + 以及 **pricing（每条价格带 doc_quote 文档原文）和 doc_url（照抄出处行）** +4. 把提取摘要给用户确认（端点/能力/定价原文/出处），确认后： +5. `apply_llm_config` 落库供应商/适配模板/模型（模型描述自动带 [出处:URL]） +6. `apply_model_pricing` 建定价方案并挂 ppid（定价描述带出处URL+文档原文； + 一个方案只服务一个模型，无 model 维度无 filters，维度平铺；幂等拉链更新） +7. `test_model_call` 真实调用测试（前置：账号已配 api_key，没配则如实告知用户去补， + 不要伪造结果）——生成类模型传 params（image_file/resolution/duration 等） +8. `check_model_accounting` 检查记账（传上一步的 task_ref）： + - accounted 且 amount_match=true → 全链通过，汇报金额 + - created → 等 60~120 秒再查一次（出账循环 60 秒一轮） + - failed → 按 reason/hint 处置（未映射产品→告知用户在产品管理导入产线模型） +9. 汇报：配置/定价/测试/记账四段结果 + 出处 URL + 遗留事项（如模板需人工核对项） ## 硬规则 -- **定价按文档原文记录**，单位照文档所写如实标注（写入 description），禁止自行换算、推导或统一单位；单位是否吻合由测试阶段的记账核对验证 -- cost_*（供应商成本）未知时与 price_* 相同 -- 非 openai_compat 协议的模型：模板会存档但运行时暂不可调用，必须如实告知 +- **定价只允许来自文档原文**（doc_quote 强制），禁止编造/换算/推导 +- **出处必留**：模型注册表描述字段与定价方案描述字段都必须保存文档 URL +- 能力分类不存在 → 先加分类（字典/种子/提示词/端点注释四处同步）再配模型，禁止塞近似分类 +- 生成类模型媒体铁律：上行 b64media2url(request, xxx_file)、下行 downloadfile2url(request, url)； + apply 返回的 media_audit 警告必须如实转告 +- 非标协议骨架模板需人工核对的项（__note__）必须如实转告用户，不假装完成 - 文档抓取失败/内容不足时如实说明，不要凭记忆编造 API 格式 +- 测试没有 api_key 就停在配置阶段如实汇报，禁止跳过测试谎称完成 - 所有工具仅 owner 组织角色可用，权限报错时如实转告用户 ## 平台知识 @@ -681,6 +1040,9 @@ PLATFORM_HANDLERS = { "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, }