refactor(llm): agent调用收敛到模型治理统一推理API——llm_bridge改HTTP自调用薄客户端(短期token+本进程端口),删旧llm表直查;llm_proxy委托chat_inference;gateway/agent_loop/bug_flow模型解析与缺省模型改走治理链/统一解析,清除写死模型名

pipeline_service/agent_loop.py

@@ -107,10 +107,10 @@ async def _resolve_llm_context(sor, project_id, role, model_name=None):
       3. 用户当前模型（项目创建者 created_by → pipeline_agent_settings.default_llm_id → llm.name，缺省）
       4. RoleSpec.model_name（角色专属模型）
       5. 产线 default_model（pipelines.default_model）
-      6. 全局默认 "deepseek-v4-pro"
+      6. 留空 → 治理链按机构策略选缺省模型（主→备链），不再写死模型名
 
     org_id：从 sd_projects.org_id 读，传给 llm_bridge 做多租户隔离
-    （llm 表查询只取「本机构 + 系统级 org_id='0'」的模型）。
+    （模型治理查询只取「本机构 + 系统级 org_id='0'」的模型）。
     """
     org_id = ""
     pipeline_id = ""
@@ -159,22 +159,24 @@ async def _resolve_llm_context(sor, project_id, role, model_name=None):
         except Exception:
             pass
     if not model_name:
-        # 6. 全局默认
+        # 6. 全局缺省：留空交给治理链按机构策略选缺省模型（主模型→备链）
-        model_name = "deepseek-v4-pro"
+        model_name = ""
     return model_name, org_id
 
 
 async def _check_org_llm(sor, org_id):
-    """检测机构是否配置了 LLM 模型（严格本机构隔离，不含系统级兜底）。
+    """检测机构是否配置了 LLM 模型（查模型治理新表 llm_model；系统级兜底可见）。
 
     返回 (missing: bool, available_names: list)。org_id 为空或 '0'（系统级）时不过滤，
-    视为已配置（系统级模型对超管可用）。机构没配 llm 时角色/pm/qc 应冒泡问题暂停。
+    视为已配置（系统级模型对超管可用）。机构没配模型时角色/pm/qc 应冒泡问题暂停。
     """
     if not org_id or org_id == '0':
         return False, []
     try:
         recs = await sor.sqlExe(
-            "SELECT name FROM llm WHERE org_id=${org}$ AND status='active'", {"org": org_id})
+            "SELECT name FROM llm_model "
+            "WHERE (org_id=${org}$ OR org_id='' OR org_id='0') AND status='active'",
+            {"org": org_id})
         names = [getattr(r, "name", "") or "" for r in (recs or [])]
         return (len(names) == 0), names
     except Exception:

pipeline_service/agent_loop_v2.py

@@ -471,6 +471,7 @@ class AgentExecutor:
                 model=self.model_name,
                 temperature=0,
                 org_id=self.org_id,
+                purpose='utility',
             )
             m = _re.search(r"\[[^\]]*\]", content or "")
             if m:
@@ -885,7 +886,8 @@ class AgentExecutor:
                 {"s": self.space})
             pnames = [getattr(r, "name", "") for r in (all_recs or [])]
             classify_prompt = f"用户输入: {name}\n项目列表: {', '.join(pnames)}\n\n判断用户想要哪个项目。只回复项目名或\"不存在\"。"
-            matched = await llm_call(classify_prompt, temperature=0.0, org_id=self.org_id)
+            matched = await llm_call(classify_prompt, temperature=0.0, org_id=self.org_id,
+                                     purpose='utility')
             matched = matched.strip().strip('"').strip("'")
 
             recs2 = await sor.sqlExe(
@@ -1438,6 +1440,7 @@ class AgentExecutor:
                 f"请用3-5句话总结以下对话的关键信息：\n\n{text[:4000]}",
                 temperature=0.1,
                 org_id=self.org_id,
+                purpose='utility',
             )
             return summary[:500]
         except Exception:

pipeline_service/bug_flow.py

@@ -207,7 +207,7 @@ async def _advance_bug_states(sor):
 
 
 async def _resolve_model_org(sor, project_id):
-    """解析 bug 确认用的模型 + org_id（简版：项目缺省模型 → deepseek-v4-pro）。"""
+    """解析 bug 确认用的模型 + org_id（项目缺省模型 → 产线缺省 → 空=机构策略缺省模型）。"""
     org_id = ""
     model = ""
     try:
@@ -226,7 +226,8 @@ async def _resolve_model_org(sor, project_id):
     except Exception:
         pass
     await sor.sqlExe("COMMIT", {})
-    return (model or "deepseek-v4-pro"), (org_id or "0")
+    # 模型名为空 = 交给治理链按机构策略选缺省模型（不再写死模型名）
+    return model, (org_id or "0")
 
 
 def _parse_bug_decisions(raw):

pipeline_service/gateway.py

@@ -101,8 +101,9 @@ class Gateway:
                     ctx["default_llm_id"] = getattr(recs[0], "default_llm_id", "") or ""
                     # 单独查 llm.name（避免 JOIN 触发两表字段 collation 不一致）
                     if ctx["default_llm_id"]:
+                        # 2026-09-04 切换到模型治理新表（旧 llm 表停用）
                         llm_recs = await sor.sqlExe(
-                            "SELECT name FROM llm WHERE id=${id}$ AND status='active'",
+                            "SELECT name FROM llm_model WHERE id=${id}$ AND status='active'",
                             {"id": ctx["default_llm_id"]})
                         if llm_recs:
                             ctx["default_llm_name"] = getattr(llm_recs[0], "name", "") or ""
@@ -206,7 +207,7 @@ class Gateway:
                                          model_id: str) -> str:
         """校验前端选中的模型并持久化到项目。返回 llm.name（空 = 无效/未持久化，调用方走回退链）。
 
-        model_id 兼容两种取值：llm.id（主键）或 llm.model_id（API 模型名，如 deepseek-v4-pro）——
+        model_id 兼容两种取值：llm_model.id（主键）或 llm_model.vendor_model_id（API 模型名）——
         前端 UiCode 的 valueField 历史上两种都用过，这里统一解析，避免语义漂移。
         多租户隔离：非系统级用户只能选本机构 + 系统级共享模型。
         """
@@ -224,9 +225,10 @@ class Gateway:
                         org_id = getattr(urecs[0], "orgid", "") or ""
                 except Exception:
                     pass
+                # 2026-09-04 切换到模型治理新表（旧 llm 表停用）
                 recs = await sor.sqlExe(
-                    "SELECT id, name, org_id FROM llm "
+                    "SELECT id, name, org_id FROM llm_model "
-                    "WHERE status='active' AND (id=${m}$ OR model_id=${m}$) LIMIT 1",
+                    "WHERE status='active' AND (id=${m}$ OR vendor_model_id=${m}$) LIMIT 1",
                     {"m": model_id})
                 await sor.sqlExe("COMMIT", {})
                 if not recs:

pipeline_service/intent_classifier.py

@@ -50,6 +50,7 @@ async def intent_classify(intents: list, message: str, context: dict = None) ->
             prompt=f"{prompt}\n\n用户输入：{message}",
             model=None,
             temperature=0.2,
+            purpose='utility',
         )
         raw = raw.strip()
         if raw.startswith("```"):

pipeline_service/llm_bridge.py

@@ -1,292 +1,137 @@
 """
-LLM bridge for pipeline handlers.
+LLM bridge for pipeline handlers — 统一收敛到模型治理模块（pipeline-llm）推理 API。
 
-Provides a simple async interface for handlers to call LLM APIs.
+2026-09-04 改造：产线平台所有模型调用切换到 /pipeline-llm/api/v1（OpenAI 兼容，
-Looks up model config from the llm database table first,
+分类照 llmage）。本模块不再直查旧 `llm` 表——签名保持不变，内部改为：
-falls back to environment variables.
+    签发内部短期 token（机构隔离，真 key 不出进程）
+    → HTTP 自调用本进程 /pipeline-llm/api/v1/chat/completions
+    → 门禁链（限流/限额/主备容错/端点轮转/预授权）+ 双维度记账由端点侧执行
+
+自调用走 127.0.0.1:<本进程端口>（web/worker 进程都起 HTTP，worker 端口 9090+N）。
+每次自调用带机构/用户上下文，用量流水可归属到真实用户。
 """
 
 import json
 import logging
-import os
+import time
 
 logger = logging.getLogger("pipeline.llm_bridge")
 
-# Cache: model_name -> {api_base, api_key, model_id}
+# 内部自调用 token 缓存：(org_id, user_id, model_name) -> {token, calls, expires_at}
-_model_cache: dict = {}
+# 每次 LLM 调用都签发新 token 会让 tokens 表膨胀，故按上下文缓存复用；
+# 接近调用上限或临近过期时换新（上限/过期由签发侧强制，本地计数只是提前量）。
+_token_cache: dict = {}
+_TOKEN_MAX_LOCAL_CALLS = 400        # token max_calls=500，留 100 余量防竞态
+_TOKEN_REFRESH_BEFORE_EXPIRY = 600  # 距过期不足 10 分钟即换新
+_TOKEN_TTL_HOURS = 8
 
 
-def _decrypt_key(encrypted: str) -> str:
+async def _self_base_url():
-    """Decrypt api_key stored with password_encode. Falls back to plaintext."""
+    """本进程推理 API 基址（自调用，不出本机）。"""
-    if not encrypted:
+    port = 9090
-        return ""
     try:
-        from appPublic.rc4 import unpassword
+        from ahserver.serverenv import ServerEnv
-        from appPublic.jsonConfig import getConfig
+        p = getattr(ServerEnv(), 'port', None)
-        config = getConfig()
+        if p:
-        # 与 ahserver.globalEnv.get_password_key() 一致：password_key 为空时用默认 key
+            port = int(p)
-        key = config.password_key or 'QRIVSRHrthhwyjy176556332'
+    except (TypeError, ValueError):
-        # unpassword(code, key)：code=密文，key=密钥（之前参数顺序写反了）
+        port = 9090
-        return unpassword(encrypted, key)
     except Exception:
-        return encrypted  # already plaintext or decrypt failed
+        port = 9090
+    return "http://127.0.0.1:%d/pipeline-llm/api/v1" % port
 
 
-async def _get_model_config(model_name: str = None, org_id: str = None) -> dict:
+async def _get_internal_token(org_id, user_id, model_name):
-    """Look up model config from llm table. Returns dict with api_base, api_key, model_id.
+    """取/发内部短期 token。失败抛 ValueError（消息真实可行动）。"""
+    key = (org_id or '0', user_id or '', model_name or '')
+    now = time.time()
+    ent = _token_cache.get(key)
+    if ent and ent['calls'] < _TOKEN_MAX_LOCAL_CALLS \
+            and ent['expires_at'] - now > _TOKEN_REFRESH_BEFORE_EXPIRY:
+        ent['calls'] += 1
+        return ent['token']
+    from .llm_proxy import create_llm_token
+    ok, token = await create_llm_token(
+        org_id or '0', project_id='', task_id='', model_name=model_name or '',
+        purpose='internal_bridge', ttl_hours=_TOKEN_TTL_HOURS,
+        max_calls=500, created_by=user_id or '')
+    if not ok:
+        raise ValueError('内部 LLM token 签发失败：%s（模型治理模块未就绪或机构标识缺失）' % token)
+    _token_cache[key] = {
+        'token': token, 'calls': 1,
+        'expires_at': now + _TOKEN_TTL_HOURS * 3600,
+    }
+    return token
 
-    org_id 多租户隔离：非空时只取本机构(org_id=${org}$)的模型，不含系统级兜底；
-    org_id 为空时不过滤（向后兼容）。
-    """
-    global _model_cache
 
-    cache_key = f"{model_name or ''}:{org_id or ''}"
+async def _http_chat(payload, org_id, user_id, model_name):
-    if model_name and cache_key in _model_cache:
+    """POST 本进程推理端点。返回上游响应 dict；失败抛 ValueError（消息真实可行动）。"""
-        return _model_cache[cache_key]
+    import aiohttp
 
+    token = await _get_internal_token(org_id, user_id, model_name)
+    base = await _self_base_url()
+    headers = {"Authorization": "***" + token, "Content-Type": "application/json"}
     try:
-        from sqlor.dbpools import DBPools
+        async with aiohttp.ClientSession() as session:
-        db = DBPools()
+            async with session.post(
-        dbname = "pipeline"
+                base + "/chat/completions", headers=headers, json=payload,
-        async with db.sqlorContext(dbname) as sor:
+                timeout=aiohttp.ClientTimeout(total=330, connect=30),
-            if model_name:
+            ) as resp:
-                # name/model_id 双匹配：与 gateway._resolve_and_persist_model 语义一致——
+                text = await resp.text()
-                # 调用方传 llm.name 或 API 模型名（model_id）都能解析，避免语义漂移。
+                status = resp.status
-                sql = ("SELECT api_base, api_key, model_id FROM llm "
-                       "WHERE status='active' AND (name=${name}$ OR model_id=${name}$)")
-                params = {"name": model_name}
-                if org_id:
-                    # 与 gateway 一致：本机构模型 + 系统级共享（org_id 为空/0）
-                    sql += " AND (org_id=${org}$ OR org_id='' OR org_id='0')"
-                    params["org"] = org_id
-                sql += " LIMIT 1"
-                recs = await sor.sqlExe(sql, params)
-            else:
-                sql = "SELECT api_base, api_key, model_id, name FROM llm WHERE status='active'"
-                params = {}
-                if org_id:
-                    sql += " AND org_id=${org}$"
-                    params["org"] = org_id
-                sql += " ORDER BY id LIMIT 1"
-                recs = await sor.sqlExe(sql, params)
-            if recs:
-                r = recs[0]
-                cfg = {
-                    "api_base": getattr(r, "api_base", "") or "",
-                    "api_key": _decrypt_key(getattr(r, "api_key", "") or ""),
-                    "model_id": getattr(r, "model_id", "") or "",
-                }
-                _model_cache[cache_key] = cfg
-                return cfg
     except Exception as e:
-        logger.warning("llm_bridge: DB lookup failed: %s", e)
+        raise ValueError(
-
+            'LLM 推理端点不可达（%s/chat/completions）：%s。'
-    return {}
+            '请检查本进程服务是否正常' % (base, e))
+    try:
+        data = json.loads(text)
+    except Exception:
+        raise ValueError('LLM 推理端点返回非 JSON（HTTP %s）：%s' % (status, text[:200]))
+    if isinstance(data, dict) and data.get('error'):
+        err = data['error']
+        msg = err.get('message', '') if isinstance(err, dict) else str(err)
+        raise ValueError(msg or 'LLM 推理失败（无详情）')
+    return data
 
 
 def _no_llm_error(model_name=None, org_id=None) -> ValueError:
-    """模型不可用的错误必须真实可行动：写明模型名与原因，前端原样展示、运维据此处理。
+    """兜底错误（正常路径错误消息来自端点侧，这里只防解析异常）。"""
-
-    禁止笼统的 'No LLM API configured'——用户看到它只会卡住干等。
-    """
-    if model_name:
-        return ValueError(
-            f"模型「{model_name}」不可用：llm 表中未找到可用配置"
-            f"（name/model_id 不匹配、status 非 active"
-            + (f"，或不属于当前机构 org={org_id}" if org_id else "")
-            + "）。请在模型下拉里改选可用模型，或联系管理员在模型管理中补齐配置。"
-        )
     return ValueError(
-        "没有可用模型：当前会话未指定模型，且 llm 表无启用模型。"
+        "模型「%s」调用失败：机构 %s 未完成模型治理接入（无容错策略或模型未注册）。"
-        "请在模型下拉中选择模型，或联系管理员配置模型。"
+        "请在模型治理→组织容错策略配置主/备模型。" % (model_name or '(缺省)', org_id or '(未指定)'))
-    )
-
-
-# 瞬时错误重试：超时/连接错误/限流/服务端5xx 均重试；配置错误(无key)、鉴权/参数4xx 不重试
-_RETRYABLE_STATUS = (429, 500, 502, 503, 504)
-_LLM_MAX_ATTEMPTS = 3
-# 大上下文（角色 agent 多轮累积）生成偏慢，180s 曾触发超时（空 err=），放宽到 5 分钟；
-# connect 单独设 30s，连接建立失败能快速失败并重试，而不是干等 5 分钟。
-_LLM_TOTAL_TIMEOUT = 300
-_LLM_CONNECT_TIMEOUT = 30
-
-
-async def _post_chat_completion(url: str, headers: dict, payload: dict) -> dict:
-    """POST /chat/completions，带瞬时错误重试。返回解析后的 JSON dict。"""
-    import aiohttp
-    import asyncio
-
-    last_exc = None
-    for attempt in range(_LLM_MAX_ATTEMPTS):
-        try:
-            async with aiohttp.ClientSession() as session:
-                async with session.post(
-                    url, headers=headers, json=payload,
-                    timeout=aiohttp.ClientTimeout(total=_LLM_TOTAL_TIMEOUT, connect=_LLM_CONNECT_TIMEOUT),
-                ) as resp:
-                    if resp.status != 200:
-                        text = await resp.text()
-                        err = ValueError("LLM API error %d: %s" % (resp.status, text[:300]))
-                        if resp.status in _RETRYABLE_STATUS and attempt < _LLM_MAX_ATTEMPTS - 1:
-                            last_exc = err
-                            await asyncio.sleep(2 * (attempt + 1))
-                            continue
-                        raise err
-                    data = await resp.json()
-                    # 网关偶发返回 200 但内容无 choices（错误 JSON），视为可重试的瞬时异常；
-                    # 不处理会导致上层 llm_call_msgs_native 的 data["choices"] 抛 KeyError。
-                    if not isinstance(data, dict) or "choices" not in data:
-                        err = ValueError("LLM 响应缺 choices: %s" % json.dumps(data, ensure_ascii=False)[:300])
-                        if attempt < _LLM_MAX_ATTEMPTS - 1:
-                            last_exc = err
-                            await asyncio.sleep(2 * (attempt + 1))
-                            continue
-                        raise err
-                    return data
-        except (asyncio.TimeoutError, aiohttp.ClientError) as e:
-            last_exc = e
-            if attempt < _LLM_MAX_ATTEMPTS - 1:
-                logger.warning("llm_bridge: 瞬时错误重试 %d/%d: %s", attempt + 1, _LLM_MAX_ATTEMPTS, e)
-                await asyncio.sleep(2 * (attempt + 1))
-                continue
-            raise
-
-    raise last_exc if last_exc else ValueError("LLM call failed")
-
-
-# ────────────────────────── pipeline_llm 治理钩子 ──────────────────────────
-# 机构配置了治理（llm_org_policy / llm_model 有记录）→ 调用走门禁链（限流/限额/
-# 主备容错/端点选择/预授权），调用后按实际用量结算（双维度记账）。
-# 未配置治理 → 返回 __LEGACY__ 走下方旧 llm 表逻辑（向后兼容，现有调用零改动）。
-# 治理真实失败（限流/余额不足/候选耗尽）→ 抛 ValueError，消息真实可行动，禁止静默回退
-# （否则治理被绕过，限额形同虚设）。
-
-async def _resolve_cfg_with_govern(model, org_id, user_id, est_text):
-    """返回 (cfg, gctx)。gctx 非空 = 本次调用受治理（调用后须结算）。"""
-    gctx = None
-    try:
-        from pipeline_llm.gateway import govern_resolve
-    except ImportError:
-        govern_resolve = None
-    if govern_resolve is not None and org_id:
-        try:
-            est = max((len(est_text) if est_text else 0) // 2, 200)
-            ok, res = await govern_resolve(
-                org_id=org_id, user_id=user_id or '',
-                model_name=model or '', est_tokens=est)
-            if ok and isinstance(res, dict):
-                gctx = res
-            elif res != '__LEGACY__':
-                raise ValueError(res)
-        except ValueError:
-            raise
-        except Exception as e:
-            logger.warning("llm_bridge: 治理前置异常（回退旧表）: %s", e)
-    if gctx:
-        cfg = {"api_base": gctx["api_base"], "api_key": gctx["api_key"],
-               "model_id": gctx["model_id"]}
-        return cfg, gctx
-    cfg = await _get_model_config(model, org_id=org_id)
-    return cfg, None
-
-
-async def _settle_govern(gctx, data, est_text):
-    """调用成功后结算：优先上游真实 usage，缺失则按文本长度估算（note=est）。"""
-    if not gctx:
-        return
-    try:
-        from pipeline_llm.gateway import govern_settle
-        usage = (data or {}).get('usage') or {}
-        rt = int(usage.get('prompt_tokens') or 0)
-        ct = int(usage.get('completion_tokens') or 0)
-        note = ''
-        if not usage:
-            rt = rt or max((len(est_text) if est_text else 0) // 2, 100)
-            ct = ct or 200
-            note = 'est'
-        await govern_settle(gctx, True, rt, ct, note)
-    except Exception as e:
-        logger.warning("llm_bridge: 治理结算失败（不阻断调用）: %s", e)
-
-
-async def _settle_govern_failed(gctx, note):
-    """调用失败时结算：释放预授权，写 failed 流水。"""
-    if not gctx:
-        return
-    try:
-        from pipeline_llm.gateway import govern_settle
-        await govern_settle(gctx, False, 0, 0, str(note)[:200])
-    except Exception as e:
-        logger.warning("llm_bridge: 治理失败结算异常: %s", e)
-
-
-def _msgs_text(messages):
-    try:
-        return ' '.join(str(m.get('content', '')) for m in (messages or []) if isinstance(m, dict))
-    except Exception:
-        return ''
 
 
 async def llm_call(prompt: str, model: str = None, temperature: float = 0.7,
                    org_id: str = None, user_id: str = None) -> str:
     """Call LLM and return text response.
 
-    Backend priority:
+    统一走模型治理推理 API（门禁链 + 双维度记账）。
-    1. harnessed_agent.llm_chat (if loaded in ServerEnv)
+    org_id 为空 = 系统级（'0'），与旧语义（不过滤机构）等价。
-    2. DB llm table (api_base + api_key)
-    3. Environment variables (LLM_API_BASE, LLM_API_KEY, LLM_MODEL)
-
-    org_id 多租户隔离：非空时 llm 表查询只取「本机构 + 系统级」模型。
     """
-    # Priority 1: harnessed_agent
+    # 兼容旧优先级：harnessed_agent（若宿主加载了独立推理后端）
     try:
         from ahserver.serverenv import ServerEnv
         env = ServerEnv()
-        if hasattr(env, 'llm_chat'):
+        fn = getattr(env, 'llm_chat', None)
-            result = await env.llm_chat(prompt, model=model, temperature=temperature)
+        if callable(fn):
+            result = await fn(prompt, model=model, temperature=temperature)
             if isinstance(result, dict):
                 return result.get("content", result.get("text", str(result)))
             return str(result)
     except Exception:
         pass
 
-    # Priority 2: DB llm table（治理启用时前置走门禁链）
-    cfg, gctx = await _resolve_cfg_with_govern(model, org_id, user_id, prompt)
-    if cfg.get("api_key") and cfg.get("api_base"):
-        api_base = cfg["api_base"]
-        api_key = cfg["api_key"]
-        model_id = cfg.get("model_id") or model or "default"
-        logger.info("llm_bridge: using %s model config for %s -> %s",
-                    "governed" if gctx else "DB", model, api_base)
-    else:
-        # Priority 3: Environment variables
-        api_base = os.environ.get("LLM_API_BASE", "https://api.openai.com/v1")
-        api_key = os.environ.get("LLM_API_KEY", "")
-        model_id = model or os.environ.get("LLM_MODEL", "gpt-4o-mini")
-
-    if not api_key:
-        raise _no_llm_error(model, org_id)
-
-    import aiohttp
-
-    headers = {
-        "Authorization": f"Bearer {api_key}",
-        "Content-Type": "application/json",
-    }
     payload = {
-        "model": model_id,
+        "model": model or '',
         "messages": [{"role": "user", "content": prompt}],
         "temperature": temperature,
     }
-
+    data = await _http_chat(payload, org_id or '0', user_id or '', model or '')
-    url = api_base.rstrip("/") + "/chat/completions"
     try:
-        data = await _post_chat_completion(url, headers, payload)
+        return data["choices"][0]["message"]["content"]
-    except Exception as e:
+    except (KeyError, IndexError, TypeError) as e:
-        await _settle_govern_failed(gctx, e)
+        raise ValueError('LLM 响应缺 choices: %s' % (
-        raise
+            json.dumps(data, ensure_ascii=False, default=str)[:300])) from e
-    await _settle_govern(gctx, data, prompt)
-    return data["choices"][0]["message"]["content"]
 
 
 async def call_llm(tenant_id: str, prompt: str, model: str = None, temperature: float = 0.7) -> str:
@@ -297,32 +142,13 @@ async def call_llm(tenant_id: str, prompt: str, model: str = None, temperature:
 async def llm_call_msgs(messages: list, model: str = None, temperature: float = 0.7,
                         org_id: str = None, user_id: str = None) -> str:
     """Call LLM with full message array (system/user/assistant)."""
-    import aiohttp
+    payload = {"model": model or '', "messages": messages, "temperature": temperature}
-
+    data = await _http_chat(payload, org_id or '0', user_id or '', model or '')
-    cfg, gctx = await _resolve_cfg_with_govern(model, org_id, user_id, _msgs_text(messages))
-    if cfg.get("api_key") and cfg.get("api_base"):
-        api_base = cfg["api_base"]
-        api_key = cfg["api_key"]
-        model_id = cfg.get("model_id") or model or "default"
-    else:
-        api_base = os.environ.get("LLM_API_BASE", "https://api.openai.com/v1")
-        api_key = os.environ.get("LLM_API_KEY", "")
-        model_id = model or os.environ.get("LLM_MODEL", "gpt-4o-mini")
-
-    if not api_key:
-        raise _no_llm_error(model, org_id)
-
-    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
-    payload = {"model": model_id, "messages": messages, "temperature": temperature}
-
-    url = api_base.rstrip("/") + "/chat/completions"
     try:
-        data = await _post_chat_completion(url, headers, payload)
+        return data["choices"][0]["message"]["content"]
-    except Exception as e:
+    except (KeyError, IndexError, TypeError) as e:
-        await _settle_govern_failed(gctx, e)
+        raise ValueError('LLM 响应缺 choices: %s' % (
-        raise
+            json.dumps(data, ensure_ascii=False, default=str)[:300])) from e
-    await _settle_govern(gctx, data, _msgs_text(messages))
-    return data["choices"][0]["message"]["content"]
 
 
 async def llm_call_msgs_native(messages: list, tools: list = None, model: str = None,
@@ -334,37 +160,17 @@ async def llm_call_msgs_native(messages: list, tools: list = None, model: str =
         {"content": str, "tool_calls": [{"id","type","function":{"name","arguments"}}]}
     当模型返回 tool_calls 时，content 通常为空字符串。
     """
-    import aiohttp
+    payload = {"model": model or '', "messages": messages, "temperature": temperature}
-
-    cfg, gctx = await _resolve_cfg_with_govern(model, org_id, user_id, _msgs_text(messages))
-    if cfg.get("api_key") and cfg.get("api_base"):
-        api_base = cfg["api_base"]
-        api_key = cfg["api_key"]
-        model_id = cfg.get("model_id") or model or "default"
-    else:
-        api_base = os.environ.get("LLM_API_BASE", "https://api.openai.com/v1")
-        api_key = os.environ.get("LLM_API_KEY", "")
-        model_id = model or os.environ.get("LLM_MODEL", "gpt-4o-mini")
-
-    if not api_key:
-        raise _no_llm_error(model, org_id)
-
-    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
-    payload = {"model": model_id, "messages": messages, "temperature": temperature}
     if tools:
         payload["tools"] = tools
         payload["tool_choice"] = "auto"
-
+    data = await _http_chat(payload, org_id or '0', user_id or '', model or '')
-    url = api_base.rstrip("/") + "/chat/completions"
     try:
-        data = await _post_chat_completion(url, headers, payload)
+        msg = data["choices"][0]["message"]
-    except Exception as e:
+    except (KeyError, IndexError, TypeError) as e:
-        await _settle_govern_failed(gctx, e)
+        raise ValueError('LLM 响应缺 choices: %s' % (
-        raise
+            json.dumps(data, ensure_ascii=False, default=str)[:300])) from e
-    await _settle_govern(gctx, data, _msgs_text(messages))
-    msg = data["choices"][0]["message"]
     return {
         "content": msg.get("content") or "",
         "tool_calls": msg.get("tool_calls") or [],
     }
-

pipeline_service/llm_proxy.py

@@ -159,7 +159,7 @@ async def verify_llm_token(token):
     async with db.sqlorContext(dbname) as sor:
         recs = await sor.sqlExe(
             "SELECT id, org_id, project_id, task_id, model_name, status, expires_at, "
-            "max_calls, call_count FROM pipeline_llm_tokens WHERE token=${t}$",
+            "max_calls, call_count, created_by FROM pipeline_llm_tokens WHERE token=${t}$",
             {"t": token})
         await sor.sqlExe("COMMIT", {})
         if not recs:
@@ -187,6 +187,7 @@ async def verify_llm_token(token):
             "project_id": getattr(r, 'project_id', '') or '',
             "task_id": getattr(r, 'task_id', '') or '',
             "model_name": getattr(r, 'model_name', '') or '',
+            "created_by": '',
         }
 
 
@@ -251,13 +252,14 @@ async def _record_usage(token_id, usage):
 
 
 async def proxy_chat_completion(token, payload):
-    """OpenAI 兼容代理转发。
+    """OpenAI 兼容代理转发（2026-09-04 起委托模型治理统一推理引擎）。
 
     token  运行环境持有的短期 token（当 api_key 用）
     payload 客户端原始请求体（含 model/messages/tools/temperature 等）
 
     返回 (True, 上游响应 dict) 或 (False, 错误信息)。
-    真 api_key 由 llm_bridge 按 token 绑定的 org_id 从 llm 表解析，不下发给调用方。
+    真 api_key 由治理链解析，不下发给调用方。门禁链（限流/限额/主备容错/
+    端点轮转/预授权）+ 双维度记账在推理引擎内执行。
     """
     # 失败限速前置：窗口内鉴权失败过多直接拒绝（防 token 枚举）
     if _rate_limited(token):
@@ -272,82 +274,24 @@ async def proxy_chat_completion(token, payload):
 
     org_id = info['org_id']
     # 模型选择：token 绑定了 model_name 则强制用它（防运行环境越权指定贵模型）；
-    # 否则用请求里的 model；都没有则由 llm_bridge 取该机构第一个 active 模型。
+    # 否则用请求里的 model；都没有则由治理链取该机构策略缺省模型。
-    model_name = info.get('model_name') or payload.get('model') or None
+    model_name = info.get('model_name') or payload.get('model') or ''
 
-    # pipeline_llm 治理前置：机构配了治理 → 门禁链（限流/限额/主备容错/端点选择/预授权）；
-    # 未配置 → __LEGACY__ 走旧 llm 表。治理真实失败（限流/余额不足）→ 抛错，禁止静默回退。
-    gctx = None
     try:
-        from pipeline_llm.gateway import govern_resolve
+        from pipeline_llm.inference import chat_inference
-        _est = 200
+        data = await chat_inference(
-        try:
+            org_id, info.get('created_by', '') or '', payload,
-            _msgs = payload.get('messages') or []
+            model_name=model_name,
-            _est = max(sum(len(str(m.get('content', ''))) for m in _msgs if isinstance(m, dict)) // 2, 200)
-        except Exception:
-            _est = 200
-        gok, gres = await govern_resolve(
-            org_id=org_id, user_id='', model_name=model_name or '', est_tokens=_est,
             task_ref='proxy:%s' % (info.get('project_id') or ''))
-        if gok and isinstance(gres, dict):
-            gctx = gres
-        elif gres != '__LEGACY__':
-            return False, gres
     except ImportError:
-        gctx = None
+        return False, "模型治理模块（pipeline-llm）未安装，推理引擎不可用"
     except Exception as e:
-        logger.warning("proxy_chat_completion: 治理前置异常（回退旧表）: %s", e)
+        logger.warning("proxy_chat_completion inference failed: %s", e)
-
+        return False, str(e)[:300]
-    if gctx:
-        api_base = gctx['api_base']
-        api_key = gctx['api_key']
-        model_id = gctx['model_id']
-    else:
-        from .llm_bridge import _get_model_config
-        cfg = await _get_model_config(model_name, org_id=org_id)
-        if not (cfg.get('api_key') and cfg.get('api_base')):
-            return False, f"机构 {org_id} 未配置可用模型（模型治理/llm 表 status=active）"
-        api_base = cfg['api_base']
-        api_key = cfg['api_key']          # 只在本进程内存中使用，不返回给调用方
-        model_id = cfg.get('model_id') or model_name or 'default'
-
-    # 透传客户端参数（messages/tools/temperature/max_tokens 等），但 model 换成真实 model_id，
-    # 且不透传 stream（代理暂不支持流式）。
-    from .llm_bridge import _post_chat_completion
-    upstream = {k: v for k, v in payload.items() if k not in ('model', 'stream')}
-    upstream['model'] = model_id
-
-    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
-    url = api_base.rstrip('/') + '/chat/completions'
-    try:
-        data = await _post_chat_completion(url, headers, upstream)
-    except Exception as e:
-        logger.warning("proxy_chat_completion upstream failed: %s", e)
-        # 治理路径：失败结算（释放预授权，写 failed 流水）
-        if gctx:
-            try:
-                from pipeline_llm.gateway import govern_settle
-                await govern_settle(gctx, False, 0, 0, str(e)[:200])
-            except Exception:
-                pass
-        return False, f"上游调用失败: {str(e)[:300]}"
 
     await _record_usage(info['id'], data.get('usage'))
-    # 治理路径：按上游真实 usage 结算（双维度记账）
+    logger.info("proxy_chat_completion: org=%s project=%s model=%s ok (governed)",
-    if gctx:
+                org_id, info.get('project_id'), model_name or '(default)')
-        try:
-            from pipeline_llm.gateway import govern_settle
-            _u = data.get('usage') or {}
-            await govern_settle(
-                gctx, True,
-                int(_u.get('prompt_tokens') or 0),
-                int(_u.get('completion_tokens') or 0),
-                '' if _u else 'est')
-        except Exception as e:
-            logger.warning("proxy_chat_completion: 治理结算失败（不阻断）: %s", e)
-    logger.info("proxy_chat_completion: org=%s project=%s model=%s ok%s",
-                org_id, info.get('project_id'), model_id,
-                " (governed)" if gctx else "")
     return True, data
 
 
@@ -392,3 +336,4 @@ def load_llm_proxy():
     env.revoke_task_tokens = revoke_task_tokens
     env.proxy_chat_completion = proxy_chat_completion
     env.list_llm_tokens = list_llm_tokens
+    env.record_llm_token_usage = _record_usage