From 08cf9272cdcdacd423f06a03a0c4c8bde20cd2b3 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Mon, 7 Sep 2026 16:47:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(selection):=20=E4=BB=BB=E5=8A=A1=E8=87=AA?= =?UTF-8?q?=E9=80=82=E5=BA=94=E8=87=AA=E5=8A=A8=E9=80=89=E5=9E=8Bauto=5Fse?= =?UTF-8?q?lect=5Fmodel+models=5Fcatalog(=E6=9C=AC=E6=9C=BA=E6=9E=84+owner?= =?UTF-8?q?=E6=9C=BA=E6=9E=84=E5=80=99=E9=80=89,utility=E8=AF=AD=E4=B9=89?= =?UTF-8?q?=E5=8C=B9=E9=85=8D,owner=E7=A1=AC=E6=A0=A1=E9=AA=8C=E5=8F=8C?= =?UTF-8?q?=E4=BF=9D=E9=99=A9);=E6=B2=BB=E7=90=86=E9=93=BEgovern=5Fresolve?= =?UTF-8?q?=E5=8A=A0=E6=A8=A1=E5=9E=8B=E5=BD=92=E5=B1=9E=E6=A0=A1=E9=AA=8C?= =?UTF-8?q?=E2=80=94=E2=80=94=E4=BB=96=E6=9C=BA=E6=9E=84=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E5=89=94=E9=99=A4=E9=A1=BA=E5=BB=B6=E6=9B=BF=E4=BB=A3,?= =?UTF-8?q?=E6=98=BE=E5=BC=8F=E6=8C=87=E5=AE=9A=E8=B6=8A=E6=9D=83=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B=E6=8A=A5=E5=8F=AF=E8=A1=8C=E5=8A=A8=E9=94=99=E8=AF=AF?= =?UTF-8?q?(2026-09-07=E7=94=A8=E6=88=B7=E9=9C=80=E6=B1=82)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_llm/gateway.py | 21 ++++++ pipeline_llm/selection.py | 152 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 173 insertions(+) diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index b7091dc..ea6be1c 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -483,6 +483,25 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', return False, '__LEGACY__' # 指定模型不在新表 → 旧表兜底 return False, ('生效策略未配置模型：请在模型治理→组织容错策略配置主/备模型，' '或在调用时指定模型名') + # ⑤.5 模型归属硬校验（2026-09-07 用户需求）：只有平台 owner（'0'/系统级共享） + # 或本机构的模型可调用；他机构模型从链中剔除，链内替代模型（策略备链）自然顺延。 + # 显式指定的模型被剔除且链内无替代 → 报可行动错误（禁止静默回退/越权放行）。 + from .selection import _owner_allowed + allowed = [m for m in models if _owner_allowed(m.get('org_id'), org_id)] + if len(allowed) != len(models): + for m in models: + if not _owner_allowed(m.get('org_id'), org_id): + logger.info("pipeline_llm: 模型 %s 归属机构 %s，调用方 %s 无权使用，" + "已从模型链剔除（顺延替代模型）", + m.get('name'), m.get('org_id'), org_id or '(未指定)') + if not allowed: + if model_name: + return False, ('模型「%s」归属其他机构（非平台 owner 共享、也非本机构模型），' + '不允许调用。请选择本机构或平台共享模型' + '（模型治理→模型注册可查看归属）' % model_name) + return False, ('生效策略模型链全部归属其他机构，调用方 %s 无可调用模型——' + '请检查机构策略配置' % (org_id or '(未指定)')) + models = allowed # ⑥ 逐模型尝试（辅助 → 主 → 备容错） last_err = '' chosen = None @@ -496,6 +515,8 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', # 本机构链全不可用 → 容错落 owner 链 if chosen is None and policy_org != '0': models0, pref0 = await _model_chain(sor, '0', '', purpose) + # owner 容错链同样过归属校验（策略里可能配了他机构注册的模型） + models0 = [m for m in models0 if _owner_allowed(m.get('org_id'), org_id)] for m in models0: ok, cand = await _pick_candidate(sor, m, pref0) if ok and isinstance(cand, tuple): diff --git a/pipeline_llm/selection.py b/pipeline_llm/selection.py index 3d333e1..1e65d53 100644 --- a/pipeline_llm/selection.py +++ b/pipeline_llm/selection.py @@ -162,6 +162,156 @@ async def resolve_model_name(model_ref, org_id='', capabilities=()): return getattr(recs[0], 'name', '') or '' +def _owner_allowed(model_org, org_id): + """模型归属校验（2026-09-07 用户需求）：平台 owner（'0'/空 = 系统级共享） + 或本机构的模型才可调用；其他机构的模型一律拒绝（由调用方顺延替代候选）。""" + m = (model_org or '').strip() + return m in ('', '0') or m == (org_id or '').strip() + + +async def models_catalog(org_id, capabilities=(), limit=50): + """本机构可用模型明细（本机构 + 平台 owner('0')/系统级共享）。 + + 供 agent 的 list_models 工具与 auto_select_model 候选池共用——与推理链 + （govern_resolve）同一机构可见性语义，杜绝两处 SQL 漂移。 + 返回 [dict]: name/vendor_model_id/capability/description/vendor/org_id。 + """ + db, dbname = _get_sor() + caps = _norm_caps(capabilities) + async with db.sqlorContext(dbname) as sor: + sql = ("SELECT m.id, m.name, m.vendor_model_id, m.capability, m.description, " + "m.org_id, v.name AS vendor_name " + "FROM llm_model m LEFT JOIN llm_vendor v ON v.id=m.vendor_id " + "WHERE m.status='active'") + params = {} + if org_id and org_id != '0': + sql += " AND (m.org_id=${org}$ OR m.org_id='' OR m.org_id='0')" + params['org'] = org_id + if caps: + # sqlor 的 IN 列表必须展开占位符（传 list 会崩）：${c0}$,${c1}$,... + ph = [] + for i, c in enumerate(caps): + k = 'c%d' % i + ph.append('${%s}$' % k) + params[k] = c + sql += (" AND COALESCE(NULLIF(m.capability,''),'t2t') IN (%s)" + % ','.join(ph)) + sql += " ORDER BY m.capability, m.name LIMIT %d" % int(limit) + recs = await sor.sqlExe(sql, params) + await sor.sqlExe("COMMIT", {}) + out = [] + for r in (recs or []): + out.append({ + 'name': getattr(r, 'name', '') or '', + 'vendor_model_id': getattr(r, 'vendor_model_id', '') or '', + 'capability': (getattr(r, 'capability', '') or 't2t').strip().lower(), + 'description': (getattr(r, 'description', '') or '').strip(), + 'vendor': getattr(r, 'vendor_name', '') or '', + 'org_id': getattr(r, 'org_id', '') or '', + }) + # owner 双保险（SQL 已过滤，防未来改动破坏不变量） + return [c for c in out if _owner_allowed(c['org_id'], org_id)] + + +async def auto_select_model(org_id, user_input, user_id='', capabilities='chat'): + """根据用户输入自动匹配最合适的模型（任务自适应路由）。 + + 候选池 = models_catalog（本机构 + 平台 owner('0')/系统级共享，与推理链 + 可见性一致），按能力白名单过滤（缺省 'chat' = CHAT_CAPS 对话能力； + 传 '' = 全能力，供 invoke_model 生成类任务自动选型）。 + 匹配 = utility 廉价模型语义选择（purpose='utility' 现成治理链，temperature=0）， + 依据 = 模型名 + 能力类型 + description（模型注册时填写的擅长说明）。 + + Owner 硬校验：LLM 返回的模型名逐个对照候选池（池外名字 = 幻觉，跳过）， + 命中候选即天然满足 owner ∈ ('', '0', 本机构)——池已过滤 + 此处双保险。 + 全部无效 → 兜底机构策略缺省模型（inference._pick_default_model_name）。 + + 返回 (model_name, reason)；无候选/匹配失败返回 ('', 原因)——调用方沿用 + 原有回退链（策略主备），不因此报错。 + """ + candidates = await models_catalog(org_id, capabilities=capabilities) + if not candidates: + return '', '无可用模型候选（本机构+平台owner机构均无该能力的启用模型）' + if len(candidates) == 1: + # 唯一候选：跳过 LLM 匹配（省一次 utility 调用） + return candidates[0]['name'], '唯一候选' + + # ── utility 模型语义匹配 ── + catalog = '\n'.join( + '- %s [%s]%s%s' % ( + c['name'], c['capability'], + (' ' + c['vendor']) if c['vendor'] else '', + ('：' + c['description'][:150]) if c['description'] else '') + for c in candidates) + task = (user_input or '').strip()[:600] + prompt = ( + '你是平台模型调度器。以下是本机构当前可用的模型目录' + '（格式：模型名 [能力类型] 供应商：描述）：\n%s\n\n' + '能力类型含义：t2t=文本对话 i2t=图像理解 m2t=多媒体理解 t2i=文生图 ' + 'i2v=图生视频 t2v=文生视频 r2v=参考生视频 tts=语音合成 asr=语音识别 ' + 'embedding=向量化 rerank=重排序\n\n' + '用户任务：\n%s\n\n' + '请根据任务的内容领域、语言、复杂度，从目录中选出最适合完成该任务的模型' + '（只能选目录中的模型名，按适合度从高到低排序，最多 3 个）。' + '只返回 JSON 对象：{"models": ["模型名1", "模型名2"], "reason": "一句话理由"}。' + '没有适合的模型时 models 返回空数组。' % (catalog, task)) + + from .inference import chat_inference + try: + data = await chat_inference( + org_id or '0', user_id or '', + {'messages': [{'role': 'user', 'content': prompt}], + 'temperature': 0, '_purpose': 'utility'}) + content = ((data.get('choices') or [{}])[0].get('message') or {}).get('content') or '' + except Exception as e: + logger.warning("auto_select_model 匹配调用失败（走策略链兜底）: %s", e) + return '', '匹配调用失败: %s' % str(e)[:120] + + # 解析 + 逐候选 owner/存在性校验 + import json as _json + import re as _re + picked = [] + reason = '' + m = _re.search(r'\{[\s\S]*\}', content or '') + if m: + try: + obj = _json.loads(m.group(0)) + if isinstance(obj, dict): + picked = [str(x).strip() for x in (obj.get('models') or []) if str(x or '').strip()] + reason = str(obj.get('reason') or '')[:120] + except Exception: + picked = [] + if not picked: + # 兜底：整体 JSON 解析失败时按候选名逐个在文本中找（顺序 = 目录序） + for c in candidates: + if c['name'] and c['name'] in (content or ''): + picked.append(c['name']) + break + by_name = {} + for c in candidates: + by_name[c['name']] = c + if c['vendor_model_id']: + by_name.setdefault(c['vendor_model_id'], c) + for name in picked: + c = by_name.get(name) + if c and _owner_allowed(c['org_id'], org_id): + return c['name'], (reason or '语义匹配') + if name: + logger.warning("auto_select_model: 跳过无效/越权候选 %s", name) + + # LLM 结果全无效 → 机构策略缺省模型兜底（治理语义一致） + try: + from .inference import _pick_default_model_name + _caps = _norm_caps(capabilities) + cap0 = _caps[0] if _caps else 't2t' + fallback = await _pick_default_model_name(org_id or '0', cap0) + if fallback: + return fallback, '匹配无效，回退机构策略缺省模型' + except Exception as e: + logger.debug("auto_select_model 策略兜底失败: %s", e) + return '', '未能匹配到合适模型（沿用原回退链）' + + async def org_available_models(org_id, capability=''): """机构可用模型注册名列表（本机构 + 系统级共享）。空 = 机构未配置模型。""" db, dbname = _get_sor() @@ -185,4 +335,6 @@ def load_selection(): env.llm_model_options = model_options env.llm_resolve_model_name = resolve_model_name env.llm_org_available_models = org_available_models + env.llm_models_catalog = models_catalog + env.llm_auto_select_model = auto_select_model logger.info("[pipeline_llm] selection loaded")