pipeline-service/pipeline_service/platform_model_tools.py

490 lines
25 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""pipeline_service.platform_model_tools — 平台模型工具唯一实现2026-09-07
用户需求:通用助手/产线 agent 可调用平台 owner 机构 + 本机构的全部
pipeline-llm 注册模型,按任务与用户输入自动匹配合适模型完成任务。
- 候选可见性/owner 硬校验/自动选型pipeline_llm.selection
models_catalog / auto_select_model与推理治理链同一机构语义
- 实际调用llm_bridge.llm_infer → /pipeline-llm/api/v1/chat/completions
统一推理端点(门禁链 + 记账 + 同步/异步分流全在治理层,本层零旁路)
消费方(薄壳委托,禁止复制逻辑):
- agent_loop_v2.AgentExecutor._t_list_platform_models / _t_invoke_model
(会话 agent驾驶舱 + 通用助手)
- agent_loop._exec_agent_tool产线角色 agent v1
"""
import json
import logging
logger = logging.getLogger("pipeline.platform_model_tools")
_CAP_DESC = {
"t2t": "文本对话", "i2t": "图像理解", "m2t": "多媒体理解",
"t2i": "文生图", "i2v": "图生视频", "t2v": "文生视频",
"r2v": "参考生视频", "tts": "语音合成", "asr": "语音识别",
"embedding": "向量化", "rerank": "重排序",
}
# ─────────────────────────────────────────────────────────────────
# 完备性硬门禁2026-09-08 用户定夺一C+二A
#
# 用户原则:「意图识别后还要能识别用户输入是否有完成任务的完备条件,
# 如果不够,必须询问得到,不能做不好还做,浪费钱财」。
#
# 分工一C意图识别归主循环 LLMsystem prompt 引导,零额外调用);
# **完备性硬门禁归代码**——放在贵工具入口、上游调用之前。理由:完备性是
# 能力契约的确定性检查(该能力要求哪些输入、传没传),不是语义猜测,
# 代码判定可靠且不依赖模型自觉deepseek 类模型无视 prompt 指令有实测前科)。
#
# 范围二A只护花钱/不可逆动作(生成类调用真花钱)。闲聊/查询不设门禁。
#
# 返回 QUESTION: 前缀 → agent 循环把它当 ask_user 抛给用户(见
# agent_loop_v2 run loop / v1 的 ask 处理),不调上游、一分钱不花。
# ─────────────────────────────────────────────────────────────────
# 能力 → 必备输入契约(缺任一即不完备,必须先问用户要)
# 键名对应调用方 params 的三数组契约字段 / task 文本
_CAP_REQUIRED = {
"t2i": ("task",), # 文生图:必须有画面描述
"t2v": ("task",), # 文生视频:必须有画面描述
"tts": ("task",), # 语音合成:必须有要念的文本
"i2v": ("image_files",), # 图生视频:必须有输入图
"i2i": ("image_files",), # 图生图:必须有输入图
"asr": ("audio_files",), # 语音识别:必须有音频
"r2v": ("any_media",), # 参考生视频:至少一类参考媒体
}
_MEDIA_KEYS = ("image_files", "audio_files", "video_files")
# 必备输入的中文显示名(追问话术里给用户看,不用内部字段名)
_REQ_LABEL = {
"image_files": "输入图片",
"audio_files": "输入音频",
"video_files": "参考视频",
"any_media": "参考素材(图片/视频/音频任一)",
}
# task 类必备输入按能力的显示名(文生图要「画面描述」,语音合成要「朗读文本」)
_TASK_LABEL = {
"t2i": "画面描述", "t2v": "画面描述", "i2v": "运动/画面变化描述",
"tts": "朗读文本", "asr": "识别需求说明",
}
# 缺输入时对用户的追问话术(按能力,说清要什么、什么格式)
_CAP_ASK = {
"t2i": "生成图片需要先有画面描述。请说明:画什么主体、风格(写实/插画/商务示意图等)、"
"比例或尺寸、图中是否要有文字及内容。",
"t2v": "生成视频需要先有画面描述。请说明:画面内容、时长、分辨率、风格。",
"tts": "语音合成需要先有要朗读的文本。请提供文本内容,并说明音色/语速偏好(可选)。",
"i2v": "图生视频必须先提供输入图片(公网 URL 或上传)。另外请说明想要的运动/画面变化。",
"i2i": "图生图必须先提供输入图片(公网 URL 或上传)。另外请说明要怎么改(风格化/局部重绘等)。",
"asr": "语音识别必须先提供音频文件(公网 URL 或上传)。",
"r2v": "参考生视频必须先提供至少一类参考素材(图片/视频/音频,公网 URL 或上传),"
"并说明想要的画面内容与运动。",
}
def _biz_media(biz, key):
"""业务参数里的媒体数组是否非空(三数组契约,值可为字符串或数组)。"""
v = biz.get(key)
if isinstance(v, (list, tuple)):
return any(str(x or "").strip() for x in v)
return bool(str(v or "").strip())
def _gate_plan(cap_req, cap_model, biz):
"""能力降级门禁规划2026-09-08i2i→t2i / r2v→t2v
cap_req = 调用方显式指定的能力(空 = 自动选型,请求能力未知)。
返回 (gate_cap, use_code_table)
- 显式请求且 CAP_DEGRADE[cap_req]==cap_model如 t2i 选中 i2i 模型)且未带
模型能力必需媒体 → 契约=请求能力、代码表判定LLM/模型技能层会按模型
自身 i2i 契约误判,实测 2026-09-08 拦截合法文生图任务)
- 自动选型cap_req 空且模型是降级源i2i/r2v且未带媒体 → 任务实为
t2i/t2v 形态,契约=降级目标能力、代码表判定
- 带媒体输入 → 真 i2i/r2v 用法,契约=模型能力LLM 层
- 其余 → 契约=显式请求能力空则模型能力LLM 层
"""
try:
from pipeline_llm.selection import CAP_DEGRADE, CAP_DEGRADE_SOURCE
except ImportError:
return (cap_req or cap_model or ''), False
cap_req = (cap_req or '').strip().lower()
cap_model = (cap_model or '').strip().lower()
def _no_media():
for req in _CAP_REQUIRED.get(cap_model, ()):
if req == 'task':
continue
if req == 'any_media':
if not any(_biz_media(biz or {}, k) for k in _MEDIA_KEYS):
return True
elif not _biz_media(biz or {}, req):
return True
return False
if cap_req:
if CAP_DEGRADE.get(cap_req) == cap_model and _no_media():
return cap_req, True
return cap_req, False
src = CAP_DEGRADE_SOURCE.get(cap_model)
if src and _no_media():
return src, True
return (cap_model or ''), False
def _check_completeness_code(capability, task, biz):
"""能力契约代码表校验(**兜底层**2026-09-08 用户定夺后降级)。
判断主体是 LLMjudge_completeness本函数只在 LLM 裁决失败
(调用异常/输出不可解析)时兜底——花钱闸门不能因裁决故障失效。
返回 '' = 完备;否则返回追问话术。
"""
cap = (capability or "").strip().lower()
required = _CAP_REQUIRED.get(cap)
if not required:
return ""
missing = []
for req in required:
if req == "task":
if not (task or "").strip():
missing.append(_TASK_LABEL.get(cap, "画面/内容描述"))
elif req == "any_media":
if not any(_biz_media(biz or {}, k) for k in _MEDIA_KEYS):
missing.append(_REQ_LABEL["any_media"])
else:
if not _biz_media(biz or {}, req):
missing.append(_REQ_LABEL.get(req, req))
if missing:
ask = _CAP_ASK.get(cap, "该能力缺少必需输入,请补充:" + "".join(missing))
return ("调用「" + _CAP_DESC.get(cap, cap) + "」模型缺少必需输入("
+ "".join(missing) + "),已停在调用前、未产生任何费用。" + ask)
return ""
# ─────────────────────────────────────────────────────────────────
# 模型配套技能2026-09-08 用户定夺):
# 「每个模型的加入需要有配套的 skill模型需要什么样的输入、用户已经
# 提供的内容缺什么,应该由 LLM 判断而不是硬编码;没有 skill 的,按
# 能力的输入要求对照用户已给输入决策是否需要补充。」
#
# 判断分层(判断归 LLM契约归数据代码表只兜底
# ① 模型有配套技能skills/models/{model}/SKILL.md上线时自动生成
# → LLM 按技能里的输入契约 vs 用户已给内容判断缺什么
# ② 无技能 → LLM 按能力语义(能力字典 输入→输出模态vs 用户输入判断
# ③ LLM 裁决失败(异常/输出不可解析)→ _check_completeness_code 代码表
# 兜底——花钱闸门不能因裁决故障失效
# ─────────────────────────────────────────────────────────────────
_CAP_MODALS = {
"t2t": "文本→文本", "i2t": "图像→文本", "m2t": "多媒体→文本",
"t2i": "文本→图像", "i2i": "图像→图像", "t2v": "文本→视频",
"i2v": "图像→视频", "r2v": "参考媒体(图/视频/音频)→视频",
"tts": "文本→语音", "asr": "语音→文本",
"embedding": "文本→向量", "rerank": "候选集→排序",
}
def load_model_skill(model_name):
"""读模型配套技能正文skills/models/{model}/SKILL.md
查找顺序全局模板get_skills_base(),应用目录)→ 机构运行时目录
/d/pipeline/workspaces/{org}/skills 由各调用方 base_dir 决定,模型
技能是平台级资产放全局模板即可)。无技能返回 ''
目录约定skills/models/ 不在 SkillLoader 扫描范围global/pipelines/
orgs/users不会污染 agent 技能目录层。
"""
import os
try:
from pipeline_core.skill_pack import get_skills_base
base = get_skills_base()
except Exception:
base = os.path.join(os.getcwd(), "skills")
safe = "".join(ch for ch in (model_name or "")
if ch.isalnum() or ch in "-_.")
if not safe:
return ""
path = os.path.join(base, "models", safe, "SKILL.md")
try:
if os.path.isfile(path):
with open(path, encoding="utf-8") as f:
return f.read()[:4000]
except Exception as e:
logger.warning("load_model_skill(%s) 失败: %s", model_name, e)
return ""
async def judge_completeness(model_name, capability, task, biz,
description="", org_id="0", user_id="",
project_id=""):
"""完备性判断主体LLM 裁决2026-09-08 用户定夺)。
① 模型有配套技能 → 按技能输入契约判断;② 无技能 → 按能力语义判断;
③ LLM 失败 → 代码表兜底。
返回 '' = 完备可执行否则返回追问话术QUESTION 内容)。
"""
cap = (capability or "").strip().lower()
skill = load_model_skill(model_name)
if skill:
contract = ("该模型有配套技能(权威输入契约),按其「输入要求」判断:\n"
+ skill[:3000])
else:
modal = _CAP_MODALS.get(cap, "")
contract = ("该模型无配套技能,按能力语义判断。能力 %s%s%s"
% (cap or "?", modal or "模态未知",
"必备输入=箭头左侧模态的素材/文本;右侧是产物。"
if modal else "无法确定契约,宽松放行。"))
got = ["任务描述: " + (task or "(空)")]
for k in ("image_files", "audio_files", "video_files"):
if _biz_media(biz or {}, k):
v = biz.get(k)
n = len(v) if isinstance(v, (list, tuple)) else 1
got.append("已提供 %s: %d" % (k, n))
extra = {k: str(v)[:60] for k, v in (biz or {}).items()
if k not in ("image_files", "audio_files", "video_files")}
if extra:
got.append("其他业务参数: " + json.dumps(extra, ensure_ascii=False)[:300])
prompt = (
"你是平台模型调用的完备性裁决器。即将调用模型「%s」(能力 %s%s)完成生成任务,"
"调用会产生真实费用。请判断当前输入是否足以完成该任务。\n\n"
"%s\n\n模型描述:%s\n\n用户已提供的输入:\n%s\n\n"
"判断规则:\n"
"1. 只把「没有就无法完成本能力任务」的输入判为缺失(如文生图没有画面描述、"
"图生视频没有输入图)。\n"
"2. 有默认值/可选的偏好参数(尺寸/风格/时长/音色等)缺失**不算**不完备,不要拦。\n"
"3. 用户已提供的内容里隐含了该输入的(如任务描述里已含图片 URL算已提供。\n"
"4. 判断不了就放行complete=true宁可放行也别误拦。\n\n"
"只返回 JSON{\"complete\": true/false, \"ask\": \"缺失时向用户追问的话术"
"(说清缺什么、要什么格式、一句话),完备时空串\"}"
% (model_name, cap or "?",
("" + _CAP_DESC.get(cap, "")) if cap in _CAP_DESC else "",
contract, (description or "(无)")[:200], "\n".join(got)))
try:
from pipeline_service.llm_bridge import llm_call_msgs
raw = await llm_call_msgs(
[{"role": "user", "content": prompt}],
temperature=0, org_id=org_id or "0", user_id=user_id or "",
purpose="utility", timeout=60, project_id=project_id or "")
except Exception as e:
logger.warning("judge_completeness LLM 失败(走代码表兜底): %s", repr(e)[:150])
return _check_completeness_code(cap, task, biz)
import re as _re
txt = (raw or "").strip()
m = _re.search(r"\{[\s\S]*\}", txt)
if not m:
logger.warning("judge_completeness 输出非 JSON走代码表兜底: %s", txt[:120])
return _check_completeness_code(cap, task, biz)
try:
obj = json.loads(m.group(0))
except Exception:
return _check_completeness_code(cap, task, biz)
if not isinstance(obj, dict):
return _check_completeness_code(cap, task, biz)
if obj.get("complete") is True:
return ""
ask = str(obj.get("ask") or "").strip()
if not ask:
# LLM 判不完备但没给追问话术 → 代码表兜底生成话术(追问必须可行动)
return _check_completeness_code(cap, task, biz)
return ("调用「%s」模型前完备性检查未通过,已停在调用前、未产生任何费用。%s"
% (model_name or (_CAP_DESC.get(cap, cap)), ask))
async def tool_list_platform_models(params, org_id):
"""列出平台可用模型(本机构+平台owner机构含能力类型与描述"""
cap = str((params or {}).get("capability") or "").strip().lower()
try:
from pipeline_llm.selection import models_catalog
except ImportError:
return "FAIL: 模型治理模块pipeline-llm未安装无法列出平台模型。"
try:
caps = (cap,) if cap else ()
models = await models_catalog(org_id or "0", capabilities=caps)
except Exception as e:
return "ERROR: 列模型失败: " + str(e)[:300]
if not models:
scope = ("能力 " + cap) if cap else "全部能力"
return ("平台当前无可用模型(" + scope + ";范围=本机构+平台owner机构"
"请在模型治理→模型注册中添加。")
lines = ["平台可用模型(" + str(len(models)) + " 个,本机构+平台owner机构"]
for m in models:
cap_txt = m.get("capability") or "t2t"
cap_cn = _CAP_DESC.get(cap_txt, cap_txt)
desc = ("" + m["description"][:80]) if m.get("description") else ""
vendor = (" [" + m["vendor"] + "]") if m.get("vendor") else ""
lines.append(" - " + m.get("name", "") + vendor
+ "" + cap_cn + "/" + cap_txt + "" + desc)
return "\n".join(lines)
async def _resolve_capability(model_name, org_id):
"""按模型注册名反查 (能力类型, 描述)(门禁需知道查哪份输入契约;
描述供 LLM 裁决参考)。走 models_catalog与推理链同一机构可见性
查不到返回 ('', '')。
"""
try:
from pipeline_llm.selection import models_catalog
models = await models_catalog(org_id or "0", capabilities=())
for m in models:
if model_name in (m.get("name"), m.get("vendor_model_id")):
return ((m.get("capability") or "").strip().lower(),
m.get("description") or "")
except Exception as e:
logger.warning("_resolve_capability(%s) 失败: %s", model_name, e)
return "", ""
async def tool_invoke_model(params, org_id, user_id="", project_id="", session_id=""):
"""调用平台模型完成生成类任务(文生图/视频/语音等非对话能力)。
流程:解析 task/model/capability/params → 未指定 model 时按 task 自动
选型auto_select_model候选=本机构+平台ownercapability 空=全能力)
→ **完备性硬门禁**(缺必需输入则 QUESTION 追问,不调上游不花钱)
→ 组包 payload → llm_infer 统一推理端点 → 提取生成物 URL 返回。
失败返回真实可行动错误(禁静默粉饰)。
"""
p = params or {}
task = str(p.get("task") or "").strip()
if not task:
return "FAIL: invoke_model 需要 task任务描述/提示词)"
model = str(p.get("model") or "").strip()
cap = str(p.get("capability") or "").strip().lower()
# 业务参数JSON 字符串 → dict
biz = {}
raw_params = p.get("params")
if isinstance(raw_params, dict):
biz = dict(raw_params)
elif isinstance(raw_params, str) and raw_params.strip():
try:
parsed = json.loads(raw_params)
if isinstance(parsed, dict):
biz = parsed
except Exception:
return ("FAIL: params 不是合法 JSON 对象。媒体输入用三数组:"
"image_files/audio_files/video_filesURL或base64数组")
# 自动选型:未显式指定 model 时按 task 匹配
if not model:
try:
from pipeline_llm.selection import auto_select_model
# capability 空 = 全能力候选,让匹配器按任务选最合适能力
model, reason = await auto_select_model(
org_id or "0", task, user_id=user_id or "",
capabilities=(cap if cap else ""))
if model:
logger.info("invoke_model 自动选型: %s%sorg=%s",
model, reason, org_id or "0")
else:
return ("FAIL: 未能自动匹配到合适模型(" + str(reason) + ")。"
"可先用 list_platform_models 查看可用模型,"
"再用 model 参数指定。")
except ImportError:
return ("FAIL: 模型治理模块pipeline-llm未安装"
"无法自动选型或调用平台模型。")
except Exception as e:
return "ERROR: 自动选型失败: " + str(e)[:300]
# ── 完备性门禁2026-09-08 用户定夺:判断归 LLM代码表只兜底──
# 必须在 llm_infer 之前:缺输入直接 QUESTION 追问用户,上游一分钱不花。
# 判断分层:① 模型配套技能skills/models/{model}/SKILL.md→ LLM 按契约判;
# ② 无技能 → LLM 按能力语义判;③ LLM 失败 → 能力契约代码表兜底。
# capability/description 未显式给出时按选中模型反查。
desc = ""
cap2, desc = await _resolve_capability(model, org_id)
cap_req = cap or cap2
# 能力降级门禁规划:(契约能力, 是否代码表判定)
cap_gate, use_code = _gate_plan(cap_req, cap2, biz)
if use_code:
gap = _check_completeness_code(cap_gate, task, biz)
else:
gap = await judge_completeness(model, cap_gate, task, biz,
description=desc, org_id=org_id or "0",
user_id=user_id or "",
project_id=project_id or "")
cap = cap_gate
if gap:
logger.info("invoke_model 完备性门禁拦截(%s: %s", cap or "?", gap[:80])
return "QUESTION: " + gap
# 组包:生成类模型从 messages 末条取 prompt 文本inference._last_prompt_text
payload = {"messages": [{"role": "user", "content": task}]}
payload.update(biz)
try:
from .llm_bridge import llm_infer
data = await llm_infer(
payload, model=model, org_id=org_id or "0",
user_id=user_id or "", timeout=600, project_id=project_id or "",
session_id=session_id or "")
except Exception as e:
return "FAIL: 模型「" + model + "」调用失败:" + str(e)[:400]
# 提取生成物media本地持久 URL优先> choices[0].message.content
result_url = ""
media = data.get("media") if isinstance(data, dict) else None
if isinstance(media, dict):
result_url = (media.get("video") or media.get("image")
or media.get("audio") or media.get("glb")
or media.get("3dmodel") or "")
if not result_url and isinstance(data, dict):
try:
result_url = ((data.get("choices") or [{}])[0]
.get("message") or {}).get("content") or ""
except Exception:
result_url = ""
result_url = str(result_url or "").strip()
if result_url:
return ("OK: 模型「" + model + "」生成完成。产物地址:" + result_url)
# 无产物 URL回原始输出摘要真实不粉饰
try:
summary = json.dumps(data, ensure_ascii=False, default=str)[:500]
except Exception:
summary = str(data)[:500]
return ("模型「" + model + "」已调用但未返回可识别的产物地址。原始输出:"
+ summary)
# 平台模型工具的 v1 形态定义AGENT_TOOLS 同款 {name,description,params}
# 供 agent_loop.role_agent_run 追加到角色 agent 工具清单。
PLATFORM_MODEL_TOOLS_V1 = [
{
"name": "list_platform_models",
"description": ("列出平台当前可用的模型(本机构+平台owner机构的模型含能力类型"
"t2t对话/i2t图像理解/t2i文生图/t2v文生视频/i2v图生视频/tts语音合成/"
"asr语音识别等。需要调用非对话能力生图/视频/语音)前先查模型时用。"
"capability 参数可按能力过滤(如 t2i"),
"params": {"capability": "可选:能力类型过滤(如 t2i/t2v/tts空=全部"},
},
{
"name": "invoke_model",
"description": ("调用平台模型完成生成类任务(文生图/图生视频/文生视频/语音合成等"
"非对话能力;写代码/写文档等对话类工作由你自己完成,不要用本工具)。"
"model 可空——空时平台根据 task 自动匹配最合适的可用模型。"
"生成产物返回本地持久 URL产物需要落盘时用 write_file 记录 URL"),
"params": {
"task": "任务描述/提示词(必填,如「一只在月球上弹吉他的猫」)",
"model": "可选:模型注册名(空=按任务自动匹配)",
"capability": "可选能力类型t2i/t2v/i2v/tts/asr等自动匹配时用于过滤候选",
"params": ("可选:业务参数 JSON 字符串。媒体输入用三数组契约:"
"image_files/audio_files/video_files值为公网URL或base64的数组"
"生成参数如 resolution/duration/size 按模型文档"),
},
},
]
async def exec_platform_model_tool(tool, params, org_id, user_id="",
project_id=""):
"""v1/v2 统一分发入口(薄壳)。"""
if tool == "list_platform_models":
return await tool_list_platform_models(params, org_id)
if tool == "invoke_model":
return await tool_invoke_model(params, org_id, user_id=user_id,
project_id=project_id)
return "未实现: " + str(tool)