refactor(usage): accounting_status三态语义(created待记账/accounted成功/failed失败,用户定夺)——去na/pending分流,所有成功调用统一created进出账队列,自用vs跨机构由产品层is_self_use分流,成本账不遗漏
This commit is contained in:
parent
5ec2ca8fa4
commit
0a5974e935
@ -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",
|
||||
|
||||
@ -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)',
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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],
|
||||
})
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user