diff --git a/pipeline_llm/accounting.py b/pipeline_llm/accounting.py index 64ccf28..930499f 100644 --- a/pipeline_llm/accounting.py +++ b/pipeline_llm/accounting.py @@ -33,7 +33,8 @@ _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 " + "SELECT id, org_id, user_id, model_id, req_tokens, resp_tokens, cost, ppid, usages, " + "created_at " "FROM llm_usage WHERE accounting_status='created' AND status='SUCCEEDED' " "ORDER BY created_at LIMIT %d" % _BATCH, {}) await sor.sqlExe("COMMIT", {}) @@ -63,7 +64,7 @@ async def _settle_one(sor, row): logger.warning("pipeline_llm.accounting: model %s 无产品映射,流水 %s 置 failed", model_id, row.id) return False - usage_data = { + usage_data: dict = { 'prompt_tokens': int(getattr(row, 'req_tokens', 0) or 0), 'completion_tokens': int(getattr(row, 'resp_tokens', 0) or 0), } @@ -77,6 +78,26 @@ async def _settle_one(sor, row): usage_data.update(extra) except Exception: pass + # derived 依赖归一(2026-09-07 根治):token 定价的 cached_tokens/uncache_tokens + # 靠 YAML derived 变量引用 prompt_tokens_details.cached_tokens 计算——该键缺失时 + # DictObject 属性链抛 AttributeError,引擎兜底成 0,输入 token 永不计费 + # (实测 qwen3.8-max uncache_tokens=0 记账失败根因之一)。缺失补 0 结构, + # derived 正常求值:无缓存时 uncache=prompt_tokens、cached=0。 + ptd = usage_data.get('prompt_tokens_details') + if not isinstance(ptd, dict): + usage_data['prompt_tokens_details'] = {'cached_tokens': 0} + elif ptd.get('cached_tokens') is None: + ptd['cached_tokens'] = 0 + # 忙闲时计价维度(2026-09-07):价目表存在忙时(8点-22点)/闲时(22点-次日8点) + # 两档定价(deepseek-v4-pro-0813),按调用时刻 llm_usage.created_at 的小时数注入 + # hour——YAML 用 hour 区间 filters 选档。多余键对不用它的定价方案无影响 + # (引擎只严格检查定价项里出现的维度键)。 + created_at = getattr(row, 'created_at', None) + if created_at is not None: + try: + usage_data['hour'] = int(created_at.hour) + except Exception: + pass fn = getattr(env, 'product_accounting_generic', None) if fn is None: # 产品模块未加载 → 保留 created 下轮再试 diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index aaa86ec..b9e3b56 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -742,7 +742,16 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): usage = data.get('usage') or {} rt = int(_fnum(usage.get('prompt_tokens'))) ct = int(_fnum(usage.get('completion_tokens'))) - await govern_settle(ctx, True, rt, ct, '' if usage else 'est') + # usage 原文落 llm_usage.usages(2026-09-07 根治,对齐异步路径): + # 定价引擎的 derived 变量引用(cached_tokens=prompt_tokens_details.cached_tokens、 + # uncache_tokens=prompt_tokens - prompt_tokens_details.cached_tokens)依赖原始 usage + # 结构——此前同步路径只存 req/resp 两列、usages 为空,导致 prompt_tokens_details + # 缺失 → derived 求值失败被兜底成 0 → 输入 token 永不计费(实测 qwen3.8-max + # uncache_tokens=0 记账失败根因)。sage/llmage 同步范式即 usages=json.dumps(usage)。 + usages = dict(usage) if isinstance(usage, dict) else {} + if usages: + usages['model'] = ctx.get('model_id') or '' + await govern_settle(ctx, True, rt, ct, '' if usage else 'est', usages) return data