From e9e57cf82554e3bef50242739c9f5c3667bc6255 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Mon, 7 Sep 2026 16:57:43 +0800
Subject: [PATCH] =?UTF-8?q?fix(=E8=AE=B0=E8=B4=A6):=20=E5=90=8C=E6=AD=A5?=
 =?UTF-8?q?=E8=B0=83=E7=94=A8usage=E5=8E=9F=E6=96=87=E8=90=BDusages+?=
 =?UTF-8?q?=E5=87=BA=E8=B4=A6=E8=A1=A5prompt=5Ftokens=5Fdetails/hour?=
 =?UTF-8?q?=E7=BB=B4=E5=BA=A6=E2=80=94=E2=80=94=E6=A0=B9=E6=B2=BBderived?=
 =?UTF-8?q?=E7=BC=93=E5=AD=98=E5=8F=98=E9=87=8F=E6=B1=82=E5=80=BC=E5=A4=B1?=
 =?UTF-8?q?=E8=B4=A5(uncache=5Ftokens=3D0=E8=BE=93=E5=85=A5=E4=B8=8D?=
 =?UTF-8?q?=E8=AE=A1=E8=B4=B9)=E4=B8=8E=E5=BF=99=E9=97=B2=E6=97=B6?=
 =?UTF-8?q?=E5=88=86=E6=A1=A3=E8=AE=A1=E4=BB=B7=E7=BC=BA=E5=A4=B1(2026-09-?=
 =?UTF-8?q?07=E5=AE=9E=E6=B5=8Bqwen3.8-max=E8=AE=B0=E8=B4=A6failed?=
 =?UTF-8?q?=E6=A0=B9=E5=9B=A0)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pipeline_llm/accounting.py | 25 +++++++++++++++++++++++--
 pipeline_llm/inference.py  | 11 ++++++++++-
 2 files changed, 33 insertions(+), 3 deletions(-)

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