fix(llm): _row_to_dict根因修复——sqlor行是DictObject(dict子类),数据在dict本体,vars()取到空实例__dict__致治理链模型字段全空(策略首次激活时暴露);init.py用量查询同修
This commit is contained in:
parent
611008cb38
commit
e0c1dc751d
@ -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())
|
||||
|
||||
@ -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)。"""
|
||||
|
||||
@ -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])
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user