"""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)