pipeline-service/pipeline_service/platform_model_tools.py

180 lines
8.8 KiB
Python
Raw 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": "重排序",
}
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 tool_invoke_model(params, org_id, user_id=""):
"""调用平台模型完成生成类任务(文生图/视频/语音等非对话能力)。
流程:解析 task/model/capability/params → 未指定 model 时按 task 自动
选型(auto_select_model,候选=本机构+平台owner,capability 空=全能力)
→ 组包 payload(messages=[user:task] + 业务参数平铺,媒体三数组契约由
inference 层归一)→ 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_files(URL或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(%s)org=%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]
# 组包:生成类模型从 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)
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=""):
"""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)
return "未实现: " + str(tool)