From 6d0ecc5d212d141605de3a85d8437e5ee9a73a59 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 11 Sep 2026 15:29:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(wallet):=20=E8=B4=A6=E5=8F=B7=E9=92=B1?= =?UTF-8?q?=E5=8C=85=E9=97=A8=E7=A6=81+=E4=BC=9A=E8=AF=9D=E7=B2=98?= =?UTF-8?q?=E6=80=A7=E8=BD=AE=E8=AF=A2+=E5=87=BA=E8=B4=A6=E5=90=8E?= =?UTF-8?q?=E6=89=A3=E5=87=8F+=E4=BD=99=E9=A2=9D=E4=B8=8D=E8=B6=B3?= =?UTF-8?q?=E5=BE=85=E5=8A=9E(2026-09-11=E7=94=A8=E6=88=B7=E5=AE=9A?= =?UTF-8?q?=E5=A4=BA)=E2=80=94=E2=80=94=E2=91=A0=E9=97=A8=E7=A6=81:balance?= =?UTF-8?q?<=E9=98=88=E5=80=BC(params=E7=BC=BA=E7=9C=815=E5=85=83)?= =?UTF-8?q?=E4=BB=8E=E8=BD=AE=E8=AF=A2=E9=9B=86=E5=89=94=E9=99=A4,?= =?UTF-8?q?=E5=85=A8=E5=89=94=E9=99=A4=E6=8A=A5=E5=8F=AF=E8=A1=8C=E5=8A=A8?= =?UTF-8?q?=E9=94=99=E8=AF=AF;=E2=91=A1=E7=B2=98=E6=80=A7:=E5=90=8C?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E5=90=8C=E6=A8=A1=E5=9E=8B=E5=9B=BA=E5=AE=9A?= =?UTF-8?q?=E8=B4=A6=E5=8F=B7(Redis=20llm=5Fsticky=20TTL1h=E6=BB=91?= =?UTF-8?q?=E5=8A=A8=E7=BB=AD=E6=9C=9F)=E7=9C=81=E4=B8=8A=E6=B8=B8?= =?UTF-8?q?=E7=BC=93=E5=AD=98=E9=92=B1,=E7=B2=98=E6=80=A7=E5=A4=B1?= =?UTF-8?q?=E6=95=88=E8=87=AA=E5=8A=A8=E9=87=8D=E9=80=89,session=5Fid?= =?UTF-8?q?=E7=BB=8Fpayload.=5Fsession=5Fid=E9=80=8F=E4=BC=A0;=E2=91=A2?= =?UTF-8?q?=E8=BD=AE=E8=AF=A2:=E8=B7=A8=E4=BC=9A=E8=AF=9DINCR=E5=8F=96?= =?UTF-8?q?=E6=A8=A1=E6=9B=BF=E4=BB=A3=E4=BD=99=E9=A2=9D=E5=8A=A0=E6=9D=83?= =?UTF-8?q?;=E2=91=A3=E6=89=A3=E5=87=8F:=E4=BA=A7=E5=93=81=E8=AE=B0?= =?UTF-8?q?=E8=B4=A6=E6=88=90=E5=8A=9F=E5=90=8E=E6=8C=89=E6=B5=81=E6=B0=B4?= =?UTF-8?q?account=5Fid=E6=89=A3=E8=B0=83=E7=94=A8=E8=B4=A6=E5=8F=B7?= =?UTF-8?q?=E9=92=B1=E5=8C=85(supplier=5Fcost),=E4=B8=8Ecost=E5=9B=9E?= =?UTF-8?q?=E5=A1=AB=E5=90=8C=E4=BA=8B=E5=8A=A1=E5=8E=9F=E5=AD=90,cost?= =?UTF-8?q?=E2=89=A00=E5=B9=82=E7=AD=89=E8=B7=B3=E8=BF=87=E9=98=B2?= =?UTF-8?q?=E9=87=8D=E7=BD=AE=E9=87=8D=E8=B7=91=E9=87=8D=E5=A4=8D=E6=89=A3?= =?UTF-8?q?,=E5=85=81=E8=AE=B8=E9=80=8F=E6=94=AF=E4=B8=BA=E8=B4=9F;?= =?UTF-8?q?=E2=91=A4=E5=BE=85=E5=8A=9E:=E7=8A=B6=E6=80=81=E6=B4=BE?= =?UTF-8?q?=E7=94=9Fprovider(=E5=85=85=E5=80=BC=E5=88=B0=E8=B4=A6=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E6=B6=88=E5=A4=B1=E5=A4=A9=E7=84=B6=E5=8D=95=E5=8F=91?= =?UTF-8?q?)=E7=BB=99=E5=B1=9E=E4=B8=BB=E6=9C=BA=E6=9E=84operator=E8=A7=92?= =?UTF-8?q?=E8=89=B2(params=E5=8F=AF=E9=85=8D,=E6=8C=89.=E5=90=8E=E7=BC=80?= =?UTF-8?q?=E5=8A=A8=E6=80=81=E5=8C=B9=E9=85=8D)+=E8=AF=A6=E6=83=85?= =?UTF-8?q?=E5=BC=B9=E7=AA=97=E5=86=85=E7=9B=B4=E6=8E=A5=E5=85=85=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 27 ++++- pipeline_llm/accounting.py | 59 ++++++++++- pipeline_llm/gateway.py | 139 +++++++++++++++++++------ pipeline_llm/inference.py | 18 +++- pipeline_llm/init.py | 11 ++ pipeline_llm/todos.py | 114 ++++++++++++++++++++ wwwroot/api/llm_wallet_todo_popup.dspy | 119 +++++++++++++++++++++ 7 files changed, 449 insertions(+), 38 deletions(-) create mode 100644 pipeline_llm/todos.py create mode 100644 wwwroot/api/llm_wallet_todo_popup.dspy diff --git a/README.md b/README.md index 156fd85..f92a777 100644 --- a/README.md +++ b/README.md @@ -30,11 +30,34 @@ ``` ① 个人限流 → ② 组织限流 → ③ 个人额度 → ④ 组织池(预授权) -→ ⑤ 组织策略选模型(主→备) → ⑥ 账号×端点候选池(偏好过滤+余额加权+冷却避让) -→ ⑦ 调用 → ⑧ 结算(实际用量，多退少补，双维度记账) +→ ⑤ 组织策略选模型(主→备) → ⑥ 账号×端点候选池(偏好过滤+钱包门禁+冷却避让+会话粘性+轮询) +→ ⑦ 调用 → ⑧ 结算(实际用量，双维度记账) ``` 任何一级不过：快速失败 + 明确中文原因（不挂起、不静默）。 +### 账号钱包门禁与会话粘性（2026-09-11 用户定夺） + +- **钱包粒度**：1 个供应商 = 多个 llm_account（每个 APIKEY 账号一个钱包， + balance 字段；充值入口 llm_account_recharge）。accounting 的供应商记账 + 粒度不变（记账金额 = supplier_cost）。 +- **门禁**：选账号时 balance < 阈值（params `llm_wallet_balance_threshold`， + 缺省 5 元）→ 从轮询集剔除；全部剔除 → 报可行动错误（禁静默回退）。 +- **粘性**：同一会话对同一模型固定同一账号（Redis `llm_sticky:{session_id}: + {model_id}`，TTL 1h 滑动续期）——上游 KV/前缀缓存按账号(api_key)隔离， + 固定账号命中缓存省钱；粘性账号冷却/剔除时自动重选。session_id 经 + llm_bridge payload `_session_id` 透传（不进 token 缓存键）。 +- **轮询**：跨会话 Redis INCR `llm_rr:{model_id}` 取模轮流用不同账号； + Redis 不可用退化为首个可用账号（fail-open）。 +- **扣减**：产品记账（product_accounting_generic）**成功之后**，出账 worker + 按流水 account_id 扣「调用模型的那个钱包」balance -= supplier_cost，与 + cost 回填同一事务（原子）；cost≠0 = 已扣过，幂等跳过（防 accounted 行 + 重置重跑重复扣）。允许透支为负（60s 出账窗口并发超扣是真实成本，宁可 + 透支不可丢钱流/打死流水）。 +- **待办**：余额不足是状态 → todos.py provider 状态派生待办（充值到账自动 + 消失，天然单发）给属主机构 operator 角色（params `llm_wallet_todo_role` + 可配，按角色名 `.` 后缀动态匹配不锁 orgtypeid）；详情弹窗内直接充值 + （llm_wallet_todo_popup.dspy）。 + ## 集成方式（复用不改建） 模块只提供治理引擎与配置管理；实际 LLM 转发仍由宿主已有的 diff --git a/pipeline_llm/accounting.py b/pipeline_llm/accounting.py index 38ecabf..5ba1b86 100644 --- a/pipeline_llm/accounting.py +++ b/pipeline_llm/accounting.py @@ -36,7 +36,7 @@ _trace_cleanup_day = '' # trace 过期清理已跑日期（进程内每日一 async def _get_pending(sor): recs = await sor.sqlExe( - "SELECT id, org_id, user_id, model_id, cost, ppid, usages, " + "SELECT id, org_id, user_id, model_id, account_id, cost, ppid, usages, " "created_at " "FROM llm_usage WHERE accounting_status='created' AND status='SUCCEEDED' " "ORDER BY created_at LIMIT %d" % _BATCH, {}) @@ -44,6 +44,52 @@ async def _get_pending(sor): return recs or [] +async def _deduct_wallet(sor, row, supplier_cost): + """产品记账完成后扣减调用账号的钱包（2026-09-11 用户定夺）。 + + 钱包 = llm_account.balance。1 个供应商财务账户(accounting.account) : n 个 + llm_account 钱包——一个供应商可配多账号(钱包)，扣减主体按流水 account_id + 定位「**调用模型的那个钱包**」，扣减金额 = accounting 对该供应商的记账额 + （product_accounting_generic 返回的 supplier_cost，即 PAY* 分录的采购成本）。 + + 幂等守卫：流水 cost 列在出账成功时回填 supplier_cost（govern_settle 创建时 + cost=0）。cost≠0 ⟺ 本行已扣过钱包——accounted 行被手动重置回 created 重跑时 + 跳过扣减，防重复扣。（failed→created 追回流程安全：failed 行在扣减前就 + return False，从未回填 cost，重跑正常扣。） + + 允许扣成负值：调用→出账有 ~60s 窗口且门禁阈值(缺省5元)是事前闸，并发超扣 + 是真实成本，宁可透支为负（待办 provider 派生待办催充值），不可丢失钱流或把 + 已记账流水打回 failed（账务不可逆）。扣减失败只 warning 不回滚客户侧记账。 + """ + try: + prior_cost = float(getattr(row, 'cost', 0) or 0) + except (TypeError, ValueError): + prior_cost = 0.0 + if prior_cost != 0: + logger.info("pipeline_llm.accounting: 流水 %s 已回填 cost=%.6f（钱包已扣），" + "幂等跳过重复扣减", getattr(row, 'id', ''), prior_cost) + return + account_id = getattr(row, 'account_id', '') or '' + if not account_id: + return + try: + amount = float(supplier_cost or 0) + except (TypeError, ValueError): + amount = 0.0 + if amount <= 0: + return + try: + # 只发 UPDATE 不单独 COMMIT——与调用方回填 cost/accounted 的 UPDATE 同一 + # 事务提交（原子）：钱包扣减与 cost 回填要么都落要么都不落，杜绝 + # "已扣钱但 cost 未回填→重跑重复扣" 的窗口。 + await sor.sqlExe( + "UPDATE llm_account SET balance=balance-${c}$ WHERE id=${i}$", + {"c": round(amount, 6), "i": account_id}) + except Exception as e: + logger.warning("pipeline_llm.accounting: 账号钱包扣减失败（不回滚记账，对账修正）" + "account=%s cost=%.6f: %s", account_id, amount, str(e)[:160]) + + async def _find_product_id(sor, model_id): """llm_model.id → product.id（产品层导入时按 resource_ref_id 建立）。""" recs = await sor.sqlExe( @@ -140,11 +186,18 @@ async def _settle_one(sor, row): await sor.sqlExe("COMMIT", {}) logger.error("pipeline_llm.accounting: 流水 %s 出账失败: %s", row.id, msg) return False - # 回填客户应付 + 置已出账 + # 产品记账完成（2026-09-11 用户定夺）：记账成功之后扣减调用账号的钱包—— + # supplier_cost 是定价引擎算出的真实进货成本，扣 llm_account.balance（流水 + # account_id 定位"调用模型的钱包"）；同时把 cost 回填进流水（对账可查每笔 + # 钱包扣了多少）。扣减失败只 warning，不回滚客户侧账务。 + supplier_cost = float((result or {}).get('supplier_cost', 0) or 0) + await _deduct_wallet(sor, row, supplier_cost) + # 回填客户应付 + 成本侧金额 + 置已出账 await sor.sqlExe( - "UPDATE llm_usage SET charge=${c}$, accounting_status='accounted', " + "UPDATE llm_usage SET charge=${c}$, cost=${k}$, accounting_status='accounted', " "note=CONCAT(COALESCE(note,''),'|accounted:',${o}$) WHERE id=${i}$", {"c": float((result or {}).get('customer_amount', 0) or 0), + "k": round(supplier_cost, 6), "o": str((result or {}).get('orderid', ''))[:40], "i": row.id}) await sor.sqlExe("COMMIT", {}) diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index 89a6ba6..bd64898 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -30,6 +30,7 @@ logger = logging.getLogger("pipeline_llm.gateway") DBNAME = "pipeline" GOVERN_REDIS_DB = 4 # 治理计数器命名空间（会话用 db3，互不干扰） _COOLDOWN_SEC = 60 # 429 冷却窗口 +_STICKY_TTL_SEC = 3600 # 会话粘性键 TTL（滑动续期，1h 无调用自动失效） _EST_TOKENS_PER_CALL = 800 # 历史常量（预授权已废弃，保留防外部引用） # 治理开关缓存：机构是否启用治理（有策略记录或有 llm_model 记录） @@ -106,6 +107,30 @@ def _row_to_dict(r): return {k: v for k, v in vars(r).items() if not callable(v)} +async def _get_param(sor, name, default): + """读 appbase params 表配置（params_name→params_value），表不存在/无值兜底 default。""" + try: + recs = await sor.sqlExe( + "SELECT params_value FROM params WHERE params_name=${n}$ LIMIT 1", {"n": name}) + if recs: + v = str(getattr(recs[0], 'params_value', '') or '').strip() + if v: + return v + except Exception: + pass + return default + + +async def _wallet_threshold(sor) -> float: + """账号钱包余额门禁阈值（2026-09-11 用户定夺：余额<阈值的账号从轮询集剔除 + + 派待办给属主机构 operator 催充值）。params 可配，缺省 5 元。""" + v = await _get_param(sor, 'llm_wallet_balance_threshold', '5') + try: + return float(v) + except (TypeError, ValueError): + return 5.0 + + def _parse_json(s, default): if not s: return default @@ -303,8 +328,20 @@ async def _model_protocol(sor, model: dict) -> str: return (getattr(recs[0], 'protocol', '') or '').strip() if recs else '' -async def _pick_candidate(sor, model: dict, pref: str): - """构建候选并选一个：协议过滤 → 区域偏好过滤 → 冷却避让 → 余额加权。 +async def _pick_candidate(sor, model: dict, pref: str, session_id: str = ''): + """构建候选并选一个：协议过滤 → 区域偏好过滤 → 钱包余额门禁 → + 冷却避让 → 会话粘性（同会话固定账号省上游缓存钱）→ 跨会话轮询。 + + 账号钱包门禁（2026-09-11 用户定夺）：llm_account.balance 是该账号的钱包 + （充值入口 llm_account_recharge）；出账成功后按 llm_usage.account_id 扣减。 + balance < 阈值（params llm_wallet_balance_threshold，缺省 5 元）→ 账号从 + 轮询集剔除；余额状态派生待办给属主机构 operator 催充值（todos.py provider）。 + 全部剔除 → 报可行动错误（禁静默回退）。 + + 会话粘性（2026-09-11 用户定夺）：同一会话对同一模型固定同一账号——上游 + KV/前缀缓存按账号(api_key)隔离，固定账号可命中缓存省钱。粘性键 + llm_sticky:{session_id}:{model_id}（TTL 1h 滑动续期）；粘性账号失效 + （停用/冷却/余额不足）时自动重选并写回。无 session_id 的调用退化为纯轮询。 返回 (ok, result)：ok 时 result=(account_dict, endpoint_dict)；否则 result=错误信息。 """ @@ -365,25 +402,67 @@ async def _pick_candidate(sor, model: dict, pref: str): return False, ('端点偏好为「仅 %s」，但没有可用的该区域端点。按合规要求不跨端点降级——' '请配置该区域端点或调整组织策略的端点偏好' % ('国内' if 'domestic' in pref else '国际')) - # 冷却避让 + 余额加权选择 + # 账号钱包门禁（2026-09-11 用户定夺）：余额 < 阈值的账号从轮询集剔除。 + # 阈值 params 可配（llm_wallet_balance_threshold，缺省 5 元）；余额不足由 + # 待办 provider 派生待办催充值（同一事项只发一次=状态派生，充值到账自动消失）。 + wallet_min = await _wallet_threshold(sor) + low = [(acc, ep, i) for acc, ep, i in pairs if _fnum(acc.get('balance')) < wallet_min] + pairs = [(acc, ep, i) for acc, ep, i in pairs if _fnum(acc.get('balance')) >= wallet_min] + if not pairs: + names = '、'.join(sorted({str(a.get('name') or a.get('id', '')) for a, _, _ in low})) + return False, ('模型「%s」的全部候选账号钱包余额低于 %.2f 元（%s），已从轮询集剔除——' + '请为账号充值（模型治理→供应商→供应商账号→充值）或调低门禁阈值' + % (model.get('name', ''), wallet_min, names or '无')) r = _redis() - best, best_score = None, -1.0 - for acc, ep, i in pairs: - cool_key = 'llm_cool:%s:%d' % (acc.get('id', ''), i) - acct_key = 'llm_cool:%s:*' % (acc.get('id', ''),) + + def _in_cooldown(acc, i): + if r is None: + return False try: - if r is not None and (r.get(cool_key) or r.get(acct_key)): - continue # 限流冷却中（端点级或账号级） + return bool(r.get('llm_cool:%s:%d' % (acc.get('id', ''), i)) + or r.get('llm_cool:%s:*' % (acc.get('id', ''),))) + except Exception: + return False + + # 会话粘性（2026-09-11 用户定夺：同会话固定账号，省上游缓存钱）： + # 粘性账号仍在可用集且未冷却 → 直接用并续期；失效 → 落到轮询重选并写回。 + sticky_key = '' + if session_id: + sticky_key = 'llm_sticky:%s:%s' % (session_id, model.get('id', '')) + try: + if r is not None: + want = r.get(sticky_key) or '' + if isinstance(want, bytes): + want = want.decode() + want = str(want).strip() + if want: + hit = [(acc, ep, i) for acc, ep, i in pairs + if acc.get('id', '') == want and not _in_cooldown(acc, i)] + if hit: + r.set(sticky_key, want, ex=_STICKY_TTL_SEC) # 滑动续期 + acc, ep, i = hit[0] + return True, (acc, ep) + except Exception as e: + logger.warning("pipeline_llm: 会话粘性查询失败（退化为轮询）: %s", e) + # 冷却避让 + 跨会话轮询：Redis INCR 原子计数对可用集取模（同模型同供应商 + # 各会话轮流用不同账号；Redis 不可用时退化为首个可用账号，fail-open）。 + usable = [(acc, ep, i) for acc, ep, i in pairs if not _in_cooldown(acc, i)] + if not usable: + return False, ('模型「%s」的全部候选账号都在限流冷却中，请稍后重试' + % model.get('name', '')) + idx = 0 + try: + if r is not None: + idx = int(r.incr('llm_rr:%s' % model.get('id', ''))) % len(usable) + except Exception: + idx = 0 + acc, ep, i = usable[idx] + if sticky_key: + try: + if r is not None: + r.set(sticky_key, acc.get('id', ''), ex=_STICKY_TTL_SEC) except Exception: pass - score = _fnum(acc.get('balance')) - if score > best_score: - best, best_score = (acc, ep, i), score - if best is None: - return False, '模型「%s」的全部候选（账号×端点）都在限流冷却中，请稍后重试' % model.get('name', '') - acc, ep, i = best - if _fnum(acc.get('balance')) <= 0: - return False, '模型「%s」的候选账号余额均不足，请为账号充值' % model.get('name', '') return True, (acc, ep) @@ -462,7 +541,7 @@ async def _active_policy_org(sor, org_id: str) -> str: async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', est_tokens: int = 0, task_ref: str = '', purpose: str = '', - project_id: str = ''): + project_id: str = '', session_id: str = ''): """门禁链 ①-⑥。 策略来源（2026-09 商业化默认）： @@ -473,6 +552,10 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', purpose='utility'：辅助任务（分类/选择/摘要），链 = 辅助→主→备。显式 model_name 优先。 + session_id（2026-09-11 用户定夺）：会话粘性键——同一会话对同一模型固定同一 + 账号（上游 KV/前缀缓存按账号隔离，固定账号命中缓存省钱）；跨会话轮询。 + 为空（如运行环境 token 路径只绑 project_id）退化为纯轮询，不报错。 + 记账决策（2026-09）：**不做组织池预授权**—— 本机构模型只记成本侧（供应商账号扣减），不产生组织/个人消费； owner 模型三方账（客户/商户/供应商）走异步产品计费链路。 @@ -539,7 +622,7 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', last_err = '' chosen = None for m in models: - ok, cand = await _pick_candidate(sor, m, pref) + ok, cand = await _pick_candidate(sor, m, pref, session_id) if ok and isinstance(cand, tuple): chosen = (m, cand) break @@ -561,7 +644,7 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', if (m.get('capability') or 't2t').strip().lower() in (_want, _alt)] for m in models0: - ok, cand = await _pick_candidate(sor, m, pref0) + ok, cand = await _pick_candidate(sor, m, pref0, session_id) if ok and isinstance(cand, tuple): chosen = (m, cand) policy_org = '0' @@ -590,6 +673,7 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', 'org_id': org_id or '', 'task_ref': task_ref or '', 'project_id': project_id or '', + 'session_id': session_id or '', 'policy_org_id': policy_org, } @@ -624,8 +708,12 @@ async def govern_settle(ctx: dict, ok_call: bool, note: str = '', usages: dict = try: model = ctx.get('model_row') or {} # 2026-09-04：模型表不再存单价/成本四字段——成本与售价统一走 ppid 定价引擎， - # 出账由异步记账循环（accounting.py）经产品层计算；此处 cost 恒 0， - # 供应商账号余额不再按调用扣减（真实成本以定价引擎结算为准）。 + # 出账由异步记账循环（accounting.py）经产品层计算；此处 cost 恒 0。 + # 2026-09-11 用户定夺：账号钱包（llm_account.balance）扣减在**产品记账 + # 完成之后**——出账 worker 记账成功拿到 supplier_cost 再扣本流水 + # account_id 对应账号的余额（accounting.py::_deduct_wallet）；调用时点 + # 不扣（真实成本只有定价引擎能算）。流水的 account_id 就是"调用模型的 + # 钱包"指针，扣减与对账都按它定位。 cost = 0.0 caller_org = ctx.get('org_id') or '' async with db.sqlorContext(dbname) as sor: @@ -633,13 +721,6 @@ async def govern_settle(ctx: dict, ok_call: bool, note: str = '', usages: dict = # 上游限流(429)失败 → 账号级冷却，轮转避让下一轮不再选它 if '429' in str(note or ''): _mark_cooldown(ctx.get('account_id', ''), '*') - else: - # 成本侧：供应商账号余额扣减（本机构/owner 模型都扣——真实钱流） - 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", {}) from appPublic.uniqueID import getID usages_str = '' if usages: diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index 3f2a358..87ed077 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -142,12 +142,16 @@ def _texts_len(messages): # ────────────────────────── 解析链 ────────────────────────── async def _resolve_call(sor, org_id, user_id, model_name, capability, - est_tokens, task_ref, purpose='', project_id=''): + est_tokens, task_ref, purpose='', project_id='', + session_id=''): """门禁链 ①-⑥ + 加载模型/适配模板。 purpose='utility'：辅助任务（分类/选择/摘要），模型链为 辅助→主→备 （需机构策略配置 utility_model_id；未配则等同普通链）。 + session_id：会话粘性（2026-09-11 用户定夺：同会话固定账号省上游缓存钱， + 跨会话轮询）；空=纯轮询。 + 返回 dict（govern ctx + payload 所需全部信息）。任何一级不过抛 GovernError。 ⚠️ 本层无 __LEGACY__ 兜底——新接口只认新表；机构未配策略直接报可行动错误。 """ @@ -158,7 +162,7 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability, 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 '', purpose=purpose or '', - project_id=project_id or '') + project_id=project_id or '', session_id=session_id or '') if not ok: if res == '__LEGACY__': # 全平台无策略/指定模型不在新表：不再有旧表兜底，报可行动错误 @@ -842,10 +846,12 @@ async def _async_inference(ctx, payload, req_timeout): # ────────────────────────── 对外主入口 ────────────────────────── async def chat_inference(org_id, user_id, payload, model_name='', task_ref='', - project_id=''): + project_id='', session_id=''): # project_id（2026-09-10 用户定夺）：token 绑定的项目 ID 透传落 llm_usage.project_id， # 非项目调用不传（govern_settle 兜底哨兵'0'），支撑按项目统计费用。 # 调用方（llm_proxy/v1 dspy）负责传入。 + # session_id（2026-09-11 用户定夺）：会话粘性账号键——同会话固定账号省上游 + # 缓存钱；调用方经 payload._session_id 或显式参数传入（llm_bridge 透传）。 """统一推理：门禁链 → 上游调用 → 结算。返回上游响应 dict（OpenAI 兼容形态）。 payload: 客户端请求体（messages 必需；tools/temperature/max_tokens 透传； @@ -871,13 +877,17 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref='', purpose = (payload.pop('_purpose', '') or '') # 单次调用超时覆盖（长文本提取等慢任务）：剥离不透传，上限 900 秒 req_timeout = int(_fnum(payload.pop('_timeout', 0)) or 0) + # 会话粘性键（2026-09-11 用户定夺）：payload._session_id 优先于显式参数 + # （llm_bridge 经 payload 透传，与 _purpose/_timeout 同通道）；剥离不上行 + _sid = (payload.pop('_session_id', '') or '') + session_id = str(_sid or session_id or '') 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, '', est, task_ref, purpose, - project_id=project_id or '') + project_id=project_id or '', session_id=session_id) # 调用批次 ID（2026-09-10 用户定夺：上行下行原文唯一落盘 + llm_usage 关联）： # 一次 chat_inference = N 次上游往返（重试/异步轮询），trace 按 call_id+seq 组织， # llm_usage.call_id 同值 → 流水可反查全部原文（trace.record_roundtrip / llm_call_trace） diff --git a/pipeline_llm/init.py b/pipeline_llm/init.py index e794df5..94d55e7 100644 --- a/pipeline_llm/init.py +++ b/pipeline_llm/init.py @@ -828,6 +828,17 @@ def load_pipeline_llm(): 'load_product_category_product': load_product_category_product, } env.load_product_category_product_llm = load_product_category_product + # 平台待办 provider（软注册，2026-09-11：账号钱包余额不足 → 属主机构 + # operator 角色催充值待办；pipeline-service 没装/没钩子时只告警不崩， + # 与 ticket 模块同款模式） + try: + from pipeline_service.human_task_capability import register_todo_provider + from .todos import list_wallet_todos + register_todo_provider(list_wallet_todos) + logger.info("[pipeline_llm] wallet todo provider registered") + except ImportError: + logger.warning("[pipeline_llm] pipeline_service.register_todo_provider 不可用，" + "钱包余额待办不会出现在平台待办（模块可独立运行）") logger.info("[pipeline_llm] v1.0.0 loaded — 模型治理模块就绪") diff --git a/pipeline_llm/todos.py b/pipeline_llm/todos.py new file mode 100644 index 0000000..3e03320 --- /dev/null +++ b/pipeline_llm/todos.py @@ -0,0 +1,114 @@ +# -*- coding: utf-8 -*- +"""pipeline_llm.todos — 平台待办 provider：账号钱包余额不足（2026-09-11 用户定夺）。 + +钱包余额改变的两个目的（用户原话）： + 1) 哪个钱包什么时候该充值了 → 本 provider 派生待办给属主机构 operator 催充值 + 2) 哪个钱包不能再调用 llm 了 → gateway._pick_candidate 门禁剔除（balance<阈值） + +待办由**状态派生**（不建通知表）：balance < 阈值 的 active 账号即待办；充值 +到账余额回升 → 待办自动消失。「同一事项只发一次」不变量天然满足（human-task- +issuance-gate 模式，与 ticket.todos 同款）。 + +接收人：账号属主机构（llm_account.org_id，空则回溯 llm_vendor.org_id）中持有 +operator 角色的用户。角色名 params 可配（llm_wallet_todo_role，缺省 operator） +——角色/机构类型运行时可动态增删，禁硬编码（用户铁律）；roles 为 +「{orgtypeid}.{name}」格式，按 . 后的名字匹配，不锁死 orgtypeid。 + +provider 签名（pipeline_service.human_task_capability.register_todo_provider）： + async fn(user_id, roles, limit) -> [todo dict] +""" + +import logging + +from .gateway import _get_db, _get_param + +logger = logging.getLogger("pipeline_llm.todos") + +_DEFAULT_ROLE = 'operator' + + +async def _wallet_threshold(sor) -> float: + v = await _get_param(sor, 'llm_wallet_balance_threshold', '5') + try: + return float(v) + except (TypeError, ValueError): + return 5.0 + + +def _match_role(roles, want): + """roles=['reseller.operator',...]；匹配 . 后的角色名（orgtypeid 运行时动态）。""" + for r in (roles or []): + name = str(r).split('.')[-1].strip().lower() + if name == want: + return True + return False + + +async def list_wallet_todos(user_id, roles=None, limit=100): + """平台待办聚合器调用的 provider。roles 缺省时自查。""" + if not user_id: + return [] + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + threshold = await _wallet_threshold(sor) + want_role = str(await _get_param( + sor, 'llm_wallet_todo_role', _DEFAULT_ROLE)).strip().lower() + # 用户机构（属主机构匹配用） + recs = await sor.sqlExe( + "SELECT orgid FROM users WHERE id=${u}$ LIMIT 1", {"u": str(user_id)}) + await sor.sqlExe("COMMIT", {}) + user_org = str(getattr(recs[0], 'orgid', '') or '') if recs else '' + if roles is None: + rrecs = await sor.sqlExe( + "SELECT r.orgtypeid, r.name FROM userrole ur JOIN role r ON ur.roleid=r.id " + "WHERE ur.userid=${u}$", {"u": str(user_id)}) + await sor.sqlExe("COMMIT", {}) + roles = [] + for r in (rrecs or []): + o = getattr(r, 'orgtypeid', '') or '' + n = getattr(r, 'name', '') or '' + if o and n: + roles.append(f"{o}.{n}") + if not _match_role(roles, want_role): + return [] + # 余额不足的 active 账号（属主机构=用户机构；账号 org_id 空回溯供应商 org_id） + recs = await sor.sqlExe( + "SELECT a.id, a.name, a.balance, a.org_id, a.vendor_id, v.name AS vendor_name, " + "COALESCE(NULLIF(a.org_id,''), v.org_id, '0') AS owner_org " + "FROM llm_account a LEFT JOIN llm_vendor v ON v.id=a.vendor_id " + "WHERE a.status='active' AND a.balance < ${t}$ " + "ORDER BY a.balance ASC LIMIT ${lim}$", + {"t": threshold, "lim": int(limit) * 4}) + await sor.sqlExe("COMMIT", {}) + + todos = [] + seen = 0 + for r in (recs or []): + owner_org = str(getattr(r, 'owner_org', '') or '0') + if owner_org != (user_org or '0'): + continue # 只发给属主机构的人 + if seen >= int(limit): + break + seen += 1 + acc_id = str(getattr(r, 'id', '') or '') + balance = float(getattr(r, 'balance', 0) or 0) + acc_name = str(getattr(r, 'name', '') or acc_id) + vendor_name = str(getattr(r, 'vendor_name', '') or '') + todos.append({ + 'id': 'llm_wallet_' + acc_id, + 'source': 'llm_wallet', + 'task_type': 'llm_wallet_recharge', + 'badge': '钱包充值', + 'title': '【钱包充值】%s%s' % ( + ('%s·' % vendor_name) if vendor_name else '', acc_name), + 'description': '模型供应商账号钱包余额 %.2f 元，低于门禁阈值 %.2f 元——' + '该账号已从模型调用轮询集剔除（充值到账后自动恢复并关闭本待办）' + % (balance, threshold), + 'account_id': acc_id, + 'balance': round(balance, 4), + 'threshold': threshold, + 'project_id': '', + 'status': 'pending', + 'created_at': None, + }) + return todos diff --git a/wwwroot/api/llm_wallet_todo_popup.dspy b/wwwroot/api/llm_wallet_todo_popup.dspy new file mode 100644 index 0000000..67e32a4 --- /dev/null +++ b/wwwroot/api/llm_wallet_todo_popup.dspy @@ -0,0 +1,119 @@ +# llm_wallet_todo_popup.dspy — 钱包余额不足待办详情弹窗（2026-09-11 用户定夺） +# 平台待办 source='llm_wallet' 的「查看内容并处理」入口。 +# 入参: id=llm_wallet_ 或 account_id=<账号id> +# 展示账号钱包状态 + 充值表单（金额/备注）→ POST llm_account_recharge.dspy； +# 充值到账余额回升 → 待办由状态派生自动消失（provider 不再返回）。 + +import json as _json + +uid = await get_user() +if not uid: + return {"widgettype": "Message", "options": {"title": "未登录", "message": "请先登录后查看待办"}} + +pk = params_kw or {} +acc_id = str(pk.get('account_id') or '').strip() +if not acc_id: + raw = str(pk.get('id') or '').strip() + if raw.startswith('llm_wallet_'): + acc_id = raw[len('llm_wallet_'):] + else: + acc_id = raw +if not acc_id: + return {"widgettype": "Message", "options": {"title": "打开失败", "message": "缺少账号 id"}} + +user_org = str(await get_userorgid() or '0') + +dbname = get_module_dbname('pipeline-llm') +async with DBPools().sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT a.id, a.name, a.balance, a.total_recharge, a.status, a.org_id, " + "v.name AS vendor_name, COALESCE(NULLIF(a.org_id,''), v.org_id, '0') AS owner_org " + "FROM llm_account a LEFT JOIN llm_vendor v ON v.id=a.vendor_id " + "WHERE a.id=${i}$ LIMIT 1", {"i": acc_id}) + await sor.sqlExe("COMMIT", {}) + trecs = await sor.sqlExe( + "SELECT params_value FROM params WHERE params_name='llm_wallet_balance_threshold' LIMIT 1", {}) + await sor.sqlExe("COMMIT", {}) + +if not recs: + return {"widgettype": "Message", "options": {"title": "打开失败", "message": "账号不存在或已删除"}} + +row = recs[0] +owner_org = str(row['owner_org'] or '0') +# 机构隔离：平台业主('0')可看全部；否则只看本属主机构的账号 +if user_org != '0' and owner_org not in ('', user_org): + return {"widgettype": "Message", "options": {"title": "无权访问", "message": "该账号属于其他机构"}} + +balance = float(row['balance'] or 0) +threshold = 5.0 +try: + if trecs and str(trecs[0]['params_value'] or '').strip(): + threshold = float(str(trecs[0]['params_value']).strip()) +except Exception: + pass +acc_name = str(row['name'] or acc_id) +vendor_name = str(row['vendor_name'] or '') + +head_md = ['## 供应商账号钱包', '', + '**账号**：' + (('【%s】' % vendor_name) if vendor_name else '') + acc_name, + '**当前余额**：%.4f 元 ｜ **门禁阈值**：%.2f 元 ｜ **累计充值**：%.4f 元' + % (balance, threshold, float(row['total_recharge'] or 0)), + '**账号状态**：' + str(row['status'] or ''), ''] +if balance < threshold: + head_md += ['> ⚠️ 余额低于门禁阈值——该账号已从模型调用轮询集剔除，' + '调用其下模型将顺延到其他账号；全部账号不足时调用报可行动错误。', + '> 充值到账后待办自动消失、账号自动恢复轮询。', ''] +else: + head_md += ['> ✅ 余额已恢复到阈值以上，账号在轮询集中正常可用，本待办即将自动关闭。', ''] + +recharge_url = entire_url('/pipeline-llm/api/llm_account_recharge.dspy') + + +def _read_input_js(wid): + return ("var cw=bricks.getWidgetById(" + _json.dumps(wid) + ",bricks.app);var cv='';" + "if(cw){cv=(typeof cw.resultValue==='function')?cw.resultValue():" + "((cw.dom_element&&cw.dom_element.value)||'');}" + "cv=(cv===null||cv===undefined)?'':String(cv).trim();") + + +_tail = ("var pw=bricks.getWidgetById('llm_wallet_todo_pw',bricks.app);" + "if(pw&&pw.destroy){pw.destroy();}" + "if(window.refreshTodo){window.refreshTodo();}" + "if(d&&d.success){var mo=new bricks.Message({title:'充值成功',message:d.message||'钱包已充值'});mo.open();}" + "else{var mf=new bricks.Message({title:'充值失败',message:(d&&(d.message||d.error))||'充值失败'});mf.open();}") + +do_js = (_read_input_js('llm_wallet_amt') + + "var amt=parseFloat(cv);" + + "if(isNaN(amt)||amt<=0){new bricks.Message({title:'请填写金额',message:'充值金额必须是大于 0 的数字'}).open();return;}" + + _read_input_js('llm_wallet_note') + + "var r=await fetch(" + _json.dumps(recharge_url) + ",{method:'POST'," + "headers:{'Content-Type':'application/json'}," + "body:JSON.stringify({account_id:" + _json.dumps(acc_id) + ",amount:amt,note:cv})});" + "var d=await r.json();" + _tail) + +sub = [ + {"widgettype": "MdWidget", "options": {"mdtext": '\n'.join(head_md), "width": "100%"}}, + {"widgettype": "HBox", + "options": {"width": "100%", "gap": "8px", "alignItems": "center"}, + "subwidgets": [ + {"widgettype": "Text", "options": {"text": "充值金额(元)：", "cfontsize": 0.9, "color": "#334155"}}, + {"widgettype": "UiText", "id": "llm_wallet_amt", + "options": {"name": "llm_wallet_amt", "placeholder": "如 100.00", "width": "160px"}}, + {"widgettype": "UiText", "id": "llm_wallet_note", + "options": {"name": "llm_wallet_note", "placeholder": "备注（可选，如银行流水号）", + "flex": "1 1 auto"}}, + {"widgettype": "Button", + "options": {"label": "💰 确认充值", "css": "primary small"}, + "binds": [{"wid": "self", "event": "click", "actiontype": "script", + "target": "self", "script": do_js}]}]}] + +return { + "widgettype": "PopupWindow", + "id": "llm_wallet_todo_pw", + "options": {"title": "钱包充值 — " + acc_name, "width": "52%", "height": "46%", + "auto_open": True, "resizable": True}, + "subwidgets": [{ + "widgettype": "VScrollPanel", + "options": {"css": "filler", "width": "100%", "height": "100%", "padding": "14px"}, + "subwidgets": sub}] +}