368 lines
17 KiB
Python
368 lines
17 KiB
Python
"""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")
|