From f3f82dd4482fa92eebfa6625c1eac4e32487ac46 Mon Sep 17 00:00:00 2001 From: ymq Date: Fri, 4 Sep 2026 17:05:42 +0800 Subject: [PATCH] =?UTF-8?q?refactor(llm):=20agent=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E6=94=B6=E6=95=9B=E5=88=B0=E6=A8=A1=E5=9E=8B=E6=B2=BB=E7=90=86?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=E6=8E=A8=E7=90=86API=E2=80=94=E2=80=94llm=5F?= =?UTF-8?q?bridge=E6=94=B9HTTP=E8=87=AA=E8=B0=83=E7=94=A8=E8=96=84?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF(=E7=9F=AD=E6=9C=9Ftoken+=E6=9C=AC?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E7=AB=AF=E5=8F=A3),=E5=88=A0=E6=97=A7llm?= =?UTF-8?q?=E8=A1=A8=E7=9B=B4=E6=9F=A5;llm=5Fproxy=E5=A7=94=E6=89=98chat?= =?UTF-8?q?=5Finference;gateway/agent=5Floop/bug=5Fflow=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E8=A7=A3=E6=9E=90=E4=B8=8E=E7=BC=BA=E7=9C=81=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E6=94=B9=E8=B5=B0=E6=B2=BB=E7=90=86=E9=93=BE/=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=E8=A7=A3=E6=9E=90,=E6=B8=85=E9=99=A4=E5=86=99?= =?UTF-8?q?=E6=AD=BB=E6=A8=A1=E5=9E=8B=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_service/agent_loop.py | 16 +- pipeline_service/agent_loop_v2.py | 5 +- pipeline_service/bug_flow.py | 5 +- pipeline_service/gateway.py | 10 +- pipeline_service/intent_classifier.py | 1 + pipeline_service/llm_bridge.py | 396 +++++++------------------- pipeline_service/llm_proxy.py | 89 ++---- 7 files changed, 141 insertions(+), 381 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 5e4c91d..1dde182 100644 --- a/pipeline_service/agent_loop.py +++ b/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. 全局默认 - model_name = "deepseek-v4-pro" + # 6. 全局缺省：留空交给治理链按机构策略选缺省模型（主模型→备链） + 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: diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 85b5f27..ff26e08 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/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: diff --git a/pipeline_service/bug_flow.py b/pipeline_service/bug_flow.py index a3ee743..9f8a2fb 100644 --- a/pipeline_service/bug_flow.py +++ b/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): diff --git a/pipeline_service/gateway.py b/pipeline_service/gateway.py index 638fb6f..1a99dc4 100644 --- a/pipeline_service/gateway.py +++ b/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 " - "WHERE status='active' AND (id=${m}$ OR model_id=${m}$) LIMIT 1", + "SELECT id, name, org_id FROM llm_model " + "WHERE status='active' AND (id=${m}$ OR vendor_model_id=${m}$) LIMIT 1", {"m": model_id}) await sor.sqlExe("COMMIT", {}) if not recs: diff --git a/pipeline_service/intent_classifier.py b/pipeline_service/intent_classifier.py index 99b6c43..e2d1a8b 100644 --- a/pipeline_service/intent_classifier.py +++ b/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("```"): diff --git a/pipeline_service/llm_bridge.py b/pipeline_service/llm_bridge.py index ef95e33..8482c1a 100644 --- a/pipeline_service/llm_bridge.py +++ b/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. -Looks up model config from the llm database table first, -falls back to environment variables. +2026-09-04 改造：产线平台所有模型调用切换到 /pipeline-llm/api/v1（OpenAI 兼容， +分类照 llmage）。本模块不再直查旧 `llm` 表——签名保持不变，内部改为： + 签发内部短期 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} -_model_cache: dict = {} +# 内部自调用 token 缓存：(org_id, user_id, model_name) -> {token, calls, expires_at} +# 每次 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: - """Decrypt api_key stored with password_encode. Falls back to plaintext.""" - if not encrypted: - return "" +async def _self_base_url(): + """本进程推理 API 基址（自调用，不出本机）。""" + port = 9090 try: - from appPublic.rc4 import unpassword - from appPublic.jsonConfig import getConfig - config = getConfig() - # 与 ahserver.globalEnv.get_password_key() 一致：password_key 为空时用默认 key - key = config.password_key or 'QRIVSRHrthhwyjy176556332' - # unpassword(code, key)：code=密文，key=密钥（之前参数顺序写反了） - return unpassword(encrypted, key) + from ahserver.serverenv import ServerEnv + p = getattr(ServerEnv(), 'port', None) + if p: + port = int(p) + except (TypeError, ValueError): + port = 9090 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: - """Look up model config from llm table. Returns dict with api_base, api_key, model_id. +async def _get_internal_token(org_id, user_id, model_name): + """取/发内部短期 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 ''}" - if model_name and cache_key in _model_cache: - return _model_cache[cache_key] +async def _http_chat(payload, org_id, user_id, model_name): + """POST 本进程推理端点。返回上游响应 dict；失败抛 ValueError（消息真实可行动）。""" + 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 - db = DBPools() - dbname = "pipeline" - async with db.sqlorContext(dbname) as sor: - if model_name: - # name/model_id 双匹配：与 gateway._resolve_and_persist_model 语义一致—— - # 调用方传 llm.name 或 API 模型名（model_id）都能解析，避免语义漂移。 - 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 + async with aiohttp.ClientSession() as session: + async with session.post( + base + "/chat/completions", headers=headers, json=payload, + timeout=aiohttp.ClientTimeout(total=330, connect=30), + ) as resp: + text = await resp.text() + status = resp.status except Exception as e: - logger.warning("llm_bridge: DB lookup failed: %s", e) - - return {} + raise ValueError( + 'LLM 推理端点不可达（%s/chat/completions）：%s。' + '请检查本进程服务是否正常' % (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 表无启用模型。" - "请在模型下拉中选择模型，或联系管理员配置模型。" - ) - - -# 瞬时错误重试：超时/连接错误/限流/服务端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 '' + "模型「%s」调用失败：机构 %s 未完成模型治理接入（无容错策略或模型未注册）。" + "请在模型治理→组织容错策略配置主/备模型。" % (model_name or '(缺省)', org_id or '(未指定)')) 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: - 1. harnessed_agent.llm_chat (if loaded in ServerEnv) - 2. DB llm table (api_base + api_key) - 3. Environment variables (LLM_API_BASE, LLM_API_KEY, LLM_MODEL) - - org_id 多租户隔离：非空时 llm 表查询只取「本机构 + 系统级」模型。 + 统一走模型治理推理 API（门禁链 + 双维度记账）。 + org_id 为空 = 系统级（'0'），与旧语义（不过滤机构）等价。 """ - # Priority 1: harnessed_agent + # 兼容旧优先级：harnessed_agent（若宿主加载了独立推理后端） try: from ahserver.serverenv import ServerEnv env = ServerEnv() - if hasattr(env, 'llm_chat'): - result = await env.llm_chat(prompt, model=model, temperature=temperature) + fn = getattr(env, 'llm_chat', None) + 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, } - - url = api_base.rstrip("/") + "/chat/completions" + data = await _http_chat(payload, org_id or '0', user_id or '', model or '') try: - data = await _post_chat_completion(url, headers, payload) - except Exception as e: - await _settle_govern_failed(gctx, e) - raise - await _settle_govern(gctx, data, prompt) - return data["choices"][0]["message"]["content"] + return data["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as e: + raise ValueError('LLM 响应缺 choices: %s' % ( + json.dumps(data, ensure_ascii=False, default=str)[:300])) from e 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 - - 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" + payload = {"model": model or '', "messages": messages, "temperature": temperature} + data = await _http_chat(payload, org_id or '0', user_id or '', model or '') try: - data = await _post_chat_completion(url, headers, payload) - except Exception as e: - await _settle_govern_failed(gctx, e) - raise - await _settle_govern(gctx, data, _msgs_text(messages)) - return data["choices"][0]["message"]["content"] + return data["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as e: + raise ValueError('LLM 响应缺 choices: %s' % ( + json.dumps(data, ensure_ascii=False, default=str)[:300])) from e 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 - - 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} + payload = {"model": model or '', "messages": messages, "temperature": temperature} if tools: payload["tools"] = tools payload["tool_choice"] = "auto" - - url = api_base.rstrip("/") + "/chat/completions" + data = await _http_chat(payload, org_id or '0', user_id or '', model or '') try: - data = await _post_chat_completion(url, headers, payload) - except Exception as e: - await _settle_govern_failed(gctx, e) - raise - await _settle_govern(gctx, data, _msgs_text(messages)) - msg = data["choices"][0]["message"] + msg = data["choices"][0]["message"] + except (KeyError, IndexError, TypeError) as e: + raise ValueError('LLM 响应缺 choices: %s' % ( + json.dumps(data, ensure_ascii=False, default=str)[:300])) from e return { "content": msg.get("content") or "", "tool_calls": msg.get("tool_calls") or [], } - diff --git a/pipeline_service/llm_proxy.py b/pipeline_service/llm_proxy.py index eed2373..f0208ba 100644 --- a/pipeline_service/llm_proxy.py +++ b/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_name = info.get('model_name') or payload.get('model') or None + # 否则用请求里的 model；都没有则由治理链取该机构策略缺省模型。 + 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 - _est = 200 - try: - _msgs = payload.get('messages') or [] - _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, + from pipeline_llm.inference import chat_inference + data = await chat_inference( + org_id, info.get('created_by', '') or '', payload, + model_name=model_name, 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) - - 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]}" + logger.warning("proxy_chat_completion inference failed: %s", e) + return False, str(e)[:300] await _record_usage(info['id'], data.get('usage')) - # 治理路径：按上游真实 usage 结算（双维度记账） - if gctx: - 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 "") + logger.info("proxy_chat_completion: org=%s project=%s model=%s ok (governed)", + org_id, info.get('project_id'), model_name or '(default)') 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