"""pipeline_llm.selection — 模型选择的唯一收敛点（下拉数据/引用解析/缺省模型）。 2026-09-04 收敛改造：此前模型选择逻辑散在 pipeline-sdlc（cockpit/get_model_options/ set_agent_model）、pipeline-core（agent_model_options、旧 llm CRUD）、pipeline-service （gateway/agent_loop 内联 SQL）多处，各查各的表、语义漂移。现全部收敛到本模块： - model_options() 所有模型下拉的唯一数据源 - resolve_model_name() 模型引用（id/vendor_model_id/name）→ 注册名的唯一解析点 - org_available_models() 机构可用模型清单（冒泡检测用） - 缺省模型由机构策略决定（inference._pick_default_model_name），不再写死模型名 其他模块只允许薄壳委托（dspy 3-5 行）或函数调用，禁止再内联模型表 SQL。 """ import logging from sqlor.dbpools import DBPools from ahserver.serverenv import ServerEnv logger = logging.getLogger("pipeline_llm.selection") # 会话/chat 形态可用的能力类型（2026-09-06 用户定夺）：文本输出的对话能力。 # 会话 agent 模型下拉只列这些；embedding/rerank/图视频生成等形态各有专属入口。 CHAT_CAPS = ('t2t', 'i2t', 'm2t') # 能力降级表（2026-09-08 用户定夺，与 r2v→t2v 同例）：请求能力无可用模型时， # 允许降级用「超集能力」模型执行任务。方向 = 降级目标模型的输入契约覆盖 # 请求能力： # i2i（图像→图像，接受文本提示+可选图）可执行 t2i（文本→图像）任务 # r2v（参考媒体→视频，接受文本+参考媒体）可执行 t2v（文本→视频）任务 # 单向降级：t2i 模型不能接 i2i（缺源图）、t2v 不能接 r2v——反向永远非法。 CAP_DEGRADE = {'t2i': 'i2i', 't2v': 'r2v'} # 反向索引：降级源能力 → 请求能力（门禁契约换算用：i2i 模型无图输入时按 t2i 契约校验） CAP_DEGRADE_SOURCE = {v: k for k, v in CAP_DEGRADE.items()} def degrade_expand(caps): """能力白名单 + 其合法降级源能力（候选池扩展用）。无映射时原样返回。""" out = [str(c).strip().lower() for c in (caps or ()) if str(c or '').strip()] for c in list(out): d = CAP_DEGRADE.get(c) if d and d not in out: out.append(d) return tuple(out) def _norm_caps(capabilities): """capabilities 参数归一：'chat' 哨兵 → CHAT_CAPS；逗号串 → tuple； tuple/list 原样。返回 tuple（空 = 不过滤）。dspy 薄壳只传 'chat'，零 import。""" if isinstance(capabilities, str): s = capabilities.strip().lower() if not s: return () if s == 'chat': return CHAT_CAPS return tuple(c.strip().lower() for c in s.split(',') if c.strip()) return tuple(str(c).strip().lower() for c in (capabilities or ()) if str(c or '').strip()) def _get_sor(): env = ServerEnv() fn = getattr(env, 'get_module_dbname', None) dbname = 'pipeline' if callable(fn): try: dbname = fn('pipeline_llm') or 'pipeline' except Exception: dbname = 'pipeline' return DBPools(), dbname async def model_options(org_id, uid='', session_id='', pipeline_id='', value_field='id', capabilities=()): """模型选择下拉的唯一数据源（对齐 llmage 分类：capability=能力类型）。 机构语义与推理链一致：本机构 + 系统级共享（org_id 空/'0'）可见。 selected 标记：项目已设模型（sd_projects.default_model，存 name）> 个人全局选择（pipeline_agent_settings.default_llm_id，存 id）。 value_field: 'id'=下拉值用模型 id（AgentIO 场景）；'name'=用注册名（角色模型配置场景）。 capabilities: 能力类型白名单（tuple/list，如会话 agent 传 CHAT_CAPS=('t2t','i2t','m2t')）。 空 = 不过滤（历史行为）。注意：存量模型 capability 为空的行按 't2t' 语义对待， 过滤时用 COALESCE+NULLIF 把空串归一成 't2t' 再比对（DDL 有默认值， 但历史迁移行可能是空串）。 """ db, dbname = _get_sor() rows = [] caps = _norm_caps(capabilities) async with db.sqlorContext(dbname) as sor: sql = ("SELECT m.id, m.name, m.vendor_model_id, m.capability, 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.name" recs = await sor.sqlExe(sql, params) # 项目已设模型（按会话解析当前项目） project_model = '' if uid: try: from pipeline_service.workspace import get_session_project_id _pid = await get_session_project_id( sor, uid, session_id or '', pipeline_id or '') if _pid: _p = await sor.sqlExe( "SELECT default_model FROM sd_projects WHERE id=${p}$", {"p": _pid}) await sor.sqlExe("COMMIT", {}) if _p: project_model = getattr(_p[0], 'default_model', '') or '' except Exception as e: logger.debug("model_options 项目模型解析跳过: %s", e) # 个人全局默认选择 current_llm_id = '' if uid: try: _s = await sor.sqlExe( "SELECT default_llm_id FROM pipeline_agent_settings WHERE user_id=${u}$", {"u": uid}) if _s: current_llm_id = getattr(_s[0], 'default_llm_id', '') or '' except Exception: pass for r in (recs or []): vname = getattr(r, 'vendor_name', '') or '' rows.append({ 'value': r.id if value_field != 'name' else r.name, 'text': r.name + (' (' + vname + ')' if vname else ''), 'provider': vname, 'model_id': r.id, 'model_id_text': r.name, 'capabilities': getattr(r, 'capability', '') or 't2t', 'selected': ((r.name == project_model) if project_model else (r.id == current_llm_id)), }) return rows async def resolve_model_name(model_ref, org_id='', capabilities=()): """模型引用解析的唯一入口：id / vendor_model_id / name → 模型注册名。 机构隔离：非系统级机构只能解析「本机构 + 系统级共享」模型。 capabilities: 能力白名单（'chat' 哨兵 / tuple），非空时能力不符也解析失败 （会话 agent 个人默认模型/角色模型入口传 'chat'，防把 embedding 等非对话 模型持久化成会话模型）。 解析不到返回 ''（调用方决定回退/报错，禁止自行另查表）。 """ if not model_ref: return '' caps = _norm_caps(capabilities) db, dbname = _get_sor() async with db.sqlorContext(dbname) as sor: recs = await sor.sqlExe( "SELECT name, org_id, capability FROM llm_model WHERE status='active' " "AND (id=${m}$ OR vendor_model_id=${m}$ OR name=${m}$) LIMIT 1", {"m": model_ref}) await sor.sqlExe("COMMIT", {}) if not recs: return '' m_org = getattr(recs[0], 'org_id', '') or '' if org_id and org_id != '0' and m_org not in ('', '0', org_id): logger.warning("resolve_model_name: 模型 %s 不属于机构 %s", model_ref, org_id) return '' if caps: cap = (getattr(recs[0], 'capability', '') or 't2t').strip().lower() if cap not in caps: logger.warning("resolve_model_name: 模型 %s 能力 %s 不在白名单 %s", model_ref, cap, caps) return '' 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) # 能力降级（2026-09-08）：候选池含请求能力的合法降级源模型 # （t2i→i2i、t2v→r2v），无可用 t2i 模型时 i2i 模型可承接文生图任务 if caps: caps = degrade_expand(caps) 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' '能力降级规则：无 t2i 模型时 i2i（图生图）模型可承接文生图任务；' '无 t2v 模型时 r2v（参考生视频）模型可承接文生视频任务；' '反向（t2i 接图生图、t2v 接参考生视频）非法。\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() async with db.sqlorContext(dbname) as sor: sql = "SELECT name, capability FROM llm_model WHERE status='active'" params = {} if org_id and org_id != '0': sql += " AND (org_id=${org}$ OR org_id='' OR org_id='0')" params['org'] = org_id if capability: sql += " AND capability=${c}$" params['c'] = capability recs = await sor.sqlExe(sql, params) await sor.sqlExe("COMMIT", {}) return [getattr(r, 'name', '') or '' for r in (recs or [])] def load_selection(): """注册到 ServerEnv（供各模块 dspy 薄壳调用）。""" env = ServerEnv() 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")