From 611008cb38786e5a72765912f7e428fdb64c4fd2 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 4 Sep 2026 17:03:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(llm):=20=E7=BB=9F=E4=B8=80=E6=8E=A8?= =?UTF-8?q?=E7=90=86API(v1/chat/completions+v1/models=E6=8C=89=E8=83=BD?= =?UTF-8?q?=E5=8A=9B=E5=88=86=E7=B1=BB=E7=85=A7llmage)+=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E9=80=89=E6=8B=A9=E6=94=B6=E6=95=9B=E5=B1=82selection(?= =?UTF-8?q?=E4=B8=8B=E6=8B=89/=E5=BC=95=E7=94=A8=E8=A7=A3=E6=9E=90?= =?UTF-8?q?=E5=94=AF=E4=B8=80=E5=85=A5=E5=8F=A3)+=E6=97=A7=E8=A1=A8?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=E8=84=9A=E6=9C=AC;agent=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E7=BB=8Fllm=5Fbridge=20HTTP=E8=87=AA=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E6=94=B6=E6=95=9B=E8=87=B3=E6=AD=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- json/llm_org_policy_list.json | 6 + models/llm_org_policy.json | 12 + models/llm_usage.json | 10 +- mysql.ddl.sql | 4 +- pipeline_llm/accounting.py | 161 ++++++++++++++ pipeline_llm/gateway.py | 158 ++++++++----- pipeline_llm/inference.py | 318 +++++++++++++++++++++++++++ pipeline_llm/init.py | 27 +++ pipeline_llm/product_interface.py | 196 +++++++++++++++++ pipeline_llm/selection.py | 147 +++++++++++++ scripts/load_path.py | 12 +- scripts/migrate_legacy_llm.py | 220 ++++++++++++++++++ wwwroot/api/v1/chat/completions.dspy | 79 +++++++ wwwroot/api/v1/models.dspy | 45 ++++ 14 files changed, 1331 insertions(+), 64 deletions(-) create mode 100644 pipeline_llm/accounting.py create mode 100644 pipeline_llm/inference.py create mode 100644 pipeline_llm/product_interface.py create mode 100644 pipeline_llm/selection.py create mode 100644 scripts/migrate_legacy_llm.py create mode 100644 wwwroot/api/v1/chat/completions.dspy create mode 100644 wwwroot/api/v1/models.dspy diff --git a/json/llm_org_policy_list.json b/json/llm_org_policy_list.json index adaf0a7..79174b4 100644 --- a/json/llm_org_policy_list.json +++ b/json/llm_org_policy_list.json @@ -14,6 +14,12 @@ "valueField": "value", "textField": "text" }, + "utility_model_id": { + "uitype": "code", + "dataurl": "{{entire_url('../api/get_llm_model_options.dspy')}}", + "valueField": "value", + "textField": "text" + }, "endpoint_pref": { "uitype": "code", "dataurl": "{{entire_url('../api/get_llm_endpoint_pref_options.dspy')}}", diff --git a/models/llm_org_policy.json b/models/llm_org_policy.json index 699375c..07a1357 100644 --- a/models/llm_org_policy.json +++ b/models/llm_org_policy.json @@ -30,6 +30,12 @@ "type": "str", "length": 32 }, + { + "name": "utility_model_id", + "title": "辅助模型ID", + "type": "str", + "length": 32 + }, { "name": "backup_model_ids", "title": "备模型ID列表(JSON)", @@ -79,6 +85,12 @@ "valuefield": "id", "textfield": "name" }, + { + "field": "utility_model_id", + "table": "llm_model", + "valuefield": "id", + "textfield": "name" + }, { "field": "endpoint_pref", "table": "appcodes_kv", diff --git a/models/llm_usage.json b/models/llm_usage.json index 51fa6ce..58487ef 100644 --- a/models/llm_usage.json +++ b/models/llm_usage.json @@ -69,12 +69,20 @@ }, { "name": "charge", - "title": "消费金额(组织池)", + "title": "客户应付(异步三方账计算)", "type": "float", "length": 12, "dec": 6, "default": "0" }, + { + "name": "accounting_status", + "title": "出账状态(na仅成本/pending待三方账/accounted已出账)", + "type": "str", + "length": 20, + "nullable": "no", + "default": "'na'" + }, { "name": "ppid", "title": "定价项目ID", diff --git a/mysql.ddl.sql b/mysql.ddl.sql index 2c5f1b9..6823366 100644 --- a/mysql.ddl.sql +++ b/mysql.ddl.sql @@ -78,6 +78,7 @@ CREATE TABLE IF NOT EXISTS llm_org_policy ( `id` varchar(32) NOT NULL comment '主键ID', `org_id` varchar(32) NOT NULL comment '机构ID', `primary_model_id` varchar(32) comment '主模型ID', + `utility_model_id` varchar(32) comment '辅助模型ID(轻量任务:分类/选择/摘要)', `backup_model_ids` text comment '备模型ID列表(JSON)', `endpoint_pref` varchar(20) NOT NULL DEFAULT 'any' comment '端点偏好', `status` varchar(20) NOT NULL DEFAULT 'active' comment '状态', @@ -124,7 +125,8 @@ CREATE TABLE IF NOT EXISTS llm_usage ( `req_tokens` int NOT NULL DEFAULT 0 comment '请求token', `resp_tokens` int NOT NULL DEFAULT 0 comment '响应token', `cost` decimal(12,6) NOT NULL DEFAULT 0 comment '成本金额(账号侧)', - `charge` decimal(12,6) NOT NULL DEFAULT 0 comment '消费金额(组织池侧)', + `charge` decimal(12,6) NOT NULL DEFAULT 0 comment '客户应付(异步三方账计算,本机构模型恒0)', + `accounting_status` varchar(20) NOT NULL DEFAULT 'na' comment '出账状态(本机构模型仅成本=na/owner模型待三方账=pending/已出账=accounted)', `ppid` varchar(32) comment '定价项目ID', `task_ref` varchar(100) comment '调用来源', `status` varchar(20) NOT NULL DEFAULT 'ok' comment '状态(ok/failed/recharge)', diff --git a/pipeline_llm/accounting.py b/pipeline_llm/accounting.py new file mode 100644 index 0000000..9ee4410 --- /dev/null +++ b/pipeline_llm/accounting.py @@ -0,0 +1,161 @@ +"""pipeline_llm.accounting — 异步出账循环（owner 模型三方账）。 + +扫 llm_usage 中 accounting_status='pending' 的用量流水（owner 模型、调用方≠平台）， +逐条映射模型→产品，调产品层 product_accounting_generic 出三方账 +（客户付/商户营收/供应商成本，客户折扣精确到产品），成功置 'accounted'。 + +与账号/存储的同步记账分工明确（2026-09 用户定夺）： + - 账号/存储：purchase_realtime 同步记账 + - 模型（本模块）：调用时只记成本侧 + pending 流水，出账走本循环（异步） + +单循环守卫（多进程部署时防重复出账）：Redis SETNX llm_acc:lock（db4，与限流同库）。 +""" + +import asyncio +import logging +import time + +from ahserver.serverenv import ServerEnv + +from .gateway import _get_db, _redis + +logger = logging.getLogger("pipeline_llm.accounting") + +_SCAN_INTERVAL = 60 # 扫描间隔（秒） +_BATCH = 200 # 每批处理条数 +_LOCK_KEY = 'llm_acc:lock' +_LOCK_TTL = 90 # 锁 TTL（略大于单批预算耗时） + + +async def _get_pending(sor): + recs = await sor.sqlExe( + "SELECT id, org_id, user_id, model_id, req_tokens, resp_tokens, cost, ppid " + "FROM llm_usage WHERE accounting_status='pending' AND status='ok' " + "ORDER BY created_at LIMIT %d" % _BATCH, {}) + await sor.sqlExe("COMMIT", {}) + return recs or [] + + +async def _find_product_id(sor, model_id): + """llm_model.id → product.id（产品层导入时按 resource_ref_id 建立）。""" + recs = await sor.sqlExe( + "SELECT id FROM product WHERE resource_ref_id=${r}$ AND status='1' LIMIT 1", + {"r": model_id}) + await sor.sqlExe("COMMIT", {}) + return (getattr(recs[0], 'id', '') or '') if recs else '' + + +async def _settle_one(sor, row): + """单条出账。成功置 accounted；失败置 failed 并记录原因（不静默重试死循环）。""" + env = ServerEnv() + model_id = getattr(row, 'model_id', '') or '' + product_id = await _find_product_id(sor, model_id) + if not product_id: + await sor.U('llm_usage', { + 'id': row.id, 'accounting_status': 'failed', + 'note': '出账失败：模型(%s)未映射到产品，请先在产品管理导入产线模型' % model_id, + }) + await sor.sqlExe("COMMIT", {}) + logger.warning("pipeline_llm.accounting: model %s 无产品映射，流水 %s 置 failed", + model_id, row.id) + return False + usage_data = { + 'prompt_tokens': int(getattr(row, 'req_tokens', 0) or 0), + 'completion_tokens': int(getattr(row, 'resp_tokens', 0) or 0), + } + fn = getattr(env, 'product_accounting_generic', None) + if fn is None: + # 产品模块未加载 → 保留 pending 下轮再试 + return False + result = await fn( + product_id=product_id, + usage_data=usage_data, + userorgid=getattr(row, 'org_id', '') or '', + userid=getattr(row, 'user_id', '') or '', + ) + if not result or not result.get('success', result.get('orderid')): + msg = str((result or {}).get('message', result))[:180] + await sor.U('llm_usage', { + 'id': row.id, 'accounting_status': 'failed', + 'note': '出账失败：%s' % msg, + }) + await sor.sqlExe("COMMIT", {}) + logger.error("pipeline_llm.accounting: 流水 %s 出账失败: %s", row.id, msg) + return False + # 回填客户应付 + 置已出账 + await sor.sqlExe( + "UPDATE llm_usage SET charge=${c}$, accounting_status='accounted', " + "note=CONCAT(COALESCE(note,''),'|accounted:',${o}$) WHERE id=${i}$", + {"c": float((result or {}).get('customer_amount', 0) or 0), + "o": str((result or {}).get('orderid', ''))[:40], + "i": row.id}) + await sor.sqlExe("COMMIT", {}) + return True + + +async def _run_one_pass(): + """一轮扫描-出账。返回处理条数。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + rows = await _get_pending(sor) + if not rows: + return 0 + ok_n = 0 + for row in rows: + try: + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + if await _settle_one(sor, row): + ok_n += 1 + except Exception as e: + logger.error("pipeline_llm.accounting: 流水 %s 出账异常: %s", + getattr(row, 'id', ''), e) + try: + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + await sor.U('llm_usage', { + 'id': row.id, 'accounting_status': 'failed', + 'note': '出账异常：%s' % str(e)[:180], + }) + await sor.sqlExe("COMMIT", {}) + except Exception: + pass + return ok_n + + +async def llm_usage_accounting_loop(): + """后台循环：抢分布式锁 → 出账一批 → 睡。永不抛出。""" + logger.info("[pipeline_llm] 异步出账循环已启动（间隔 %ds）", _SCAN_INTERVAL) + while True: + try: + r = _redis() + if r is not None: + # 抢锁（多进程只跑一个）；Redis 故障则直接跑（与限流 fail-open 一致） + got = False + try: + got = bool(r.set(_LOCK_KEY, str(time.time()), nx=True, ex=_LOCK_TTL)) + except Exception: + got = True + if not got: + await asyncio.sleep(_SCAN_INTERVAL) + continue + await _run_one_pass() + except Exception as e: + logger.error("pipeline_llm.accounting: 循环异常（继续）: %s", e) + await asyncio.sleep(_SCAN_INTERVAL) + + +def start_accounting_loop(): + """注册到当前事件循环（load 时调用）。已有则跳过。""" + try: + loop = asyncio.get_event_loop() + except RuntimeError: + loop = None + if loop is None or not loop.is_running(): + logger.info("[pipeline_llm] 无运行中的事件循环，跳过出账循环注册") + return + for t in asyncio.all_tasks(loop): + coro = getattr(t, 'get_coro', lambda: None)() + if coro is not None and getattr(coro, '__name__', '') == 'llm_usage_accounting_loop': + return + loop.create_task(llm_usage_accounting_loop()) diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index d85f81b..903549d 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -210,29 +210,34 @@ async def _release_quota(sor, org_id: str, user_id: str, amount: float): # ────────────────────────── 门禁 ⑤⑥：策略选模型 + 候选轮转 ────────────────────────── async def _org_policy(sor, org_id: str): - """组织容错策略。无记录返回 (主模型空, 备链空, 端点偏好 any)。""" + """组织容错策略。无记录返回 (主模型空, 辅助模型空, 备链空, 端点偏好 any)。""" if not org_id: - return '', [], 'any' + return '', '', [], 'any' recs = await sor.sqlExe( - "SELECT primary_model_id, backup_model_ids, endpoint_pref FROM llm_org_policy " - "WHERE org_id=${o}$ AND status='active' LIMIT 1", {"o": org_id}) + "SELECT primary_model_id, utility_model_id, backup_model_ids, endpoint_pref " + "FROM llm_org_policy WHERE org_id=${o}$ AND status='active' LIMIT 1", {"o": org_id}) await sor.sqlExe("COMMIT", {}) if not recs: - return '', [], 'any' + return '', '', [], 'any' r = recs[0] return (getattr(r, 'primary_model_id', '') or '', + getattr(r, 'utility_model_id', '') or '', _parse_json(getattr(r, 'backup_model_ids', ''), []), getattr(r, 'endpoint_pref', '') or 'any') -async def _model_chain(sor, org_id: str, model_name: str): +async def _model_chain(sor, org_id: str, model_name: str, purpose: str = ''): """解析模型链 [主, 备...]，返回 llm_model 记录 dict 列表。 - 调用方指定 model_name：在 llm_model 找到 → [它] + 策略备链补充；找不到 → 空（走旧表兜底） - 未指定：策略主模型 + 备链；无策略 → 空 + - purpose='utility'（辅助任务：分类/选择/摘要）：链 = [辅助模型] + 主 + 备。 + 显式指定 model_name 时忽略 purpose（显式优先）；未配辅助模型时等同普通链。 + 辅助模型不可用（账号/端点全挂）由 govern_resolve 的逐模型尝试顺延到主模型， + 与主备容错同一机制，无新增降级路径。 """ chain_ids = [] - primary, backups, pref = await _org_policy(sor, org_id) + primary, utility, backups, pref = await _org_policy(sor, org_id) if model_name: recs = await sor.sqlExe( "SELECT id FROM llm_model WHERE status='active' AND (name=${n}$ OR vendor_model_id=${n}$) LIMIT 1", @@ -243,6 +248,8 @@ async def _model_chain(sor, org_id: str, model_name: str): chain_ids.append(getattr(recs[0], 'id', '')) chain_ids.extend([b for b in backups if b not in chain_ids]) else: + if purpose == 'utility' and utility: + chain_ids.append(utility) if primary: chain_ids.append(primary) chain_ids.extend([b for b in backups if b not in chain_ids]) @@ -409,20 +416,54 @@ async def governance_enabled(sor, org_id: str) -> bool: return enabled +async def _active_policy_org(sor, org_id: str) -> str: + """确定生效策略机构：本机构有 active 策略 → 本机构； + 否则平台 owner('0') 有 active 策略 → '0'（商业化默认：用平台资源即付费）； + 都没有 → ''（调用方回退旧 llm 表，零风险）。 + """ + if org_id and org_id != '0': + recs = await sor.sqlExe( + "SELECT COUNT(*) AS c FROM llm_org_policy WHERE org_id=${o}$ AND status='active'", + {"o": org_id}) + await sor.sqlExe("COMMIT", {}) + if recs and int(_fnum(getattr(recs[0], 'c', 0))) > 0: + return org_id + recs = await sor.sqlExe( + "SELECT COUNT(*) AS c FROM llm_org_policy WHERE org_id='0' AND status='active'", {}) + await sor.sqlExe("COMMIT", {}) + if recs and int(_fnum(getattr(recs[0], 'c', 0))) > 0: + return '0' + return '' + + async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', - est_tokens: int = 0, task_ref: str = ''): - """门禁链 ①-⑥ + 预授权。 + est_tokens: int = 0, task_ref: str = '', purpose: str = ''): + """门禁链 ①-⑥。 + + 策略来源（2026-09 商业化默认）： + 本机构有策略 → 用本机构模型链； + 本机构无策略 → 用平台 owner('0') 策略（用平台资源即付费）； + owner 也没策略 → __LEGACY__ 旧表兜底。 + 本机构链路全不可用 → 容错顺延 owner 链（辅助→主→备同样机制）。 + + purpose='utility'：辅助任务（分类/选择/摘要），链 = 辅助→主→备。显式 model_name 优先。 + + 记账决策（2026-09）：**不做组织池预授权**—— + 本机构模型只记成本侧（供应商账号扣减），不产生组织/个人消费； + owner 模型三方账（客户/商户/供应商）走异步产品计费链路。 + 防滥用第一道闸保留：限流。 返回 (ok, result)： ok=True → result = {api_base, api_key(明文,仅进程内), model_id, name, - account_id, endpoint_region, reserved(冻结金额), - model_row(定价信息), user_id, org_id, task_ref} + account_id, endpoint_region, reserved(恒0), + model_row, user_id, org_id, task_ref, policy_org_id} ok=False → result = 错误信息（真实可行动） - 机构未启用治理 → (False, '__LEGACY__') 特殊标记：调用方走旧 llm 表逻辑 + 无任何策略 → (False, '__LEGACY__')：调用方走旧 llm 表逻辑 """ db, dbname = _get_db() async with db.sqlorContext(dbname) as sor: - if not await governance_enabled(sor, org_id or ''): + policy_org = await _active_policy_org(sor, org_id or '') + if not policy_org: return False, '__LEGACY__' # ① 个人限流 if user_id: @@ -443,14 +484,14 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', ok, msg = await _rate_check('org', org_id, olimit) if not ok: return False, msg - # ⑤ 模型链（含 ⑥ 偏好） - models, pref = await _model_chain(sor, org_id or '', model_name or '') + # ⑤ 模型链（含 ⑥ 偏好）—— 显式 model_name 时全局查找 + models, pref = await _model_chain(sor, policy_org, model_name or '', purpose) if not models: if model_name: return False, '__LEGACY__' # 指定模型不在新表 → 旧表兜底 - return False, ('机构未配置模型容错策略且未指定模型：请在模型治理→组织容错策略配置主/备模型，' + return False, ('生效策略未配置模型：请在模型治理→组织容错策略配置主/备模型，' '或在调用时指定模型名') - # ⑥ 逐模型尝试（主 → 备容错） + # ⑥ 逐模型尝试（辅助 → 主 → 备容错） last_err = '' chosen = None for m in models: @@ -459,20 +500,25 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', chosen = (m, cand) break last_err = cand if isinstance(cand, str) else last_err - logger.info("pipeline_llm: 模型 %s 候选不可用（%s），尝试备模型", m.get('name'), cand) + logger.info("pipeline_llm: 模型 %s 候选不可用（%s），尝试链内下一模型", m.get('name'), cand) + # 本机构链全不可用 → 容错落 owner 链 + if chosen is None and policy_org != '0': + models0, pref0 = await _model_chain(sor, '0', '', purpose) + for m in models0: + ok, cand = await _pick_candidate(sor, m, pref0) + if ok and isinstance(cand, tuple): + chosen = (m, cand) + policy_org = '0' + pref = pref0 + logger.info("pipeline_llm: 本机构模型链不可用，容错使用平台 owner 模型 %s", + m.get('name')) + break + last_err = cand if isinstance(cand, str) else last_err if chosen is None: - return False, '主/备模型全部不可用。最后原因：%s' % last_err + return False, '模型链全部不可用。最后原因：%s' % last_err model, chosen_pair = chosen acc, ep = chosen_pair - # ③④ 预授权（按预估用量冻结组织池+个人额度） - price_in = _fnum(model.get('price_input')) - price_out = _fnum(model.get('price_output')) - est = int(est_tokens or 0) or _EST_TOKENS_PER_CALL - est_charge = est / 1000.0 * (price_in + price_out) / 2.0 - ok, msg, reserved = await _reserve_quota(sor, org_id or '', user_id or '', est_charge) - if not ok: - return False, msg - # 组装结果（api_key 解密，只在本进程内存） + # ⑦ 组装结果（无预授权，reserved 恒 0；记账在 govern_settle 按来源分流） return True, { 'api_base': (ep.get('base_url') or '').rstrip('/'), 'api_key': decrypt_api_key(acc.get('api_key') or ''), @@ -482,70 +528,61 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', 'endpoint_region': ep.get('region') or '', 'endpoint_proxy': ep.get('proxy') or '', 'timeout': int(_fnum(ep.get('timeout')) or 60), - 'reserved': reserved, + 'reserved': 0.0, 'model_row': model, 'user_id': user_id or '', 'org_id': org_id or '', 'task_ref': task_ref or '', + 'policy_org_id': policy_org, } async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_tokens: int = 0, note: str = ''): - """结算 ⑧：按实际用量修正预授权 + 双维度记账。 + """结算 ⑧：成本侧扣减 + 用量流水（记账分流）。 - ctx govern_resolve 成功时返回的 dict - ok_call 上游调用是否成功（失败也记账：释放预授权，写 failed 流水） + ctx govern_resolve 成功时返回的 dict（含 policy_org_id） + ok_call 上游调用是否成功（失败也记账：写 failed 流水） + + 记账分流（2026-09）： + 本机构模型（含 owner 自用）：只记成本侧——供应商账号扣减， + 不动组织池/个人额度，流水 accounting_status='na'。 + owner 模型（调用方 ≠ '0'）：成本侧照常扣减，流水 + accounting_status='pending'，由异步产品计费链路出三方账 + （客户付/商户营收/供应商成本，折扣在出账时精确到产品）。 """ if not ctx or not isinstance(ctx, dict): return db, dbname = _get_db() try: model = ctx.get('model_row') or {} - price_in = _fnum(model.get('price_input')) - price_out = _fnum(model.get('price_output')) cost_in = _fnum(model.get('cost_input')) cost_out = _fnum(model.get('cost_output')) - charge = req_tokens / 1000.0 * price_in + resp_tokens / 1000.0 * price_out cost = req_tokens / 1000.0 * cost_in + resp_tokens / 1000.0 * cost_out - reserved = _fnum(ctx.get('reserved')) + policy_org = ctx.get('policy_org_id') or '' + caller_org = ctx.get('org_id') or '' + # owner 模型 = 生效策略是平台('0') 且调用方不是平台自己 + is_owner_model = policy_org == '0' and caller_org != '0' async with db.sqlorContext(dbname) as sor: if not ok_call: # 上游限流(429)失败 → 账号级冷却，轮转避让下一轮不再选它 if '429' in str(note or ''): _mark_cooldown(ctx.get('account_id', ''), '*') - await _release_quota(sor, ctx.get('org_id', ''), ctx.get('user_id', ''), reserved) else: - # 多退少补：实际消费 vs 预授权 - diff = charge - reserved - if diff > 0: - await sor.sqlExe( - "UPDATE llm_org_quota SET balance=balance-${d}$ " - "WHERE org_id=${o}$ AND status='active'", - {"d": diff, "o": ctx.get('org_id', '')}) - elif diff < 0: - await sor.sqlExe( - "UPDATE llm_org_quota SET balance=balance+${d}$ " - "WHERE org_id=${o}$ AND status='active'", - {"d": -diff, "o": ctx.get('org_id', '')}) - # 个人额度按实际修正（预授权按估算扣了，退还差额） - if ctx.get('user_id') and diff < 0: - await sor.sqlExe( - "UPDATE llm_user_quota SET quota_used=GREATEST(quota_used-${d}$,0) " - "WHERE user_id=${u}$ AND org_id=${o}$", - {"d": -diff, "u": ctx.get('user_id', ''), "o": ctx.get('org_id', '')}) - await sor.sqlExe("COMMIT", {}) - # 成本侧：账号余额扣减 + # 成本侧：供应商账号余额扣减（本机构/owner 模型都扣——真实钱流） if ctx.get('account_id') and cost > 0: await sor.sqlExe( "UPDATE llm_account SET balance=balance-${c}$ WHERE id=${i}$", {"c": cost, "i": ctx.get('account_id', '')}) await sor.sqlExe("COMMIT", {}) - # 双维度流水 + # 用量流水：本机构='na'(仅成本) / owner模型='pending'(待三方账) / 失败='na' + accounting_status = 'na' + if ok_call and is_owner_model: + accounting_status = 'pending' from appPublic.uniqueID import getID await sor.C('llm_usage', { 'id': getID(), - 'org_id': ctx.get('org_id', ''), + 'org_id': caller_org, 'user_id': ctx.get('user_id', ''), 'model_id': model.get('id', ''), 'account_id': ctx.get('account_id', ''), @@ -553,10 +590,11 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke 'req_tokens': int(req_tokens or 0), 'resp_tokens': int(resp_tokens or 0), 'cost': round(cost, 6), - 'charge': round(charge, 6), + 'charge': 0, # 客户应付由异步三方账计算（本机构模型不收费） 'ppid': model.get('ppid', '') or '', 'task_ref': ctx.get('task_ref', ''), 'status': 'ok' if ok_call else 'failed', + 'accounting_status': accounting_status, 'note': (note or '')[:200], }) await sor.sqlExe("COMMIT", {}) diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py new file mode 100644 index 0000000..bb3e965 --- /dev/null +++ b/pipeline_llm/inference.py @@ -0,0 +1,318 @@ +"""pipeline_llm.inference — 统一推理入口（分类照 llmage：按能力类型 catelogid 路由）。 + +设计（2026-09-04）：产线平台所有模型调用收敛到本模块： + - 进程内：agent/引擎经 pipeline_service.llm_bridge 走 HTTP 自调用 → + /pipeline_llm/api/v1/chat/completions（Bearer 短期 token）→ chat_inference() + - 运行环境：短期 token 代理（真 key 永不出进程，沿用原铁律）同一入口 + - 旧运行环境入口 /pipeline_core/api/llm_v1 委托同一实现（收敛唯一实现） + +门禁链复用 gateway.govern_resolve/govern_settle（限流/限额/主备容错/端点轮转/ +预授权/双维度记账）。治理真实失败抛 GovernError，消息真实可行动，禁止静默回退。 + +一期形态：t2t（OpenAI 兼容 chat/completions）。适配模板 llm_api_profile 的 +path/headers 决定上游请求形态（缺省按 OpenAI 兼容）；非 t2t 形态（图/视频/语音） +进来时按 capability 增加入口（对齐 llmage v1 端点分栏），不改本层结构。 +""" + +import json +import logging + +logger = logging.getLogger("pipeline_llm.inference") + +# OpenAI 兼容缺省适配（模型未挂 profile / profile 缺字段时的兜底） +_DEFAULT_CHAT_PATH = "/chat/completions" +_DEFAULT_CHAT_HEADERS = '{"Authorization": "***}"}' +_RETRYABLE_STATUS = (429, 500, 502, 503, 504) +_MAX_ATTEMPTS = 3 +_TOTAL_TIMEOUT = 300 +_CONNECT_TIMEOUT = 30 + +EST_TOKENS_PER_CHAR = 2 # 预估用量：中文约 1 字符≈0.5 token，保守按 2 字符 1 token + +from .gateway import GovernError # 复用治理层错误类型（消息真实可行动） + + +def _parse_json(s, default): + if not s: + return default + try: + v = json.loads(s) + return v if v is not None else default + except Exception: + return default + + +def _fnum(v): + try: + return float(v or 0) + except (TypeError, ValueError): + return 0.0 + + +def _texts_len(messages): + """messages 内容总长（估预授权用量用）。""" + try: + return sum(len(str(m.get('content', ''))) for m in (messages or []) + if isinstance(m, dict)) + except Exception: + return 0 + + +# ────────────────────────── 解析链 ────────────────────────── + +async def _resolve_call(sor, org_id, user_id, model_name, capability, + est_tokens, task_ref): + """门禁链 ①-⑥ + 预授权 + 加载模型/适配模板。 + + 返回 dict（govern ctx + payload 所需全部信息）。任何一级不过抛 GovernError。 + ⚠️ 本层无 __LEGACY__ 兜底——新接口只认新表；机构未配策略直接报可行动错误。 + """ + from .gateway import govern_resolve + + if not org_id: + raise GovernError('缺少机构标识（机构隔离必需）：内部调用须带 org_id') + ok, res = await govern_resolve( + org_id=org_id, user_id=user_id or '', model_name=model_name or '', + est_tokens=int(est_tokens or 0), task_ref=task_ref or '') + if not ok: + if res == '__LEGACY__': + # 未配策略/指定模型不在新表：不再有旧表兜底，报可行动错误 + if model_name: + raise GovernError( + '模型「%s」未注册到模型治理（模型注册表无此名或已停用）。' + '请在模型治理→模型注册中添加，或改用已注册模型' % model_name) + raise GovernError( + '机构 %s 未配置模型容错策略：请在模型治理→组织容错策略配置主/备模型' % org_id) + raise GovernError(res) + if not isinstance(res, dict): + raise GovernError('治理引擎返回异常结果: %s' % str(res)[:200]) + ctx = res + # 能力类型校验（防 t2t 入口调到非 t2t 模型） + model_row = ctx.get('model_row') or {} + cap = (model_row.get('capability') or 't2t').strip().lower() + if capability and cap != capability: + # 释放预授权再报错（调用不会发生） + from .gateway import govern_settle + await govern_settle(ctx, False, 0, 0, 'capability mismatch: %s!=%s' % (cap, capability)) + raise GovernError( + '模型「%s」能力为 %s，与请求能力 %s 不匹配。请按能力分类选择模型' % ( + model_row.get('name', ''), cap, capability)) + # 适配模板（决定上游 path/headers） + profile = {} + pid = model_row.get('profile_id') or '' + if pid: + recs = await sor.sqlExe( + "SELECT path, headers, request_template, response_template FROM llm_api_profile " + "WHERE id=${i}$ AND status='active' LIMIT 1", {"i": pid}) + await sor.sqlExe("COMMIT", {}) + if recs: + r = recs[0] + profile = { + 'path': getattr(r, 'path', '') or '', + 'headers': getattr(r, 'headers', '') or '', + } + # default_params 合并（调用方显式参数优先） + ctx['default_params'] = _parse_json(model_row.get('default_params'), {}) + ctx['profile'] = profile + ctx['capability'] = cap + return ctx + + +async def _pick_default_model_name(org_id, capability='t2t'): + """机构缺省模型：组织策略主模型 → 备链第一个； + 本机构无策略 → 平台 owner('0') 策略（与 govern_resolve 兜底一致）。都没有返回 ''。""" + from .gateway import _get_db + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + policy_orgs = [org_id] if (org_id and org_id != '0') else [] + policy_orgs.append('0') + recs = [] + for po in policy_orgs: + recs = await sor.sqlExe( + "SELECT primary_model_id, backup_model_ids FROM llm_org_policy " + "WHERE org_id=${o}$ AND status='active' LIMIT 1", {"o": po}) + await sor.sqlExe("COMMIT", {}) + if recs: + break + if not recs: + return '' + mids = [] + pm = getattr(recs[0], 'primary_model_id', '') or '' + if pm: + mids.append(pm) + for b in _parse_json(getattr(recs[0], 'backup_model_ids', ''), []): + if b and b not in mids: + mids.append(b) + for mid in mids: + r2 = await sor.sqlExe( + "SELECT name FROM llm_model WHERE id=${i}$ AND status='active' LIMIT 1", + {"i": mid}) + await sor.sqlExe("COMMIT", {}) + if r2: + n = getattr(r2[0], 'name', '') or '' + if n: + return n + return '' + + +# ────────────────────────── 上游调用 ────────────────────────── + +def _render_headers(tmpl, api_key): + """适配模板 headers：JSON 对象，值里 {api_key} 占位符替换为真实 key。""" + h = _parse_json(tmpl, None) + if not isinstance(h, dict) or not h: + h = _parse_json(_DEFAULT_CHAT_HEADERS, {"Authorization": "***}"}) + out = {} + for k, v in h.items(): + v = str(v) + if '{api_key}' in v: + v = v.replace('{api_key}', api_key) + out[str(k)] = v + # 兜底认证头（模板没写 Authorization 时补标准 Bearer） + if not any(str(k).lower() == 'authorization' for k in out): + out['Authorization'] = 'Bearer %s' % api_key + out.setdefault('Content-Type', 'application/json') + return out + + +async def _call_upstream_chat(ctx, payload): + """POST 上游 chat/completions（按适配模板 path/headers），带瞬时错误重试。 + + 返回上游响应 dict。失败抛 GovernError/ValueError（消息真实可行动）。 + """ + import aiohttp + import asyncio + + url = (ctx.get('api_base') or '').rstrip('/') + ( + (ctx.get('profile') or {}).get('path') or _DEFAULT_CHAT_PATH) + headers = _render_headers((ctx.get('profile') or {}).get('headers', ''), + ctx.get('api_key') or '') + timeout = int(_fnum(ctx.get('timeout')) or _TOTAL_TIMEOUT) + + upstream = dict(payload or {}) + upstream.pop('stream', None) # 一期代理不支持流式透传 + upstream['model'] = ctx.get('model_id') or upstream.get('model') or '' + for k, v in (ctx.get('default_params') or {}).items(): + upstream.setdefault(k, v) + + last = None + for attempt in range(_MAX_ATTEMPTS): + try: + async with aiohttp.ClientSession() as session: + async with session.post( + url, headers=headers, json=upstream, + timeout=aiohttp.ClientTimeout(total=timeout, connect=_CONNECT_TIMEOUT), + ) as resp: + if resp.status != 200: + text = await resp.text() + err = ValueError('上游调用失败 HTTP %d: %s' % (resp.status, text[:300])) + if resp.status in _RETRYABLE_STATUS and attempt < _MAX_ATTEMPTS - 1: + last = err + await asyncio.sleep(2 * (attempt + 1)) + continue + raise err + data = await resp.json(content_type=None) + if not isinstance(data, dict) or 'choices' not in data: + err = ValueError('上游响应缺 choices: %s' % ( + json.dumps(data, ensure_ascii=False, default=str)[:300])) + if attempt < _MAX_ATTEMPTS - 1: + last = err + await asyncio.sleep(2 * (attempt + 1)) + continue + raise err + return data + except (asyncio.TimeoutError, aiohttp.ClientError) as e: + last = e + if attempt < _MAX_ATTEMPTS - 1: + logger.warning("inference: 瞬时错误重试 %d/%d: %s", + attempt + 1, _MAX_ATTEMPTS, e) + await asyncio.sleep(2 * (attempt + 1)) + continue + raise ValueError('上游调用失败（重试 %d 次仍失败）: %s' % (_MAX_ATTEMPTS, e)) + raise last if last else ValueError('上游调用失败') + + +# ────────────────────────── 对外主入口 ────────────────────────── + +async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): + """t2t 推理：门禁链 → 上游调用 → 结算。返回上游响应 dict（OpenAI 兼容）。 + + payload: 客户端请求体（messages 必需；tools/temperature/max_tokens 透传）。 + model_name 空 → 机构策略缺省模型（主→备链第一个）。 + 失败抛 GovernError（消息真实可行动）。 + """ + from .gateway import govern_settle, _get_db + + if not isinstance(payload, dict) or not payload.get('messages'): + raise GovernError('请求体缺 messages') + model_name = (model_name or payload.get('model') or '').strip() + if not model_name: + model_name = await _pick_default_model_name(org_id, 't2t') + + est = max(_texts_len(payload.get('messages')) // EST_TOKENS_PER_CHAR, 200) + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + ctx = await _resolve_call( + sor, org_id, user_id or '', model_name, 't2t', est, task_ref) + try: + data = await _call_upstream_chat(ctx, payload) + except Exception as e: + # 失败结算：释放预授权 + failed 流水；上游 429 触发账号冷却 + note = str(e)[:200] + await govern_settle(ctx, False, 0, 0, + '429 upstream: ' + note if ' 429' in note else note) + raise GovernError(str(e)) + usage = data.get('usage') or {} + rt = int(_fnum(usage.get('prompt_tokens'))) + ct = int(_fnum(usage.get('completion_tokens'))) + await govern_settle(ctx, True, rt, ct, '' if usage else 'est') + return data + + +async def list_models(org_id, catelogid='', limit=200): + """按能力分类列模型（对齐 llmage v1/models 的分类方式）。 + + catelogid 空 = 全部能力。返回 [{'id','name','model_id','capability', + 'vendor','status','price_input','price_output'}]。 + 机构未配策略（无可用模型）返回空列表，由调用方决定是否提示。 + """ + from .gateway import _active_policy_org, _get_db + db, dbname = _get_db() + rows = [] + async with db.sqlorContext(dbname) as sor: + if not await _active_policy_org(sor, org_id or ''): + return rows + sql = ("SELECT m.id, m.name, m.vendor_model_id, m.capability, m.status, " + "m.price_input, m.price_output, 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 = {} + cap = (catelogid or '').strip().lower() + if cap: + sql += " AND m.capability=${c}$" + params['c'] = cap + sql += " ORDER BY m.capability, m.name LIMIT %d" % int(limit) + recs = await sor.sqlExe(sql, params) + await sor.sqlExe("COMMIT", {}) + for r in (recs or []): + rows.append({ + 'id': getattr(r, 'id', ''), + 'name': getattr(r, 'name', ''), + 'model_id': getattr(r, 'vendor_model_id', '') or getattr(r, 'name', ''), + 'capability': getattr(r, 'capability', '') or 't2t', + 'vendor': getattr(r, 'vendor_name', '') or '', + 'status': getattr(r, 'status', ''), + 'price_input': float(_fnum(getattr(r, 'price_input', 0))), + 'price_output': float(_fnum(getattr(r, 'price_output', 0))), + }) + return rows + + +def load_inference(): + """注册到 ServerEnv（供 dspy 端点调用）。""" + from ahserver.serverenv import ServerEnv + env = ServerEnv() + env.llm_chat_inference = chat_inference + env.llm_list_models = list_models + env.llm_pick_default_model_name = _pick_default_model_name + logger.info("[pipeline_llm] inference loaded") diff --git a/pipeline_llm/init.py b/pipeline_llm/init.py index c107db2..a6752f9 100644 --- a/pipeline_llm/init.py +++ b/pipeline_llm/init.py @@ -21,6 +21,9 @@ from .gateway import ( recharge_account, recharge_org, encrypt_api_key, decrypt_api_key, load_gateway, GovernError, ) +from .inference import chat_inference, list_models, load_inference +from .selection import load_selection +from .product_interface import load_product_category_product logger = logging.getLogger("pipeline_llm") @@ -491,6 +494,8 @@ async def llm_dashboard(params_kw): def load_llm(): """模块加载：注册全部函数到 ServerEnv。""" load_gateway() + load_inference() + load_selection() env = ServerEnv() env.create_llm_vendor = create_llm_vendor env.update_llm_vendor = update_llm_vendor @@ -504,4 +509,26 @@ def load_llm(): env.update_llm_org_policy = update_llm_org_policy env.llm_usage_query = llm_usage_query env.llm_dashboard = llm_dashboard + # 产品层标准接口（product_management.product_interface 调度） + from .product_interface import ( + get_product_display, check_product_availability, check_product_consumable, + execute_product_service, execute_product_service_stream, + calculate_product_cost, load_product_category_product, + ) + env.product_interface = { + 'module_name': 'pipeline_llm', + 'get_product_display': get_product_display, + 'check_product_availability': check_product_availability, + 'check_product_consumable': check_product_consumable, + 'execute_product_service': execute_product_service, + 'execute_product_service_stream': execute_product_service_stream, + 'calculate_product_cost': calculate_product_cost, + 'load_product_category_product': load_product_category_product, + } + env.load_product_category_product_llm = load_product_category_product logger.info("[pipeline_llm] v1.0.0 loaded — 模型治理模块就绪") + + +# 产品层按 resource_module 动态重载约定：load_{resource_module}()（resource_module='pipeline_llm'） +def load_pipeline_llm(): + return load_llm() diff --git a/pipeline_llm/product_interface.py b/pipeline_llm/product_interface.py new file mode 100644 index 0000000..b1ac7d5 --- /dev/null +++ b/pipeline_llm/product_interface.py @@ -0,0 +1,196 @@ +"""pipeline_llm.product_interface — 产线模型的产品层接口（照 llmage 样板）。 + +产品模块（product_management）通过 env.product_interface 调度本模块完成 +展示/预检/执行/计费。资源层只算「真实定价（原价）」，完全不碰折扣—— +客户折扣→售价、供应商折扣→成本，由产品层 calculate_product_cost 接 +（见 resource-productization 技能的折扣分工铁律）。 + +resource_ref_id = llm_model.id。 +""" + +import logging + +from ahserver.serverenv import ServerEnv +from sqlor.dbpools import get_sor_context + +from .gateway import _row_to_dict + +logger = logging.getLogger("pipeline_llm.product_interface") + +_MODULE_NAME = 'pipeline_llm' + + +async def _resolve_model(resource_ref_id): + """resource_ref_id → llm_model 行。返回 (model_dict, err)。""" + env = ServerEnv() + async with get_sor_context(env, _MODULE_NAME) as sor: + recs = await sor.sqlExe( + "SELECT id, name, vendor_model_id, capability, status, ppid, org_id, " + "price_input, price_output, cost_input, cost_output, description " + "FROM llm_model WHERE id=${i}$ LIMIT 1", {"i": resource_ref_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return None, '模型不存在（llm_model.id=%s）' % resource_ref_id + return _row_to_dict(recs[0]), None + + +async def get_product_display(resource_ref_id): + """展示定价信息。""" + m, err = await _resolve_model(resource_ref_id) + if err: + return {'success': False, 'message': err} + pi = float(m.get('price_input') or 0) + po = float(m.get('price_output') or 0) + pricing_text = '输入 %.4f 元/千token，输出 %.4f 元/千token' % (pi, po) if (pi or po) else '未定价' + return { + 'success': True, + 'pricing_text': pricing_text, + 'pricing_detail': { + 'price_input': pi, 'price_output': po, + 'capability': m.get('capability') or 't2t', + 'model_id': m.get('vendor_model_id') or m.get('name', ''), + }, + 'extra_info': m.get('description') or '', + } + + +async def check_product_availability(resource_ref_id, user_org_id=None): + """模型状态 + 定价方案是否就绪。""" + m, err = await _resolve_model(resource_ref_id) + if err: + return {'available': False, 'reason': err} + if (m.get('status') or '') != 'active': + return {'available': False, 'reason': '模型已停用（%s）' % m.get('name')} + if not (m.get('ppid') or ''): + return {'available': False, 'reason': '模型未挂定价方案（ppid 为空），请先配置定价'} + return {'available': True, 'reason': ''} + + +async def check_product_consumable(resource_ref_id, user_id, user_org_id): + """消费前综合预检：可用性 + 客户资金账户余额。""" + avail = await check_product_availability(resource_ref_id, user_org_id) + if not avail.get('available'): + return {'consumable': False, 'reason': avail.get('reason', ''), + 'pricing_available': False} + env = ServerEnv() + balance = None + if user_org_id and hasattr(env, 'getCustomerBalance'): + try: + acc_dbname = env.get_module_dbname('accounting') + from sqlor.dbpools import DBPools + async with DBPools().sqlorContext(acc_dbname) as sor: + balance = await env.getCustomerBalance(sor, user_org_id) + except Exception: + balance = None # accounting 未接入 → 不拦 + if balance is None: + return {'consumable': True, 'reason': '', 'pricing_available': True} + if float(balance) <= 0: + return {'consumable': False, 'reason': '客户资金账户余额不足（%.2f），请先充值' % balance, + 'pricing_available': True} + return {'consumable': True, 'reason': '', 'pricing_available': True} + + +async def execute_product_service(resource_ref_id, user_id, user_org_id, request_data): + """产品消费引擎入口不承接模型调用（防双重计费）。 + + 模型的真实消费路径是产线会话/推理入口（llm_bridge → inference）， + 该路径已完成成本侧结算并写 pending 用量流水，由异步三方账出客户账单。 + """ + return {'success': False, 'status': 'unsupported', + 'message': '产线模型消费走推理入口（会话/llm_v1），不经产品消费引擎'} + + +async def execute_product_service_stream(resource_ref_id, user_id, user_org_id, request_data): + """同 execute_product_service：不承接，防双重计费。""" + yield {'success': False, 'status': 'unsupported', + 'message': '产线模型消费走推理入口（会话/llm_v1），不经产品消费引擎', + 'done': True} + + +async def calculate_product_cost(resource_ref_id, usage_data, user_org_id=None): + """资源层真实定价（原价）。不碰折扣（折扣由产品层按产品精确计算）。 + + usage_data: {'prompt_tokens': N, 'completion_tokens': N}（或 req_tokens/resp_tokens） + """ + m, err = await _resolve_model(resource_ref_id) + if err: + return {'success': False, 'message': err} + ppid = m.get('ppid') or '' + if not ppid: + return {'success': False, 'message': '模型未挂定价方案（ppid 为空）'} + env = ServerEnv() + if not hasattr(env, 'buffered_charging'): + return {'success': False, 'message': '定价引擎（pricing 模块）未加载'} + usage = dict(usage_data or {}) + # 兼容两种键名 + if 'prompt_tokens' not in usage and 'req_tokens' in usage: + usage['prompt_tokens'] = usage['req_tokens'] + if 'completion_tokens' not in usage and 'resp_tokens' in usage: + usage['completion_tokens'] = usage['resp_tokens'] + try: + prices = await env.buffered_charging(ppid, usage) + except Exception as e: + logger.warning("pipeline_llm: 定价计算失败 ppid=%s: %s", ppid, e) + return {'success': False, 'message': '定价计算失败: %s' % e} + if prices is None: + return {'success': False, 'message': '定价计算返回空（ppid=%s）' % ppid} + amount = sum(float(getattr(p, 'amount', 0) or 0) for p in prices) + cost = sum(float(getattr(p, 'cost', 0) or 0) for p in prices) + return { + 'success': True, + 'original_amount': round(amount, 6), + 'amount': round(amount, 6), # 资源层不碰折扣，产品层再乘客户折扣 + 'cost': round(cost, 6), + 'discount': 1.0, + 'pricing_program_id': ppid, + } + + +async def load_product_category_product(parent_category_id): + """产品导入：把模型注册表包装成产品（标准化格式）。 + + 类别 = 能力类型（t2t/t2i/...，取 appcodes llm_capability 的中文名）； + 产品 = 所有 active 的模型。只读源数据返回，写入由产品模块负责。 + """ + env = ServerEnv() + async with get_sor_context(env, _MODULE_NAME) as sor: + models = await sor.sqlExe( + "SELECT id, name, vendor_model_id, capability, status, org_id, description, " + "ppid, price_input, price_output FROM llm_model " + "WHERE status='active' ORDER BY capability, name", {}) + await sor.sqlExe("COMMIT", {}) + caps = await sor.sqlExe( + "SELECT k, v FROM appcodes_kv WHERE parentid='llm_capability'", {}) + await sor.sqlExe("COMMIT", {}) + if not models: + return {'success': False, 'error': '模型注册表没有可导入的启用模型'} + models = [_row_to_dict(m) for m in models] + cap_names = {getattr(c, 'k', ''): getattr(c, 'v', '') for c in (caps or [])} + seen_caps = [] + for m in models: + cap = m.get('capability') or 't2t' + if cap not in seen_caps: + seen_caps.append(cap) + categories = [{ + 'source_id': cap, + 'name': cap_names.get(cap, cap), + 'description': '产线平台模型（%s）' % cap_names.get(cap, cap), + 'product_type': 'pipeline_llm_model', + 'product_type_title': '产线模型按量', + 'sort_order': 0, + } for cap in seen_caps] + products = [] + for m in models: + cap = m.get('capability') or 't2t' + name = m.get('name') or '' + products.append({ + 'source_category_id': cap, + 'resource_ref_id': m.get('id', ''), + 'product_code': m.get('vendor_model_id', '') or name, + 'product_name': name, + 'product_type': 'pipeline_llm_model', + 'brief_intro': m.get('description', '') or '', + 'sort_order': 0, + 'providerid': '', # 模型供应关系暂无对应供应商机构，成本侧用定价引擎 + }) + return {'success': True, 'categories': categories, 'products': products} diff --git a/pipeline_llm/selection.py b/pipeline_llm/selection.py new file mode 100644 index 0000000..8176e43 --- /dev/null +++ b/pipeline_llm/selection.py @@ -0,0 +1,147 @@ +"""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") + + +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'): + """模型选择下拉的唯一数据源（对齐 llmage 分类：capability=能力类型）。 + + 机构语义与推理链一致：本机构 + 系统级共享（org_id 空/'0'）可见。 + selected 标记：项目已设模型（sd_projects.default_model，存 name）> + 个人全局选择（pipeline_agent_settings.default_llm_id，存 id）。 + value_field: 'id'=下拉值用模型 id（AgentIO 场景）；'name'=用注册名（角色模型配置场景）。 + """ + db, dbname = _get_sor() + rows = [] + 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 + 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=''): + """模型引用解析的唯一入口：id / vendor_model_id / name → 模型注册名。 + + 机构隔离：非系统级机构只能解析「本机构 + 系统级共享」模型。 + 解析不到返回 ''（调用方决定回退/报错，禁止自行另查表）。 + """ + if not model_ref: + return '' + db, dbname = _get_sor() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT name, org_id 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 '' + return getattr(recs[0], 'name', '') or '' + + +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 + logger.info("[pipeline_llm] selection loaded") diff --git a/scripts/load_path.py b/scripts/load_path.py index 94fbb34..30578bd 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -3,19 +3,27 @@ Run from app root: py3/bin/python pkgs/pipeline-llm/scripts/load_path.py Requires app-root wrapper set_role_perm.py. + +v1 推理端点（/pipeline-llm/api/v1/*）授权给 any：运行环境/外部客户端用短期 +Bearer token 鉴权，没有登录会话。端点文件内含 verify_llm_token（security_check +AUTH_PATTERNS 认可）。其余管理页/CRUD 全部 logined。 """ import subprocess MOD = "pipeline-llm" -PATHS_ANY = [] # 本模块无免登录资源（全部需登录） +PATHS_ANY = [ + # 推理 API：Bearer 短期 token 鉴权，无登录会话 + f"/{MOD}/api/v1/chat/completions", + f"/{MOD}/api/v1/models", +] PATHS_LOGINED = [ f"/{MOD}/", f"/{MOD}/index.ui", ] -# api/ 与生成的 CRUD 目录用通配符一次覆盖 +# api/（管理）与生成的 CRUD 目录用通配符一次覆盖 PATHS_WILDCARD_LOGINED = [ f"/{MOD}/**", ] diff --git a/scripts/migrate_legacy_llm.py b/scripts/migrate_legacy_llm.py new file mode 100644 index 0000000..5d1f72f --- /dev/null +++ b/scripts/migrate_legacy_llm.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +"""pipeline-llm 初始化迁移：旧 `llm` 表 → 模型治理新表（幂等，可重跑）。 + +背景（2026-09-04）：产线平台模型调用统一收敛到模型治理模块推理 API， +旧 llm 表停用。本脚本把存量模型迁到新表并给每个机构配好容错策略， +切换后立即可用（「初始化设置一个可用的模型」）。 + +迁移规则： + 1. 旧行按 (provider, api_base) 归并为供应商（llm_vendor，端点目录一条） + 2. 同供应商下按 api_key 明文归并为账号（llm_account，AES 重加密） + —— RC4(password_key) 解密旧 key → AES(password_key) 加密入新表 + 3. 每旧行一个模型（llm_model，name 保持原名，存量引用不漂移； + capabilities 'text'→'t2t'，其余照能力类型标准） + 4. 给 '0' + users 表全部机构配组织容错策略：主模型 = 默认 t2t + （优先 qwen3.8-max，否则首个 t2t），备链 = 其余 t2t 模型 + 5. 定价初始 0（未定价模型调用不冻结额度；接入定价后再配） + +用法（应用根目录）： + py3/bin/python pkgs/pipeline-llm/scripts/migrate_legacy_llm.py [--dry-run] +""" +import asyncio +import json +import os +import sys + +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +# 应用根：本脚本位于 /pkgs/pipeline-llm/scripts/，上溯三级 +APP_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(SCRIPT_DIR))) +sys.path.insert(0, os.path.join(APP_ROOT, 'py3', 'lib', 'python3.10', 'site-packages')) +sys.path.insert(0, APP_ROOT) + +from sqlor.dbpools import DBPools +from appPublic.jsonConfig import getConfig +from appPublic.folderUtils import ProgramPath +from appPublic.uniqueID import getID + +CAP_MAP = {'text': 't2t', '': 't2t'} # 旧 capabilities → 能力类型标准 + + +def _decrypt_legacy(enc, password_key): + """旧表 api_key 是 RC4(password_key) 加密。解不出原样返回。""" + if not enc: + return '' + try: + from appPublic.rc4 import unpassword + return unpassword(enc, password_key) + except Exception: + return enc + + +def _encrypt_new(plain, password_key): + """新表 api_key 用 AES 单层加密（与模块 gateway.encrypt_api_key 一致）。""" + if not plain: + return '' + from appPublic.aes import aes_encode_b64 + return aes_encode_b64(password_key, plain) + + +async def _get_or_create_vendor(sor, provider, base_url, dry): + name = (provider or '').strip() or '未知供应商' + recs = await sor.sqlExe( + "SELECT id, endpoints FROM llm_vendor WHERE name=${n}$ AND status='active' LIMIT 1", + {"n": name}) + await sor.sqlExe("COMMIT", {}) + if recs: + vid = getattr(recs[0], 'id', '') + # 端点目录缺此 base_url 则补 + eps = [] + try: + eps = json.loads(getattr(recs[0], 'endpoints', '') or '[]') or [] + except Exception: + eps = [] + if not any((e.get('base_url') or '') == base_url for e in eps): + eps.append({'base_url': base_url, 'region': 'domestic', 'timeout': 60}) + if not dry: + await sor.sqlExe( + "UPDATE llm_vendor SET endpoints=${e}$ WHERE id=${i}$", + {"e": json.dumps(eps, ensure_ascii=False), "i": vid}) + await sor.sqlExe("COMMIT", {}) + idx = next(i for i, e in enumerate(eps) if (e.get('base_url') or '') == base_url) + return vid, idx + vid = getID() + eps = [{'base_url': base_url, 'region': 'domestic', 'timeout': 60}] + if not dry: + await sor.C('llm_vendor', { + 'id': vid, 'name': name, 'protocol': 'openai_compat', + 'endpoints': json.dumps(eps, ensure_ascii=False), + 'description': '由旧 llm 表迁移生成', 'status': 'active', 'org_id': '0', + }) + await sor.sqlExe("COMMIT", {}) + return vid, 0 + + +async def _get_or_create_account(sor, vendor_id, ep_idx, key_plain, password_key, dry): + """同供应商同 key 复用账号（余额/用量延续）。按解密后明文比对。""" + recs = await sor.sqlExe( + "SELECT id, api_key FROM llm_account WHERE vendor_id=${v}$", {"v": vendor_id}) + await sor.sqlExe("COMMIT", {}) + for r in (recs or []): + try: + from appPublic.aes import aes_decode_b64 + if aes_decode_b64(password_key, getattr(r, 'api_key', '') or '') == key_plain: + return getattr(r, 'id', '') + except Exception: + continue + aid = getID() + if not dry: + await sor.C('llm_account', { + 'id': aid, 'vendor_id': vendor_id, + 'name': '迁移账号-%s' % (key_plain[:6] if key_plain else 'nokey'), + 'api_key': _encrypt_new(key_plain, password_key), + 'endpoint_ids': json.dumps([ep_idx]), + 'balance': 0, 'total_recharge': 0, 'status': 'active', 'org_id': '0', + }) + await sor.sqlExe("COMMIT", {}) + return aid + + +async def main(dry=False): + config = getConfig(APP_ROOT, NS={'workdir': APP_ROOT, 'ProgramPath': ProgramPath()}) + DBPools(config.databases) + password_key = config.password_key or 'QRIVSRHrthhwyjy176556332' + + async with DBPools().sqlorContext('pipeline') as sor: + recs = await sor.sqlExe( + "SELECT id, name, provider, model_id, api_base, api_key, capabilities, org_id " + "FROM llm WHERE status='active' ORDER BY name", {}) + await sor.sqlExe("COMMIT", {}) + if not recs: + print("旧 llm 表无 active 模型，无需迁移") + return + + print("旧表 active 模型 %d 个，开始迁移（%s）..." % (len(recs), 'dry-run' if dry else '实写')) + created = {'vendor': 0, 'account': 0, 'model': 0, 'policy': 0} + t2t_model_ids = [] # (id, name) + for r in recs: + name = getattr(r, 'name', '') or '' + api_base = (getattr(r, 'api_base', '') or '').rstrip('/') + cap = CAP_MAP.get((getattr(r, 'capabilities', '') or '').strip().lower(), + (getattr(r, 'capabilities', '') or '').strip().lower() or 't2t') + key_plain = _decrypt_legacy(getattr(r, 'api_key', '') or '', password_key) + if key_plain and key_plain == getattr(r, 'api_key', ''): + print(" ⚠ %s: api_key 解密结果与密文相同（可能已是明文或解密失败），原样迁移" % name) + + vid, ep_idx = await _get_or_create_vendor(sor, getattr(r, 'provider', ''), api_base, dry) + await _get_or_create_account(sor, vid, ep_idx, key_plain, password_key, dry) + + exist = await sor.sqlExe("SELECT id FROM llm_model WHERE name=${n}$", {"n": name}) + await sor.sqlExe("COMMIT", {}) + if exist: + mid = getattr(exist[0], 'id', '') + print(" = 模型已存在跳过: %s" % name) + else: + mid = getID() + if not dry: + await sor.C('llm_model', { + 'id': mid, 'vendor_id': vid, + 'name': name, + 'vendor_model_id': getattr(r, 'model_id', '') or name, + 'capability': cap, 'sync_mode': 'sync', + 'profile_id': '', 'query_profile_ids': '', + 'ppid': '', + 'price_input': 0, 'price_output': 0, + 'cost_input': 0, 'cost_output': 0, + 'default_params': '', 'status': 'active', + 'description': '由旧 llm 表迁移生成', 'org_id': '0', + }) + await sor.sqlExe("COMMIT", {}) + created['model'] += 1 + print(" + 模型: %s (cap=%s vendor=%s)" % (name, cap, getattr(r, 'provider', ''))) + if cap == 't2t': + t2t_model_ids.append((mid, name)) + + if not t2t_model_ids: + print("⚠ 无 t2t 模型，跳过策略配置") + return + + # 默认主模型：优先 max 档（能力最强），否则名字序第一个 + primary = None + for mid, name in t2t_model_ids: + if 'max' in name.lower(): + primary = (mid, name) + break + primary = primary or t2t_model_ids[0] + backups = [mid for mid, name in t2t_model_ids if mid != primary[0]] + print("策略：主模型=%s，备链=%d 个" % (primary[1], len(backups))) + + # 给系统级 + 全部机构配策略（策略即开关：配了才走治理） + orgs = ['0'] + urecs = await sor.sqlExe("SELECT DISTINCT orgid FROM users WHERE orgid IS NOT NULL", {}) + await sor.sqlExe("COMMIT", {}) + for u in (urecs or []): + o = getattr(u, 'orgid', '') or '' + if o and o not in orgs: + orgs.append(o) + for org in orgs: + prec = await sor.sqlExe( + "SELECT id FROM llm_org_policy WHERE org_id=${o}$", {"o": org}) + await sor.sqlExe("COMMIT", {}) + if prec: + print(" = 策略已存在跳过: org=%s" % org) + continue + if not dry: + await sor.C('llm_org_policy', { + 'id': getID(), 'org_id': org, + 'primary_model_id': primary[0], + 'backup_model_ids': json.dumps(backups, ensure_ascii=False), + 'endpoint_pref': 'any', 'status': 'active', + }) + await sor.sqlExe("COMMIT", {}) + created['policy'] += 1 + print(" + 策略: org=%s" % org) + + print("\n迁移完成：%s" % json.dumps(created, ensure_ascii=False)) + print("注意：governance_enabled 有进程内缓存，部署迁移后须重启服务生效。") + + +if __name__ == '__main__': + dry = '--dry-run' in sys.argv + asyncio.run(main(dry)) diff --git a/wwwroot/api/v1/chat/completions.dspy b/wwwroot/api/v1/chat/completions.dspy new file mode 100644 index 0000000..815ef0f --- /dev/null +++ b/wwwroot/api/v1/chat/completions.dspy @@ -0,0 +1,79 @@ +# completions.dspy — OpenAI 兼容 LLM 推理端点（产线平台统一模型入口，分类照 llmage） +# +# URL: /pipeline_llm/api/v1/chat/completions +# OpenAI 兼容客户端配置： +# base_url = http:///pipeline_llm/api/v1 +# api_key = <短期 token>（pipeline_llm_tokens 签发，非真实模型 key） +# +# 鉴权：Authorization: Bearer *** token>，不依赖登录会话（运行环境无 session）。 +# 机构隔离：按 token 绑定的 org_id 走治理链解析真实模型与 key，真 key 不出服务进程。 +# +# 调用链：token 校验 → chat_inference（门禁链①-⑥+预授权）→ 上游调用 → 结算（双维度记账）。 +# 失败返回 OpenAI 错误结构，消息真实可行动。 + +auth = '' +try: + auth = request.headers.get('Authorization', '') or '' +except Exception: + auth = '' +if not auth: + auth = (params_kw or {}).get('api_key', '') or '' + +if not auth: + return json.dumps({"error": {"message": "缺少 Authorization Bearer token", + "type": "invalid_request_error", "code": "missing_token"}}, + ensure_ascii=False) + +payload = None +try: + payload = await request.json() +except Exception: + payload = None +if not isinstance(payload, dict): + try: + _raw = await request.text() + payload = json.loads(_raw) if _raw else None + except Exception: + payload = None +if not isinstance(payload, dict): + _pk = dict(params_kw or {}) + _msgs = _pk.get('messages') + if isinstance(_msgs, str): + try: + _msgs = json.loads(_msgs) + except Exception: + _msgs = None + payload = {"model": _pk.get('model', ''), "messages": _msgs} if _msgs else None + +if not isinstance(payload, dict) or not payload.get('messages'): + return json.dumps({"error": {"message": "请求体缺 messages", + "type": "invalid_request_error", "code": "missing_messages"}}, + ensure_ascii=False) + +# token 校验（沿用现有短期 token 机制：签发/校验/吊销/机构隔离齐全） +ok, info = await verify_llm_token(auth) +if not ok: + return json.dumps({"error": {"message": str(info), + "type": "invalid_request_error", "code": "invalid_token"}}, + ensure_ascii=False) + +org_id = info.get('org_id', '') or '' +user_id = info.get('created_by', '') or '' +# token 绑定 model_name 则强制（防运行环境越权指定贵模型）；否则用请求里的 model +task_ref = 'v1:%s' % (info.get('project_id') or '') + +try: + data = await llm_chat_inference( + org_id, user_id, payload, + model_name=(info.get('model_name') or payload.get('model') or ''), + task_ref=task_ref) + # 回填 token 用量 + try: + await record_llm_token_usage(info.get('id', ''), data.get('usage') or {}) + except Exception: + pass + return json.dumps(data, ensure_ascii=False, default=str) +except Exception as e: + return json.dumps({"error": {"message": str(e), + "type": "invalid_request_error", "code": "govern_error"}}, + ensure_ascii=False) diff --git a/wwwroot/api/v1/models.dspy b/wwwroot/api/v1/models.dspy new file mode 100644 index 0000000..1e6a81b --- /dev/null +++ b/wwwroot/api/v1/models.dspy @@ -0,0 +1,45 @@ +# models.dspy — 按能力分类列出可用模型（对齐 llmage v1/models 的分类方式） +# +# URL: /pipeline_llm/api/v1/models?catelogid=t2t +# catelogid 可选：t2t / t2i / t2v / embedding / rerank / tts / asr / i2t ... +# （能力类型标准 = 模型治理 appcodes llm_capability，新增走评审禁止野生标签） +# 不传 = 全部能力 +# +# 鉴权：Authorization: Bearer *** token>（与 chat/completions 一致）。 +# 机构隔离：只列本机构治理链可用的模型（无策略 = 空列表）。 + +auth = '' +try: + auth = request.headers.get('Authorization', '') or '' +except Exception: + auth = '' +if not auth: + auth = (params_kw or {}).get('api_key', '') or '' + +if not auth: + return json.dumps({"error": {"message": "缺少 Authorization Bearer token", + "type": "invalid_request_error", "code": "missing_token"}}, + ensure_ascii=False) + +ok, info = await verify_llm_token(auth) +if not ok: + return json.dumps({"error": {"message": str(info), + "type": "invalid_request_error", "code": "invalid_token"}}, + ensure_ascii=False) + +catelogid = (params_kw or {}).get('catelogid', '') or '' +rows = await llm_list_models(info.get('org_id', '') or '', catelogid=catelogid) + +# OpenAI 兼容输出 + 分类字段（catelogid 照 llmage 叫法，值 = 能力类型标准） +out = {"object": "list", "data": []} +for r in rows: + out["data"].append({ + "id": r.get('model_id', '') or r.get('name', ''), + "object": "model", + "name": r.get('name', ''), + "catelogid": r.get('capability', ''), + "vendor": r.get('vendor', ''), + "price_input": r.get('price_input', 0), + "price_output": r.get('price_output', 0), + }) +return json.dumps(out, ensure_ascii=False)