feat(llm): 统一推理API(v1/chat/completions+v1/models按能力分类照llmage)+模型选择收敛层selection(下拉/引用解析唯一入口)+旧表迁移脚本;agent调用经llm_bridge HTTP自调用收敛至此

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')}}",

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

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

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)',

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

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 "
+        "SELECT primary_model_id, utility_model_id, backup_model_ids, endpoint_pref "
-        "WHERE org_id=${o}$ AND status='active' LIMIT 1", {"o": org_id})
+        "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(冻结金额),
+                            account_id, endpoint_region, reserved(恒0),
-                            model_row(定价信息), user_id, org_id, task_ref}
+                            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
-        # ⑤ 模型链（含 ⑥ 偏好）
+        # ⑤ 模型链（含 ⑥ 偏好）—— 显式 model_name 时全局查找
-        models, pref = await _model_chain(sor, org_id or '', model_name or '')
+        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
-        # ③④ 预授权（按预估用量冻结组织池+个人额度）
+        # ⑦ 组装结果（无预授权，reserved 恒 0；记账在 govern_settle 按来源分流）
-        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 解密，只在本进程内存）
         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
+    ctx      govern_resolve 成功时返回的 dict（含 policy_org_id）
-    ok_call  上游调用是否成功（失败也记账：释放预授权，写 failed 流水）
+    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 预授权
+                # 成本侧：供应商账号余额扣减（本机构/owner 模型都扣——真实钱流）
-                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", {})
-                # 成本侧：账号余额扣减
                 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", {})

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

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

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}

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

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}/**",
 ]

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__))
+# 应用根：本脚本位于 <app>/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))

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://<host>/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)

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)