diff --git a/models/llm_usage.json b/models/llm_usage.json index 0816577..bb9f89a 100644 --- a/models/llm_usage.json +++ b/models/llm_usage.json @@ -77,11 +77,11 @@ }, { "name": "accounting_status", - "title": "出账状态(na仅成本/pending待三方账/accounted已出账)", + "title": "记账状态(created已创建待记账/accounted记账成功/failed记账失败)", "type": "str", "length": 20, "nullable": "no", - "default": "'na'" + "default": "'created'" }, { "name": "ppid", diff --git a/mysql.ddl.sql b/mysql.ddl.sql index 67ea044..2c279bc 100644 --- a/mysql.ddl.sql +++ b/mysql.ddl.sql @@ -121,7 +121,7 @@ CREATE TABLE IF NOT EXISTS llm_usage ( `resp_tokens` int NOT NULL DEFAULT 0 comment '响应token', `cost` decimal(12,6) NOT NULL DEFAULT 0 comment '成本金额(账号侧)', `charge` decimal(12,6) NOT NULL DEFAULT 0 comment '客户应付(异步三方账计算,本机构模型恒0)', - `accounting_status` varchar(20) NOT NULL DEFAULT 'na' comment '出账状态(本机构模型仅成本=na/owner模型待三方账=pending/已出账=accounted)', + `accounting_status` varchar(20) NOT NULL DEFAULT 'created' comment '记账状态(三态:created已创建待记账/accounted记账成功/failed记账失败)', `ppid` varchar(32) comment '定价项目ID', `task_ref` varchar(100) comment '调用来源', `status` varchar(20) NOT NULL DEFAULT 'ok' comment '状态(ok/failed/recharge)', diff --git a/pipeline_llm/accounting.py b/pipeline_llm/accounting.py index 064772c..dc46870 100644 --- a/pipeline_llm/accounting.py +++ b/pipeline_llm/accounting.py @@ -1,12 +1,13 @@ """pipeline_llm.accounting — 异步出账循环（owner 模型三方账）。 -扫 llm_usage 中 accounting_status='pending' 的用量流水（owner 模型、调用方≠平台）， -逐条映射模型→产品，调产品层 product_accounting_generic 出三方账 -（客户付/商户营收/供应商成本，客户折扣精确到产品），成功置 'accounted'。 +扫 llm_usage 中 accounting_status='created' 且 status='ok' 的用量流水， +逐条映射模型→产品，调产品层 product_accounting_generic 记账 +（产品层内部按 is_self_use 分流：自用只记 PAY* 采购成本；跨机构另记 PAY 客户应付， +折扣精确到产品），成功置 'accounted'，失败置 'failed'（三态：created/accounted/failed）。 与账号/存储的同步记账分工明确（2026-09 用户定夺）： - 账号/存储：purchase_realtime 同步记账 - - 模型（本模块）：调用时只记成本侧 + pending 流水，出账走本循环（异步） + - 模型（本模块）：调用时只记成本侧 + created 流水，出账走本循环（异步） 单循环守卫（多进程部署时防重复出账）：Redis SETNX llm_acc:lock（db4，与限流同库）。 """ @@ -31,7 +32,7 @@ _LOCK_TTL = 90 # 锁 TTL（略大于单批预算耗时） async def _get_pending(sor): recs = await sor.sqlExe( "SELECT id, org_id, user_id, model_id, req_tokens, resp_tokens, cost, ppid, usages " - "FROM llm_usage WHERE accounting_status='pending' AND status='ok' " + "FROM llm_usage WHERE accounting_status='created' AND status='ok' " "ORDER BY created_at LIMIT %d" % _BATCH, {}) await sor.sqlExe("COMMIT", {}) return recs or [] @@ -76,7 +77,7 @@ async def _settle_one(sor, row): pass fn = getattr(env, 'product_accounting_generic', None) if fn is None: - # 产品模块未加载 → 保留 pending 下轮再试 + # 产品模块未加载 → 保留 created 下轮再试 return False result = await fn( product_id=product_id, diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index 7d2ad33..c1a6d2c 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -492,7 +492,7 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_tokens: int = 0, note: str = '', usages: dict = None): - """结算 ⑧：成本侧扣减 + 用量流水（记账分流）。 + """结算 ⑧：成本侧扣减 + 用量流水（三态记账状态机）。 ctx govern_resolve 成功时返回的 dict（含 policy_org_id） ok_call 上游调用是否成功（失败也记账：写 failed 流水） @@ -500,12 +500,15 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke 图像张数等按量计费模型的计价因子，由异步出账读入定价引擎 （与 req_tokens/resp_tokens 并列，定价引擎按 pricing_data 选用）。 - 记账分流（2026-09）： - 本机构模型（含 owner 自用）：只记成本侧——供应商账号扣减， - 不动组织池/个人额度，流水 accounting_status='na'。 - owner 模型（调用方 ≠ '0'）：成本侧照常扣减，流水 - accounting_status='pending'，由异步产品计费链路出三方账 - （客户付/商户营收/供应商成本，折扣在出账时精确到产品）。 + accounting_status 三态（2026-09-05 用户定夺）： + created —— 已创建、待记账（status='ok' 的行由异步出账循环拾取） + accounted —— 记账成功 + failed —— 记账失败 + 自用 vs 跨机构的分流不在这里——product_accounting_generic 内部按 + is_self_use 决定记账项（自用只记 PAY* 采购成本，跨机构另记 PAY 客户应付）， + 所有成功调用统一 created 进出账队列，成本账不遗漏。 + status='failed'（调用失败）/'recharge'（充值）的行也置 created，但出账循环 + 只拾取 status='ok'，故它们不会被记账（无可计费用量）。 """ if not ctx or not isinstance(ctx, dict): return @@ -516,10 +519,7 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke # 出账由异步记账循环（accounting.py）经产品层计算；此处 cost 恒 0， # 供应商账号余额不再按调用扣减（真实成本以定价引擎结算为准）。 cost = 0.0 - policy_org = ctx.get('policy_org_id') or '' caller_org = ctx.get('org_id') or '' - # owner 模型 = 生效策略是平台('0') 且调用方不是平台自己 - is_owner_model = policy_org == '0' and caller_org != '0' async with db.sqlorContext(dbname) as sor: if not ok_call: # 上游限流(429)失败 → 账号级冷却，轮转避让下一轮不再选它 @@ -532,10 +532,6 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke "UPDATE llm_account SET balance=balance-${c}$ WHERE id=${i}$", {"c": cost, "i": ctx.get('account_id', '')}) await sor.sqlExe("COMMIT", {}) - # 用量流水：本机构='na'(仅成本) / owner模型='pending'(待三方账) / 失败='na' - accounting_status = 'na' - if ok_call and is_owner_model: - accounting_status = 'pending' from appPublic.uniqueID import getID usages_str = '' if usages: @@ -553,11 +549,11 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke 'req_tokens': int(req_tokens or 0), 'resp_tokens': int(resp_tokens or 0), 'cost': round(cost, 6), - 'charge': 0, # 客户应付由异步三方账计算（本机构模型不收费） + 'charge': 0, # 客户应付由异步出账计算（自用为 0，跨机构按折扣） 'ppid': model.get('ppid', '') or '', 'task_ref': ctx.get('task_ref', ''), 'status': 'ok' if ok_call else 'failed', - 'accounting_status': accounting_status, + 'accounting_status': 'created', 'usages': usages_str, 'note': (note or '')[:200], })