From e0c1dc751da488bbc27066c17028ae71fb9f7990 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 4 Sep 2026 17:25:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(llm):=20=5Frow=5Fto=5Fdict=E6=A0=B9?= =?UTF-8?q?=E5=9B=A0=E4=BF=AE=E5=A4=8D=E2=80=94=E2=80=94sqlor=E8=A1=8C?= =?UTF-8?q?=E6=98=AFDictObject(dict=E5=AD=90=E7=B1=BB),=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=9C=A8dict=E6=9C=AC=E4=BD=93,vars()=E5=8F=96=E5=88=B0?= =?UTF-8?q?=E7=A9=BA=E5=AE=9E=E4=BE=8B=5F=5Fdict=5F=5F=E8=87=B4=E6=B2=BB?= =?UTF-8?q?=E7=90=86=E9=93=BE=E6=A8=A1=E5=9E=8B=E5=AD=97=E6=AE=B5=E5=85=A8?= =?UTF-8?q?=E7=A9=BA(=E7=AD=96=E7=95=A5=E9=A6=96=E6=AC=A1=E6=BF=80?= =?UTF-8?q?=E6=B4=BB=E6=97=B6=E6=9A=B4=E9=9C=B2);init.py=E7=94=A8=E9=87=8F?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E5=90=8C=E4=BF=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_llm/accounting.py | 22 +++--------- pipeline_llm/gateway.py | 74 ++++---------------------------------- pipeline_llm/inference.py | 5 +-- pipeline_llm/init.py | 10 +++--- 4 files changed, 21 insertions(+), 90 deletions(-) diff --git a/pipeline_llm/accounting.py b/pipeline_llm/accounting.py index 9ee4410..acd9b92 100644 --- a/pipeline_llm/accounting.py +++ b/pipeline_llm/accounting.py @@ -124,7 +124,11 @@ async def _run_one_pass(): async def llm_usage_accounting_loop(): - """后台循环：抢分布式锁 → 出账一批 → 睡。永不抛出。""" + """后台循环：抢分布式锁 → 出账一批 → 睡。永不抛出。 + + 只由独立记账进程（app/pipeline_accounting_worker.py）运行， + 不挂进 web 服务器的事件循环（同步记账在 web，异步记账在独立进程）。 + """ logger.info("[pipeline_llm] 异步出账循环已启动（间隔 %ds）", _SCAN_INTERVAL) while True: try: @@ -143,19 +147,3 @@ async def llm_usage_accounting_loop(): except Exception as e: logger.error("pipeline_llm.accounting: 循环异常（继续）: %s", e) await asyncio.sleep(_SCAN_INTERVAL) - - -def start_accounting_loop(): - """注册到当前事件循环（load 时调用）。已有则跳过。""" - try: - loop = asyncio.get_event_loop() - except RuntimeError: - loop = None - if loop is None or not loop.is_running(): - logger.info("[pipeline_llm] 无运行中的事件循环，跳过出账循环注册") - return - for t in asyncio.all_tasks(loop): - coro = getattr(t, 'get_coro', lambda: None)() - if coro is not None and getattr(coro, '__name__', '') == 'llm_usage_accounting_loop': - return - loop.create_task(llm_usage_accounting_loop()) diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index 903549d..4fd3815 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -29,7 +29,7 @@ logger = logging.getLogger("pipeline_llm.gateway") DBNAME = "pipeline" GOVERN_REDIS_DB = 4 # 治理计数器命名空间（会话用 db3，互不干扰） _COOLDOWN_SEC = 60 # 429 冷却窗口 -_EST_TOKENS_PER_CALL = 800 # 无法估算时的单次调用预估（预授权用） +_EST_TOKENS_PER_CALL = 800 # 历史常量（预授权已废弃，保留防外部引用） # 治理开关缓存：机构是否启用治理（有策略记录或有 llm_model 记录） _govern_cache: dict = {} @@ -97,10 +97,12 @@ def _redis(): def _row_to_dict(r): + """行对象转 dict。sqlor 行是 DictObject（dict 子类），数据存在 dict 本体—— + 用 vars() 取的是空的实例 __dict__（2026-09-04 实测根因），必须 dict() 拷贝。""" try: - return {k: v for k, v in vars(r).items() if not callable(v)} - except TypeError: return dict(r) + except (TypeError, ValueError): + return {k: v for k, v in vars(r).items() if not callable(v)} def _parse_json(s, default): @@ -143,71 +145,9 @@ async def _rate_check(scope: str, scope_id: str, limit: int): return True, '' -# ────────────────────────── 门禁 ③④：额度检查 + 预授权 ────────────────────────── - -async def _reserve_quota(sor, org_id: str, user_id: str, amount: float): - """预授权：组织池 + 个人额度扣减。返回 (ok, msg, 实际冻结金额)。 - - 条件 UPDATE 防超扣；个人不足先回滚组织侧。0 元调用（未定价模型）不冻结。 - """ - if amount <= 0: - return True, '', 0.0 - # 组织池 - recs = await sor.sqlExe( - "SELECT id, balance FROM llm_org_quota WHERE org_id=${o}$ AND status='active' LIMIT 1", - {"o": org_id}) - await sor.sqlExe("COMMIT", {}) - org_quota_id = '' - if recs: - if _fnum(getattr(recs[0], 'balance', 0)) < amount: - return False, '组织余额不足：当前余额 %.4f，本次预扣 %.4f。请充值组织池（模型治理→组织限额限流）' % ( - _fnum(getattr(recs[0], 'balance', 0)), amount), 0.0 - org_quota_id = getattr(recs[0], 'id', '') - await sor.sqlExe( - "UPDATE llm_org_quota SET balance=balance-${a}$ WHERE id=${i}$ AND balance>=${a}$", - {"a": amount, "i": org_quota_id}) - await sor.sqlExe("COMMIT", {}) - # 个人额度（有记录才校验；共享池+个人上限模式） - if user_id: - recs = await sor.sqlExe( - "SELECT id, quota_limit, quota_used FROM llm_user_quota " - "WHERE user_id=${u}$ AND org_id=${o}$ AND status='active' LIMIT 1", - {"u": user_id, "o": org_id}) - await sor.sqlExe("COMMIT", {}) - if recs: - lim = _fnum(getattr(recs[0], 'quota_limit', 0)) - used = _fnum(getattr(recs[0], 'quota_used', 0)) - if lim > 0 and used + amount > lim: - if org_quota_id: # 回滚组织侧 - await sor.sqlExe( - "UPDATE llm_org_quota SET balance=balance+${a}$ WHERE id=${i}$", - {"a": amount, "i": org_quota_id}) - await sor.sqlExe("COMMIT", {}) - return False, '个人额度不足：已用 %.4f / 上限 %.4f，本次需 %.4f。请联系管理员调高个人额度' % ( - used, lim, amount), 0.0 - await sor.sqlExe( - "UPDATE llm_user_quota SET quota_used=quota_used+${a}$ WHERE id=${i}$", - {"a": amount, "i": getattr(recs[0], 'id', '')}) - await sor.sqlExe("COMMIT", {}) - return True, '', amount - - -async def _release_quota(sor, org_id: str, user_id: str, amount: float): - """释放预授权（调用失败时）。""" - if amount <= 0: - return - await sor.sqlExe( - "UPDATE llm_org_quota SET balance=balance+${a}$ WHERE org_id=${o}$ AND status='active'", - {"a": amount, "o": org_id}) - if user_id: - await sor.sqlExe( - "UPDATE llm_user_quota SET quota_used=GREATEST(quota_used-${a}$,0) " - "WHERE user_id=${u}$ AND org_id=${o}$", - {"a": amount, "u": user_id, "o": org_id}) - await sor.sqlExe("COMMIT", {}) - - # ────────────────────────── 门禁 ⑤⑥：策略选模型 + 候选轮转 ────────────────────────── +# 注：2026-09 记账重构废弃了组织池/个人额度预授权（原 _reserve_quota/_release_quota 已删除）—— +# 本机构模型只记成本侧，owner 模型走异步三方账，防滥用由限流（①②）承担。 async def _org_policy(sor, org_id: str): """组织容错策略。无记录返回 (主模型空, 辅助模型空, 备链空, 端点偏好 any)。""" diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index bb3e965..0dc3507 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -76,13 +76,14 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability, 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) + '无可用模型策略：机构 %s 未配置且平台也未配置组织容错策略。' + '请在模型治理→组织容错策略配置主/备模型' % org_id) raise GovernError(res) if not isinstance(res, dict): raise GovernError('治理引擎返回异常结果: %s' % str(res)[:200]) diff --git a/pipeline_llm/init.py b/pipeline_llm/init.py index a6752f9..2e98d08 100644 --- a/pipeline_llm/init.py +++ b/pipeline_llm/init.py @@ -444,10 +444,12 @@ async def llm_usage_query(params_kw): await sor.sqlExe("COMMIT", {}) rows = [] for r in (recs or []): - d = {} - for k, v in vars(r).items(): - if not callable(v): - d[k] = v + # sqlor 行是 DictObject（dict 子类），必须 dict() 拷贝—— + # vars() 取到的是空实例 __dict__（同 gateway._row_to_dict 根因） + try: + d = dict(r) + except (TypeError, ValueError): + d = {k: v for k, v in vars(r).items() if not callable(v)} rows.append(d) s = sums[0] if sums else None result['success'] = True