From 427cd7cbc13d75ac335d94964f21cca138f83b01 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 11 Sep 2026 12:51:31 +0800 Subject: [PATCH] =?UTF-8?q?feat(usage):=20usages=E5=85=A8=E9=87=8F?= =?UTF-8?q?=E4=B8=8D=E6=88=AA=E6=96=AD=E4=B8=BA=E8=AE=A1=E8=B4=B9=E5=94=AF?= =?UTF-8?q?=E4=B8=80=E4=BA=8B=E5=AE=9E=E6=BA=90(m0021=E5=88=A0token?= =?UTF-8?q?=E5=86=97=E4=BD=99=E5=88=97)+llm=5Fcall=5Ftrace=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E8=A1=8C=E5=8E=9F=E6=96=87=E8=BF=BD=E8=B8=AA+project?= =?UTF-8?q?=5Fid=E6=8C=89=E9=A1=B9=E7=9B=AE=E7=BB=9F=E8=AE=A1=E8=B4=B9?= =?UTF-8?q?=E7=94=A8(2026-09-10)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 11 +- docs/design-spec.md | 6 +- json/llm_call_trace_list.json | 76 ++++++ json/llm_usage_list.json | 14 - models/llm_call_trace.json | 111 ++++++++ models/llm_usage.json | 337 +++++++++++++----------- mysql.ddl.sql | 9 +- pipeline_llm/accounting.py | 38 ++- pipeline_llm/gateway.py | 35 ++- pipeline_llm/inference.py | 93 +++++-- pipeline_llm/init.py | 293 +++++++++++++++++++- pipeline_llm/product_interface.py | 9 +- pipeline_llm/trace.py | 192 ++++++++++++++ wwwroot/api/llm_call_trace_io.dspy | 10 + wwwroot/api/llm_project_cost_query.dspy | 11 + wwwroot/api/v1/chat/completions.dspy | 3 +- 16 files changed, 1005 insertions(+), 243 deletions(-) create mode 100644 json/llm_call_trace_list.json create mode 100644 models/llm_call_trace.json create mode 100644 pipeline_llm/trace.py create mode 100644 wwwroot/api/llm_call_trace_io.dspy create mode 100644 wwwroot/api/llm_project_cost_query.dspy diff --git a/README.md b/README.md index 7f471a4..9c909ce 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ | `llm_org_policy` | 组织容错策略（主备模型 + 辅助模型 utility_model_id + 端点偏好） | | `llm_org_quota` | 组织限额限流（池余额 + QPM） | | `llm_user_quota` | 个人限额限流（额度上限 + QPM） | -| `llm_usage` | 用量流水（双维度记账事实表；status 统一枚举 SUCCEEDED/FAILED/PENDING/RUNNING/recharge；accounting_status 仅 SUCCEEDED 行有意义否则 NULL；usages 字段存非 token 计价因子 JSON） | +| `llm_usage` | 用量流水（双维度记账事实表；status 统一枚举 SUCCEEDED/FAILED/PENDING/RUNNING/recharge；accounting_status 仅 SUCCEEDED 行有意义否则 NULL；**usages 是计费唯一事实源**——上游 usage 原文 JSON 全量不截断，token 计数与按量因子都在其中；req_tokens/resp_tokens 冗余列已删（m0021），统计查询走 JSON_EXTRACT；project_id 关联 sd_projects 按项目统计费用（非项目调用哨兵'0'，查询不传=不区分项目，m0025）；call_id 关联 llm_call_trace 上行/下行原文） | ## 调用门禁链 @@ -66,9 +66,12 @@ **只有 status='SUCCEEDED' 才有记账意义**：仅成功行写 created，非 SUCCEEDED 行保持 **NULL** （不冒充待记账）。出账循环扫 `accounting_status='created' AND status='SUCCEEDED'`。 自用 vs 跨机构分流由产品层 `product_accounting_generic` 的 is_self_use 决定（自用只记 PAY* 采购成本， - 跨机构另记 PAY 客户应付）。流水 `usages` 存非 token 计价因子（视频时长/ - 分辨率等），异步出账循环（独立进程，需 load_appbase 提供 get_business_date） - 读入定价引擎。未挂 ppid 的模型出账置 failed 并写明原因（不静默）。 + 跨机构另记 PAY 客户应付）。流水 `usages` 存上游 usage 原文 JSON（**全量不截断**， + 2026-09-10 用户定夺：usages 是计费唯一事实源，req_tokens/resp_tokens 冗余列删除 + m0021；此前 [:2000] 截断会产生非法 JSON → 出账 json.loads 静默失败 → derived + 兜底 0 → 永不计费），异步出账循环（独立进程，需 load_appbase 提供 + get_business_date）读入定价引擎；usages 非法 JSON 或为空 → 出账置 failed 写明 + 原因（禁静默 0 元账）。未挂 ppid 的模型出账置 failed 并写明原因（不静默）。 - **method 独立列（2026-09-06 用户定夺，m0012）**：`llm_api_profile.method`（GET/POST， 默认 POST）不塞 request_template JSON；运行时读该列决定 HTTP 动词，存量行由迁移 从 request_template 的 $.method 回填，查询步骤兼容回退解析。 diff --git a/docs/design-spec.md b/docs/design-spec.md index 71702d3..dc706ca 100644 --- a/docs/design-spec.md +++ b/docs/design-spec.md @@ -155,12 +155,14 @@ must 模式：候选端点耗尽宁可报错冒泡，绝不跨端点。 | org_id / user_id | str(32) | 消费侧维度 | | model_id / account_id | str(32) | 供给侧维度（model_id 空+source=recharge 时为充值记录） | | endpoint_region | str(20) | 实际走的端点区域 | -| req_tokens / resp_tokens | int | 实际用量 | +| usages | text | 用量原文JSON（计费唯一事实源：token计数+按量因子，全量不截断） | | cost | decimal(12,6) | 成本侧金额（账号扣减，按定价） | | charge | decimal(12,6) | 消费侧金额（组织池扣减，按产品定价） | | ppid | str(32) | 定价项目（冗余自模型，便于对账） | +| project_id | str(32) NOT NULL DEFAULT '0' | 项目ID（sd_projects.id；'0'=非项目调用哨兵值——按项目统计费用，查询不传=不区分项目，2026-09-10；token 绑定 project_id 全链透传） | | task_ref | str(100) | 调用来源（任务/会话标识） | -| status | str(20) | ok / failed / recharge（充值记录） | +| call_id | str(32) | 调用批次ID（关联 llm_call_trace 原文追踪，一次调用=N个往返；2026-09-10） | +| status | str(20) | 统一枚举 SUCCEEDED/FAILED/PENDING/RUNNING（其他如 recharge 充值记录） | | note | str(200) | 失败原因/充值备注 | | created_at | str(30) | | diff --git a/json/llm_call_trace_list.json b/json/llm_call_trace_list.json new file mode 100644 index 0000000..97cbb44 --- /dev/null +++ b/json/llm_call_trace_list.json @@ -0,0 +1,76 @@ +{ + "tblname": "llm_call_trace", + "title": "模型调用原文追踪", + "params": { + "sortby": "created_at desc", + "noedit": true, + "browserfields": { + "exclouded": [ + "id", + "req_path", + "resp_path", + "org_id" + ], + "alters": { + "url": { + "cwidth": 40 + }, + "note": { + "cwidth": 25 + }, + "kind": { + "uitype": "code", + "data": [ + {"value": "chat", "text": "对话"}, + {"value": "gen", "text": "同步生成"}, + {"value": "submit", "text": "异步提交"}, + {"value": "query", "text": "任务轮询"} + ], + "valueField": "value", + "textField": "text" + }, + "method": { + "uitype": "code", + "data": [ + {"value": "POST", "text": "POST"}, + {"value": "GET", "text": "GET"} + ], + "valueField": "value", + "textField": "text" + } + } + }, + "editexclouded": [ + "id", + "org_id", + "created_at" + ], + "toolbar": { + "tools": [ + { + "name": "view_io", + "label": "查看IO", + "selected_row": true, + "icon": "{{entire_url('/bricks/imgs/chat-user.svg')}}" + } + ] + }, + "binds": [ + { + "wid": "self", + "event": "view_io", + "actiontype": "urlwidget", + "target": "PopupWindow", + "popup_options": { + "title": "调用原文详情", + "width": "85%", + "height": "85%", + "resizable": true + }, + "options": { + "url": "{{entire_url('../api/llm_call_trace_io.dspy')}}?id=${id}$" + } + } + ] + } +} diff --git a/json/llm_usage_list.json b/json/llm_usage_list.json index 91d72ec..81a2954 100644 --- a/json/llm_usage_list.json +++ b/json/llm_usage_list.json @@ -17,8 +17,6 @@ "model_id", "account_id", "endpoint_region", - "req_tokens", - "resp_tokens", "cost", "charge", "ppid", @@ -26,18 +24,6 @@ "status", "note", "created_at" - ], - "record_toolbar": [ - { - "label": "查询统计", - "actiontype": "dspy", - "url": "{{entire_url('../api/llm_usage_query.dspy')}}", - "options": { - "icon": "search", - "cwidth": 20, - "cheight": 12 - } - } ] } } diff --git a/models/llm_call_trace.json b/models/llm_call_trace.json new file mode 100644 index 0000000..bb1d602 --- /dev/null +++ b/models/llm_call_trace.json @@ -0,0 +1,111 @@ +{ + "summary": [ + { + "name": "llm_call_trace", + "title": "模型调用原文追踪(上行下行)", + "primary": [ + "id" + ], + "catelog": "entity" + } + ], + "fields": [ + { + "name": "id", + "title": "主键ID", + "type": "str", + "length": 32, + "nullable": "no" + }, + { + "name": "call_id", + "title": "调用批次ID(与llm_usage.call_id同值,一次chat_inference=N个往返)", + "type": "str", + "length": 32 + }, + { + "name": "seq", + "title": "往返序号(同步重试0..2;异步提交0,轮询1..N)", + "type": "int", + "default": "0" + }, + { + "name": "kind", + "title": "往返类型(chat对话/submit异步提交/query任务查询)", + "type": "str", + "length": 20 + }, + { + "name": "url", + "title": "上游URL(去query防签名泄露)", + "type": "str", + "length": 500 + }, + { + "name": "method", + "title": "HTTP方法", + "type": "str", + "length": 10 + }, + { + "name": "status_code", + "title": "响应状态码", + "type": "int", + "default": "0" + }, + { + "name": "elapsed_ms", + "title": "耗时(毫秒)", + "type": "int", + "default": "0" + }, + { + "name": "req_path", + "title": "请求原文文件路径(filesroot相对;headers已脱敏)", + "type": "str", + "length": 300 + }, + { + "name": "resp_path", + "title": "响应原文文件路径(filesroot相对)", + "type": "str", + "length": 300 + }, + { + "name": "org_id", + "title": "机构ID(下载端点机构隔离用)", + "type": "str", + "length": 32 + }, + { + "name": "note", + "title": "错误摘要", + "type": "str", + "length": 200 + }, + { + "name": "created_at", + "title": "创建时间", + "type": "timestamp", + "nullable": "no" + } + ], + "indexes": [ + { + "name": "idx_llm_call_trace_call", + "idxtype": "index", + "idxfields": [ + "call_id" + ] + }, + { + "name": "idx_llm_call_trace_org", + "idxtype": "index", + "idxfields": [ + "org_id", + "created_at" + ] + } + ], + "codes": [] +} diff --git a/models/llm_usage.json b/models/llm_usage.json index 5e3863a..effb56c 100644 --- a/models/llm_usage.json +++ b/models/llm_usage.json @@ -1,162 +1,179 @@ { - "summary": [ - { - "name": "llm_usage", - "title": "用量流水(双维度记账)", - "primary": [ - "id" - ], - "catelog": "entity" - } - ], - "fields": [ - { - "name": "id", - "title": "主键ID", - "type": "str", - "length": 32, - "nullable": "no" - }, - { - "name": "org_id", - "title": "机构ID", - "type": "str", - "length": 32 - }, - { - "name": "user_id", - "title": "用户ID", - "type": "str", - "length": 32 - }, - { - "name": "model_id", - "title": "模型ID", - "type": "str", - "length": 32 - }, - { - "name": "account_id", - "title": "账号ID", - "type": "str", - "length": 32 - }, - { - "name": "endpoint_region", - "title": "端点区域", - "type": "str", - "length": 20 - }, - { - "name": "req_tokens", - "title": "请求token", - "type": "int", - "default": "0" - }, - { - "name": "resp_tokens", - "title": "响应token", - "type": "int", - "default": "0" - }, - { - "name": "cost", - "title": "成本金额(账号)", - "type": "float", - "length": 12, - "dec": 6, - "default": "0" - }, - { - "name": "charge", - "title": "客户应付(异步三方账计算)", - "type": "float", - "length": 12, - "dec": 6, - "default": "0" - }, - { - "name": "accounting_status", - "title": "记账状态(created待记账/accounted记账成功/failed记账失败;仅status=SUCCEEDED时有意义,否则NULL)", - "type": "str", - "length": 20, - "nullable": "yes" - }, - { - "name": "ppid", - "title": "定价项目ID", - "type": "str", - "length": 32 - }, - { - "name": "task_ref", - "title": "调用来源", - "type": "str", - "length": 100 - }, - { - "name": "status", - "title": "调用状态(SUCCEEDED/FAILED/PENDING/RUNNING/其他如recharge)", - "type": "str", - "length": 20, - "nullable": "no", - "default": "'SUCCEEDED'" - }, - { - "name": "usages", - "title": "用量详情(JSON)", - "type": "text", - "nullable": "yes" - }, - { - "name": "note", - "title": "备注/原因", - "type": "str", - "length": 200 - }, - { - "name": "created_at", - "title": "创建时间", - "type": "timestamp", - "nullable": "no" - } - ], - "indexes": [ - { - "name": "idx_llm_usage_org", - "idxtype": "index", - "idxfields": [ - "org_id", - "created_at" - ] - }, - { - "name": "idx_llm_usage_model", - "idxtype": "index", - "idxfields": [ - "model_id" - ] - } - ], - "codes": [ - { - "field": "model_id", - "table": "llm_model", - "valuefield": "id", - "textfield": "name" - }, - { - "field": "account_id", - "table": "llm_account", - "valuefield": "id", - "textfield": "name" - }, - { - "field": "status", - "table": "appcodes_kv", - "valuefield": "k", - "textfield": "v", - "cond": "parentid='llm_usage_status'" - } - ] + "summary": [ + { + "name": "llm_usage", + "title": "用量流水(双维度记账)", + "primary": [ + "id" + ], + "catelog": "entity" + } + ], + "fields": [ + { + "name": "id", + "title": "主键ID", + "type": "str", + "length": 32, + "nullable": "no" + }, + { + "name": "org_id", + "title": "机构ID", + "type": "str", + "length": 32 + }, + { + "name": "user_id", + "title": "用户ID", + "type": "str", + "length": 32 + }, + { + "name": "model_id", + "title": "模型ID", + "type": "str", + "length": 32 + }, + { + "name": "account_id", + "title": "账号ID", + "type": "str", + "length": 32 + }, + { + "name": "endpoint_region", + "title": "端点区域", + "type": "str", + "length": 20 + }, + { + "name": "cost", + "title": "成本金额(账号)", + "type": "float", + "length": 12, + "dec": 6, + "default": "0" + }, + { + "name": "charge", + "title": "客户应付(异步三方账计算)", + "type": "float", + "length": 12, + "dec": 6, + "default": "0" + }, + { + "name": "accounting_status", + "title": "记账状态(created待记账/accounted记账成功/failed记账失败;仅status=SUCCEEDED时有意义,否则NULL)", + "type": "str", + "length": 20, + "nullable": "yes" + }, + { + "name": "ppid", + "title": "定价项目ID", + "type": "str", + "length": 32 + }, + { + "name": "project_id", + "title": "项目ID(sd_projects.id;'0'=非项目调用,按项目统计费用,查询不传=不区分项目)", + "type": "str", + "length": 32, + "nullable": "no", + "default": "'0'" + }, + { + "name": "task_ref", + "title": "调用来源", + "type": "str", + "length": 100 + }, + { + "name": "call_id", + "title": "调用批次ID(关联llm_call_trace原文追踪,一次调用=N个往返)", + "type": "str", + "length": 32 + }, + { + "name": "status", + "title": "调用状态(SUCCEEDED/FAILED/PENDING/RUNNING/其他如recharge)", + "type": "str", + "length": 20, + "nullable": "no", + "default": "'SUCCEEDED'" + }, + { + "name": "usages", + "title": "用量原文JSON(计费唯一事实源:token计数+按量因子,全量不截断)", + "type": "text", + "nullable": "yes" + }, + { + "name": "note", + "title": "备注/原因", + "type": "str", + "length": 200 + }, + { + "name": "created_at", + "title": "创建时间", + "type": "timestamp", + "nullable": "no" + } + ], + "indexes": [ + { + "name": "idx_llm_usage_org", + "idxtype": "index", + "idxfields": [ + "org_id", + "created_at" + ] + }, + { + "name": "idx_llm_usage_model", + "idxtype": "index", + "idxfields": [ + "model_id" + ] + }, + { + "name": "idx_llm_usage_call", + "idxtype": "index", + "idxfields": [ + "call_id" + ] + }, + { + "name": "idx_llm_usage_project", + "idxtype": "index", + "idxfields": [ + "project_id", + "created_at" + ] + } + ], + "codes": [ + { + "field": "model_id", + "table": "llm_model", + "valuefield": "id", + "textfield": "name" + }, + { + "field": "account_id", + "table": "llm_account", + "valuefield": "id", + "textfield": "name" + }, + { + "field": "status", + "table": "appcodes_kv", + "valuefield": "k", + "textfield": "v", + "cond": "parentid='llm_usage_status'" + } + ] } \ No newline at end of file diff --git a/mysql.ddl.sql b/mysql.ddl.sql index 3bcca78..2dc0868 100644 --- a/mysql.ddl.sql +++ b/mysql.ddl.sql @@ -118,18 +118,19 @@ CREATE TABLE IF NOT EXISTS llm_usage ( `model_id` varchar(32) comment '模型ID', `account_id` varchar(32) comment '账号ID', `endpoint_region` varchar(20) comment '端点区域', - `req_tokens` int NOT NULL DEFAULT 0 comment '请求token', - `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) comment '记账状态(三态:created待记账/accounted记账成功/failed记账失败;仅status=SUCCEEDED的行有意义,其余NULL)', `ppid` varchar(32) comment '定价项目ID', + `project_id` varchar(32) NOT NULL DEFAULT '0' comment '项目ID(sd_projects.id;''0''=非项目调用,查询不传=不区分项目)', `task_ref` varchar(100) comment '调用来源', + `call_id` varchar(32) comment '调用批次ID(关联llm_call_trace原文追踪,一次调用=N个往返)', `status` varchar(20) NOT NULL DEFAULT 'SUCCEEDED' comment '调用状态(统一枚举:SUCCEEDED成功/FAILED失败/PENDING等待/RUNNING运行;其他如recharge)', - `usages` text comment '非token用量因子JSON(时长/分辨率等,异步出账读入定价引擎)', + `usages` text comment '用量原文JSON(计费唯一事实源:token计数+按量因子,全量不截断,异步出账读入定价引擎)', `note` varchar(200) comment '备注/原因', `created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间', PRIMARY KEY (`id`), KEY `idx_llm_usage_org` (`org_id`,`created_at`), - KEY `idx_llm_usage_model` (`model_id`) + KEY `idx_llm_usage_model` (`model_id`), + KEY `idx_llm_usage_project` (`project_id`,`created_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci comment='用量流水(双维度记账)'; diff --git a/pipeline_llm/accounting.py b/pipeline_llm/accounting.py index 930499f..d622df3 100644 --- a/pipeline_llm/accounting.py +++ b/pipeline_llm/accounting.py @@ -33,7 +33,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, " + "SELECT id, org_id, user_id, model_id, cost, ppid, usages, " "created_at " "FROM llm_usage WHERE accounting_status='created' AND status='SUCCEEDED' " "ORDER BY created_at LIMIT %d" % _BATCH, {}) @@ -64,12 +64,10 @@ async def _settle_one(sor, row): logger.warning("pipeline_llm.accounting: model %s 无产品映射，流水 %s 置 failed", model_id, row.id) return False - usage_data: dict = { - 'prompt_tokens': int(getattr(row, 'req_tokens', 0) or 0), - 'completion_tokens': int(getattr(row, 'resp_tokens', 0) or 0), - } - # 非 token 计价因子（视频时长/分辨率/模型名等，2026-09-05）： - # llm_usage.usages JSON 合并进定价引擎输入，按量计费模型（视频/图像）用 + # 计费唯一事实源 = usages JSON 全量原文（2026-09-10 用户定夺，req/resp_tokens + # 冗余列已删 m0021）：token 计数（prompt_tokens/completion_tokens/ + # prompt_tokens_details）与按量因子（时长/分辨率/张数）都在其中。 + usage_data: dict = {} usages_raw = getattr(row, 'usages', '') or '' if usages_raw: try: @@ -77,7 +75,24 @@ async def _settle_one(sor, row): if isinstance(extra, dict): usage_data.update(extra) except Exception: - pass + # usages 是计费事实源，解析失败必须显式记账失败（写明原因）， + # 禁止静默空 dict 兜底成 0 元账（历史上 [:2000] 截断即此病根） + await sor.U('llm_usage', { + 'id': row.id, 'accounting_status': 'failed', + 'note': '出账失败：usages 不是合法 JSON（长度%d），无法计费——' + '请核查流水写入方是否截断' % len(usages_raw), + }) + await sor.sqlExe("COMMIT", {}) + logger.warning("pipeline_llm.accounting: 流水 %s usages 非法 JSON，置 failed", row.id) + return False + if not usage_data: + await sor.U('llm_usage', { + 'id': row.id, 'accounting_status': 'failed', + 'note': '出账失败：usages 为空，无计费事实源——请核查上游 usage 返回', + }) + await sor.sqlExe("COMMIT", {}) + logger.warning("pipeline_llm.accounting: 流水 %s usages 为空，置 failed", row.id) + return False # derived 依赖归一（2026-09-07 根治）：token 定价的 cached_tokens/uncache_tokens # 靠 YAML derived 变量引用 prompt_tokens_details.cached_tokens 计算——该键缺失时 # DictObject 属性链抛 AttributeError，引擎兜底成 0，输入 token 永不计费 @@ -92,10 +107,15 @@ async def _settle_one(sor, row): # 两档定价（deepseek-v4-pro-0813），按调用时刻 llm_usage.created_at 的小时数注入 # hour——YAML 用 hour 区间 filters 选档。多余键对不用它的定价方案无影响 # （引擎只严格检查定价项里出现的维度键）。 + # 2026-09-10：注入浮点小时（时+分/60+秒/3600），支持 pricing 引擎 peak_times + # 时段串（如 "9:00 =~= 12:00"）的精确时刻语义；对存量整数小时区间 YAML + # （8 =~ 22 等）边界语义不变（整点值相同）。 created_at = getattr(row, 'created_at', None) if created_at is not None: try: - usage_data['hour'] = int(created_at.hour) + usage_data['hour'] = (int(created_at.hour) + + int(created_at.minute) / 60.0 + + int(created_at.second) / 3600.0) except Exception: pass fn = getattr(env, 'product_accounting_generic', None) diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index 262af00..89a6ba6 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -10,8 +10,9 @@ 集成方式（复用不改建）： - llm_bridge._get_model_config 前置调 govern_resolve：机构配了策略/新模型表才走治理， 否则返回 None 由旧 llm 表逻辑兜底（向后兼容）。 - - proxy_chat_completion（运行环境 token 路径）拿到真实 usage 后调 govern_settle 精算。 - - llm_bridge 内部路径按 prompt/response 长度估算 token 调 govern_settle（note='est'）。 + - proxy_chat_completion（运行环境 token 路径）委托 chat_inference 统一结算。 + - 结算唯一入口 govern_settle：usages 落上游 usage 原文 JSON（全量不截断， + 计费唯一事实源；token 冗余列已删 m0021，2026-09-10 用户定夺）。 限流：Redis 分钟窗口原子计数（db4，与会话 db3 隔离）。Redis 不可用时限流放行（fail-open， 记 warning）——限流失效不应阻断业务，余额检查仍在 DB 侧守住。 @@ -460,7 +461,8 @@ 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 = ''): + est_tokens: int = 0, task_ref: str = '', purpose: str = '', + project_id: str = ''): """门禁链 ①-⑥。 策略来源（2026-09 商业化默认）： @@ -587,19 +589,23 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '', 'user_id': user_id or '', 'org_id': org_id or '', 'task_ref': task_ref or '', + 'project_id': project_id or '', 'policy_org_id': policy_org, } -async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_tokens: int = 0, - note: str = '', usages: dict = None): +async def govern_settle(ctx: dict, ok_call: bool, note: str = '', usages: dict = None): """结算 ⑧：成本侧扣减 + 用量流水（三态记账状态机）。 ctx govern_resolve 成功时返回的 dict（含 policy_org_id） ok_call 上游调用是否成功（失败也记账：写 failed 流水） - usages 非 token 用量因子（JSON 落 llm_usage.usages）：视频时长/分辨率、 - 图像张数等按量计费模型的计价因子，由异步出账读入定价引擎 - （与 req_tokens/resp_tokens 并列，定价引擎按 pricing_data 选用）。 + usages 上游返回的完整用量原文 JSON（落 llm_usage.usages，**全量不截断**， + 2026-09-10 用户定夺）：token 计数（prompt_tokens/completion_tokens/ + prompt_tokens_details）与按量因子（视频时长/分辨率/图像张数）都在其中， + 由异步出账读入定价引擎。**usages 是计费唯一事实源**——req_tokens/ + resp_tokens 冗余列已删除（m0021），统计查询走 JSON_EXTRACT。 + 历史教训：[:2000] 截断产生非法 JSON → accounting json.loads 静默失败 + → derived 求值兜底 0 → 永不计费，故禁止任何形式的截断。 accounting_status 三态（2026-09-05 用户定夺；2026-09-06 补 NULL 语义）： created —— 已创建、待记账（仅 status='SUCCEEDED' 的行，出账循环拾取） @@ -638,7 +644,9 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke usages_str = '' if usages: try: - usages_str = json.dumps(usages, ensure_ascii=False, default=str)[:2000] + # 全量落库不截断（2026-09-10 用户定夺）：截断产生非法 JSON → + # accounting json.loads 静默失败 → derived 兜底 0 → 永不计费 + usages_str = json.dumps(usages, ensure_ascii=False, default=str) except Exception: usages_str = '' # status 统一枚举（2026-09-06 用户定夺）：SUCCEEDED/FAILED； @@ -651,12 +659,15 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke 'model_id': model.get('id', ''), 'account_id': ctx.get('account_id', ''), 'endpoint_region': ctx.get('endpoint_region', ''), - 'req_tokens': int(req_tokens or 0), - 'resp_tokens': int(resp_tokens or 0), 'cost': round(cost, 6), 'charge': 0, # 客户应付由异步出账计算（自用为 0，跨机构按折扣） 'ppid': model.get('ppid', '') or '', + # 项目归属（2026-09-10 用户定夺）：token 绑定的 project_id 透传， + # 非项目调用（平台测试/充值/无项目 token）写哨兵'0'（列 NOT NULL + # DEFAULT '0'，查询不传=不区分项目），按项目统计费用 + 'project_id': ctx.get('project_id') or '0', 'task_ref': ctx.get('task_ref', ''), + 'call_id': ctx.get('call_id', '') or '', 'status': 'SUCCEEDED' if ok_call else 'FAILED', 'usages': usages_str, 'note': (note or '')[:200], @@ -734,8 +745,6 @@ async def _write_recharge_usage(sor, org_id, user_id, account_id, amount, note, 'model_id': '', 'account_id': account_id or '', 'endpoint_region': '', - 'req_tokens': 0, - 'resp_tokens': 0, 'cost': round(float(amount), 6) if account_id else 0, 'charge': round(float(amount), 6) if org_id else 0, 'ppid': '', diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index b988d4e..3f2a358 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -142,7 +142,7 @@ def _texts_len(messages): # ────────────────────────── 解析链 ────────────────────────── async def _resolve_call(sor, org_id, user_id, model_name, capability, - est_tokens, task_ref, purpose=''): + est_tokens, task_ref, purpose='', project_id=''): """门禁链 ①-⑥ + 加载模型/适配模板。 purpose='utility'：辅助任务（分类/选择/摘要），模型链为 辅助→主→备 @@ -157,7 +157,8 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability, raise GovernError('缺少机构标识（机构隔离必需）：内部调用须带 org_id') 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 '') + est_tokens=int(est_tokens or 0), task_ref=task_ref or '', purpose=purpose or '', + project_id=project_id or '') if not ok: if res == '__LEGACY__': # 全平台无策略/指定模型不在新表：不再有旧表兜底，报可行动错误 @@ -178,7 +179,7 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability, if capability and cap != capability: # 释放预授权再报错（调用不会发生） from .gateway import govern_settle - await govern_settle(ctx, False, 0, 0, 'capability mismatch: %s!=%s' % (cap, capability)) + await govern_settle(ctx, False, 'capability mismatch: %s!=%s' % (cap, capability)) raise GovernError( '模型「%s」能力为 %s，与请求能力 %s 不匹配。请按能力分类选择模型' % ( model_row.get('name', ''), cap, capability)) @@ -387,6 +388,18 @@ async def _build_upstream_body(ctx, payload): return body +async def _trace_rt(ctx, seq, kind, url, method, headers, req_body, + status_code, resp_data, t0, err=''): + """原文追踪薄封装：永不抛（trace.record_roundtrip 内部已兜底，此处双保险）。""" + try: + from .trace import record_roundtrip + await record_roundtrip( + ctx, seq, kind, url, method, headers, req_body, + status_code, resp_data, int((time.time() - t0) * 1000), err=err) + except Exception as e: + print("[inference] trace 记录失败（不阻断）: %s" % str(e)[:150]) + + async def _post_upstream(ctx, payload, require_choices=True): """通用 POST 上游（path/headers/body 全模板组包），带瞬时错误重试。 @@ -406,6 +419,7 @@ async def _post_upstream(ctx, payload, require_choices=True): url = _join_url(ctx.get('api_base'), path) headers = await _render_headers(profile.get('headers', ''), ns) body = await _build_upstream_body(ctx, payload) + kind = 'chat' if require_choices else 'gen' # method 独立列（2026-09-06 用户定夺）：默认 POST，profile 声明优先 method = (profile.get('method') or 'POST').upper() timeout = int(_fnum(ctx.get('timeout')) or _TOTAL_TIMEOUT) @@ -417,6 +431,7 @@ async def _post_upstream(ctx, payload, require_choices=True): last = None for attempt in range(_MAX_ATTEMPTS): + _t0 = time.time() try: async with aiohttp.ClientSession() as session: async with session.request( @@ -425,6 +440,10 @@ async def _post_upstream(ctx, payload, require_choices=True): ) as resp: if resp.status != 200: text = await resp.text() + # 原文追踪：非 200 响应也落盘（排障关键证据，2026-09-10） + await _trace_rt(ctx, attempt, kind, url, method, headers, + body, resp.status, text, _t0, + err='HTTP %d' % resp.status) err = ValueError('上游调用失败 HTTP %d: %s' % (resp.status, text[:300])) if resp.status in _RETRYABLE_STATUS and attempt < _MAX_ATTEMPTS - 1: last = err @@ -436,16 +455,23 @@ async def _post_upstream(ctx, payload, require_choices=True): err = ValueError('上游响应%s: %s' % ( '缺 choices' if require_choices else '非 JSON 对象', json.dumps(data, ensure_ascii=False, default=str)[:300])) + await _trace_rt(ctx, attempt, kind, url, method, headers, + body, resp.status, data, _t0, err=str(err)[:200]) if attempt < _MAX_ATTEMPTS - 1: last = err await asyncio.sleep(2 * (attempt + 1)) continue raise err + await _trace_rt(ctx, attempt, kind, url, method, headers, + body, resp.status, data, _t0) return data except (asyncio.TimeoutError, aiohttp.ClientError) as e: last = e # TimeoutError 的 str() 是空串，用 repr 保证消息真实可行动 detail = repr(e) or type(e).__name__ + # 网络异常无响应体：仍落盘请求侧 + 错误摘要（往返完整性） + await _trace_rt(ctx, attempt, kind, url, method, headers, + body, 0, None, _t0, err=detail[:200]) # print 确保进应用日志（模块 logger 未接应用日志管道，2026-09-04 实测） print("[inference] 上游瞬时错误重试 %d/%d: %s (url=%s)" % ( attempt + 1, _MAX_ATTEMPTS, detail, url.split('?')[0])) @@ -507,7 +533,7 @@ async def _sync_generation_inference(ctx, payload): data = await _post_upstream(ctx, payload, require_choices=False) except Exception as e: note = str(e)[:200] - await govern_settle(ctx, False, 0, 0, + await govern_settle(ctx, False, '429 upstream: ' + note if ' 429' in note else note) raise GovernError(str(e)) # 上游业务错误（DashScope：顶层 code/message，无 output）——如实报错不粉饰 @@ -515,7 +541,7 @@ async def _sync_generation_inference(ctx, payload): msg = '上游返回业务错误: %s %s（request_id=%s）' % ( data.get('code'), str(data.get('message') or '')[:200], data.get('request_id') or '') - await govern_settle(ctx, False, 0, 0, msg[:200]) + await govern_settle(ctx, False, msg[:200]) raise GovernError(msg) out = data.get('output') or {} @@ -543,14 +569,14 @@ async def _sync_generation_inference(ctx, payload): if not content: msg = ('同步生成调用成功但未在响应中找到产物（模板渲染与形态搜索均未命中）。' '上游响应：%s' % json.dumps(data, ensure_ascii=False, default=str)[:300]) - await govern_settle(ctx, False, 0, 0, msg[:200]) + await govern_settle(ctx, False, msg[:200]) raise GovernError(msg) usages = dict(usage) if isinstance(usage, dict) else {} usages['model'] = ctx.get('model_id') or '' for k in ('resolution', 'size', 'duration'): if payload.get(k) not in (None, ''): usages.setdefault(k, payload.get(k)) - await govern_settle(ctx, True, 0, 0, '' if usage else 'no-usage', usages) + await govern_settle(ctx, True, '' if usage else 'no-usage', usages) result = { "choices": [{"message": {"content": content, "role": "assistant"}, "finish_reason": "stop"}], @@ -656,7 +682,7 @@ async def _http_request(method, url, headers, body, timeout): raise ValueError('上游请求失败（%s）：%s' % (type(e).__name__, repr(e))) -async def _async_query_once(ctx, qprofile, task_id, ns, timeout): +async def _async_query_once(ctx, qprofile, task_id, ns, timeout, seq=0): """按查询步骤模板请求一次任务状态。path 支持 {task_id} 占位符。""" path = await _render_tmpl((qprofile.get('path') or '').strip(), ns) path = path.replace('{task_id}', task_id) @@ -666,7 +692,16 @@ async def _async_query_once(ctx, qprofile, task_id, ns, timeout): url = _join_url(ctx.get('api_base'), path) headers = await _render_headers(qprofile.get('headers', ''), ns) method = (qprofile.get('method') or 'GET').upper() - status, data = await _http_request(method, url, headers, None, timeout) + _t0 = time.time() + try: + status, data = await _http_request(method, url, headers, None, timeout) + except ValueError as e: + await _trace_rt(ctx, seq, 'query', url, method, headers, + {'task_id': task_id}, 0, None, _t0, err=str(e)[:200]) + raise + await _trace_rt(ctx, seq, 'query', url, method, headers, + {'task_id': task_id}, status, data, _t0, + err='' if status == 200 else 'HTTP %d' % status) if status != 200: raise ValueError('任务查询失败 HTTP %d: %s' % ( status, str(data)[:200])) @@ -686,7 +721,7 @@ async def _async_inference(ctx, payload, req_timeout): from .gateway import govern_settle async def _fail(msg): - await govern_settle(ctx, False, 0, 0, msg) + await govern_settle(ctx, False, msg) raise GovernError(msg) qprofiles = ctx.get('query_profiles') or [] @@ -707,11 +742,17 @@ async def _async_inference(ctx, payload, req_timeout): url = _join_url(ctx.get('api_base'), path) headers = await _render_headers(profile.get('headers', ''), ns) submit_method = (profile.get('method') or 'POST').upper() + _t0 = time.time() try: status, data = await _http_request(submit_method, url, headers, body, step_timeout) except ValueError as e: + await _trace_rt(ctx, 0, 'submit', url, submit_method, headers, body, + 0, None, _t0, err=str(e)[:200]) await _fail('异步任务提交失败：%s' % e) return {} + await _trace_rt(ctx, 0, 'submit', url, submit_method, headers, body, + status, data, _t0, + err='' if status == 200 else 'HTTP %d' % status) if status != 200 or not isinstance(data, dict): await _fail('异步任务提交失败 HTTP %d: %s' % (status, str(data)[:300])) return {} @@ -728,11 +769,14 @@ async def _async_inference(ctx, payload, req_timeout): interval = _ASYNC_POLL_START last = data finished = '' + _poll_seq = 0 # trace 往返序号：提交=0，轮询从 1 递增（原文追踪每次往返各一行） while time.time() < deadline: await asyncio.sleep(interval) interval = min(interval * 1.5, _ASYNC_POLL_MAX) + _poll_seq += 1 try: - last = await _async_query_once(ctx, qprofiles[0], task_id, ns, step_timeout) + last = await _async_query_once(ctx, qprofiles[0], task_id, ns, + step_timeout, seq=_poll_seq) except ValueError as e: # 单次查询失败不立即终止（可能瞬时抖动），预算内继续 print("[inference] 异步查询瞬时失败（预算内继续）: %s" % e) @@ -782,7 +826,7 @@ async def _async_inference(ctx, payload, req_timeout): for k in ('resolution', 'size', 'duration'): if payload.get(k) not in (None, ''): usages.setdefault(k, payload.get(k)) - await govern_settle(ctx, True, 0, 0, '' if usage else 'no-usage', usages) + await govern_settle(ctx, True, '' if usage else 'no-usage', usages) result = { "choices": [{"message": {"content": content, "role": "assistant"}, "finish_reason": "stop"}], @@ -797,7 +841,11 @@ async def _async_inference(ctx, payload, req_timeout): # ────────────────────────── 对外主入口 ────────────────────────── -async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): +async def chat_inference(org_id, user_id, payload, model_name='', task_ref='', + project_id=''): + # project_id（2026-09-10 用户定夺）：token 绑定的项目 ID 透传落 llm_usage.project_id， + # 非项目调用不传（govern_settle 兜底哨兵'0'），支撑按项目统计费用。 + # 调用方（llm_proxy/v1 dspy）负责传入。 """统一推理：门禁链 → 上游调用 → 结算。返回上游响应 dict（OpenAI 兼容形态）。 payload: 客户端请求体（messages 必需；tools/temperature/max_tokens 透传； @@ -828,7 +876,13 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): 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) + sor, org_id, user_id or '', model_name, '', est, task_ref, purpose, + project_id=project_id or '') + # 调用批次 ID（2026-09-10 用户定夺：上行下行原文唯一落盘 + llm_usage 关联）： + # 一次 chat_inference = N 次上游往返（重试/异步轮询），trace 按 call_id+seq 组织， + # llm_usage.call_id 同值 → 流水可反查全部原文（trace.record_roundtrip / llm_call_trace） + from appPublic.uniqueID import getID + ctx['call_id'] = getID() if req_timeout > 0: ctx['timeout'] = min(req_timeout, 900) # 异步模型：提交→轮询→取结果（用量因子走 usages 记账） @@ -847,7 +901,7 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): return await _sync_generation_inference(ctx, payload) if cap not in CHAT_CAPS: from .gateway import govern_settle - await govern_settle(ctx, False, 0, 0, 'capability %s 无执行路径' % cap) + await govern_settle(ctx, False, 'capability %s 无执行路径' % cap) raise GovernError('模型「%s」能力为 %s：chat 入口支持对话（%s）与生成（%s）——' '%s 请走其专属入口' % ( model_row.get('name', ''), cap, '/'.join(CHAT_CAPS), @@ -857,13 +911,12 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): except Exception as e: # 失败结算：释放预授权 + failed 流水；上游 429 触发账号冷却 note = str(e)[:200] - await govern_settle(ctx, False, 0, 0, + await govern_settle(ctx, False, '429 upstream: ' + note if ' 429' in note else note) raise GovernError(str(e)) usage = data.get('usage') or {} - rt = int(_fnum(usage.get('prompt_tokens'))) - ct = int(_fnum(usage.get('completion_tokens'))) - # usage 原文落 llm_usage.usages（2026-09-07 根治，对齐异步路径）： + # usage 原文全量落 llm_usage.usages（2026-09-07 根治，对齐异步路径； + # 2026-09-10 起 usages 是计费唯一事实源，req/resp_tokens 列已删 m0021）： # 定价引擎的 derived 变量引用（cached_tokens=prompt_tokens_details.cached_tokens、 # uncache_tokens=prompt_tokens - prompt_tokens_details.cached_tokens）依赖原始 usage # 结构——此前同步路径只存 req/resp 两列、usages 为空，导致 prompt_tokens_details @@ -872,7 +925,7 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): 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) + await govern_settle(ctx, True, '' if usage else 'est', usages) return data diff --git a/pipeline_llm/init.py b/pipeline_llm/init.py index 3dc5a19..d80c96c 100644 --- a/pipeline_llm/init.py +++ b/pipeline_llm/init.py @@ -9,8 +9,10 @@ pipeline_service.llm_bridge / llm_proxy 承担——本模块在两个调用链 - llm_proxy.proxy_chat_completion：调用前后接门禁与结算（运行环境路径） """ +import html import json import logging +import os from sqlor.dbpools import DBPools from ahserver.serverenv import ServerEnv @@ -430,36 +432,64 @@ async def llm_usage_query(params_kw): result = {'success': False, 'message': ''} try: pk = params_kw or {} + # 条件列统一加 u. 前缀（2026-09-10）：JOIN llm_model/llm_account/sd_projects + # 后 org_id/status/created_at 多表同名，无前缀会触发 MariaDB 1052 歧义 conds = ["1=1"] params = {} if pk.get('org_id'): - conds.append("org_id=${o}$"); params['o'] = pk['org_id'] + conds.append("u.org_id=${o}$"); params['o'] = pk['org_id'] if pk.get('model_id'): - conds.append("model_id=${m}$"); params['m'] = pk['model_id'] + conds.append("u.model_id=${m}$"); params['m'] = pk['model_id'] if pk.get('account_id'): - conds.append("account_id=${a}$"); params['a'] = pk['account_id'] + conds.append("u.account_id=${a}$"); params['a'] = pk['account_id'] if pk.get('status'): - conds.append("status=${s}$"); params['s'] = pk['status'] + conds.append("u.status=${s}$"); params['s'] = pk['status'] + # 项目过滤（2026-09-10 用户定夺：按项目统计费用）； + # 不传/null = 不区分项目（全部），'0' = 精确查非项目调用 + _pid = pk.get('project_id') + if _pid: + conds.append("u.project_id=${pid}$"); params['pid'] = _pid if pk.get('date_from'): - conds.append("created_at>=${df}$"); params['df'] = pk['date_from'] + conds.append("u.created_at>=${df}$"); params['df'] = pk['date_from'] if pk.get('date_to'): - conds.append("created_at<=${dt}$"); params['dt'] = pk['date_to'] + ' 23:59:59' + conds.append("u.created_at<=${dt}$"); params['dt'] = pk['date_to'] + ' 23:59:59' where = " AND ".join(conds) db, dbname = _get_sor() async with db.sqlorContext(dbname) as sor: recs = await sor.sqlExe( + # token 数从 usages JSON 提取（req/resp_tokens 冗余列已删 m0021， + # usages 是唯一事实源）；JSON_VALID 守卫防脏行报错；出参键名保持 + # req_tokens/resp_tokens 兼容既有前端 "SELECT u.id, u.org_id, u.user_id, u.model_id, m.name AS model_name, " "u.account_id, a.name AS account_name, u.endpoint_region, " - "u.req_tokens, u.resp_tokens, u.cost, u.charge, u.ppid, u.task_ref, " + "CASE WHEN JSON_VALID(u.usages) THEN " + "COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.prompt_tokens')),'0') " + "ELSE '0' END AS req_tokens, " + "CASE WHEN JSON_VALID(u.usages) THEN " + "COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.completion_tokens')),'0') " + "ELSE '0' END AS resp_tokens, " + "u.cost, u.charge, u.ppid, u.project_id, " + "CASE WHEN u.project_id='0' THEN '（非项目调用）' " + "ELSE COALESCE(sp.name, '(项目已删除)') END AS project_name, " + "u.task_ref, u.call_id, " "u.status, u.note, u.created_at " "FROM llm_usage u " "LEFT JOIN llm_model m ON m.id=u.model_id " "LEFT JOIN llm_account a ON a.id=u.account_id " + "LEFT JOIN sd_projects sp ON sp.id=u.project_id " "WHERE %s ORDER BY u.created_at DESC LIMIT 200" % where, params) + # token 汇总走 JSON_EXTRACT（列已删 m0021）：CAST DECIMAL 防字符串数字； + # 非法/缺失 JSON 行按 0 计入，不中断汇总 sums = await sor.sqlExe( - "SELECT COUNT(*) AS cnt, COALESCE(SUM(req_tokens),0) AS req_t, " - "COALESCE(SUM(resp_tokens),0) AS resp_t, COALESCE(SUM(cost),0) AS cost_s, " - "COALESCE(SUM(charge),0) AS charge_s FROM llm_usage WHERE %s" % where, params) + "SELECT COUNT(*) AS cnt, " + "COALESCE(SUM(CASE WHEN JSON_VALID(usages) THEN " + "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(usages,'$.prompt_tokens')),'0') " + "AS DECIMAL(20,4)) ELSE 0 END),0) AS req_t, " + "COALESCE(SUM(CASE WHEN JSON_VALID(usages) THEN " + "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(usages,'$.completion_tokens')),'0') " + "AS DECIMAL(20,4)) ELSE 0 END),0) AS resp_t, " + "COALESCE(SUM(cost),0) AS cost_s, " + "COALESCE(SUM(charge),0) AS charge_s FROM llm_usage u WHERE %s" % where, params) await sor.sqlExe("COMMIT", {}) rows = [] for r in (recs or []): @@ -485,6 +515,247 @@ async def llm_usage_query(params_kw): return json.dumps(result, ensure_ascii=False, default=str) +async def llm_project_cost_query(params_kw): + """按项目统计费用（2026-09-10 用户定夺）：llm_usage.project_id GROUP BY。 + + 过滤：org_id / date_from / date_to / status（缺省统计全部状态）。 + 分组：project_id（'0' = 非项目调用哨兵值，一组显示'（非项目调用）'； + 其余 JOIN sd_projects 取项目名，项目已删显示'(项目已删除)'）。 + project_id 过滤：不传/null = 不区分项目（全部分组都出）。 + 汇总口径：calls 调用次数 / cost 成本合计 / charge 客户应付合计 / + req_t、resp_t token 合计（usages JSON 唯一事实源，JSON_VALID 守卫）。 + """ + result = {'success': False, 'message': ''} + try: + pk = params_kw or {} + conds = ["1=1"] + params = {} + if pk.get('org_id'): + conds.append("u.org_id=${o}$"); params['o'] = pk['org_id'] + if pk.get('status'): + conds.append("u.status=${s}$"); params['s'] = pk['status'] + if pk.get('date_from'): + conds.append("u.created_at>=${df}$"); params['df'] = pk['date_from'] + if pk.get('date_to'): + conds.append("u.created_at<=${dt}$"); params['dt'] = pk['date_to'] + ' 23:59:59' + where = " AND ".join(conds) + db, dbname = _get_sor() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT u.project_id, " + "CASE WHEN u.project_id='0' THEN '（非项目调用）' " + "ELSE COALESCE(sp.name, '(项目已删除)') END AS project_name, " + "COUNT(*) AS calls, " + "COALESCE(SUM(u.cost),0) AS cost_s, " + "COALESCE(SUM(u.charge),0) AS charge_s, " + "COALESCE(SUM(CASE WHEN JSON_VALID(u.usages) THEN " + "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.prompt_tokens')),'0') " + "AS DECIMAL(20,4)) ELSE 0 END),0) AS req_t, " + "COALESCE(SUM(CASE WHEN JSON_VALID(u.usages) THEN " + "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.completion_tokens')),'0') " + "AS DECIMAL(20,4)) ELSE 0 END),0) AS resp_t " + "FROM llm_usage u " + "LEFT JOIN sd_projects sp ON sp.id=u.project_id " + "WHERE %s GROUP BY u.project_id, sp.name " + "ORDER BY cost_s DESC, calls DESC LIMIT 500" % where, params) + await sor.sqlExe("COMMIT", {}) + rows = [] + for r0 in (recs or []): + try: + d = dict(r0) + except (TypeError, ValueError): + d = {k: v for k, v in vars(r0).items() if not callable(v)} + rows.append({ + 'project_id': d.get('project_id'), + 'project_name': d.get('project_name') or '', + 'calls': int(d.get('calls', 0) or 0), + 'cost': round(float(d.get('cost_s', 0) or 0), 6), + 'charge': round(float(d.get('charge_s', 0) or 0), 6), + 'req_tokens': int(d.get('req_t', 0) or 0), + 'resp_tokens': int(d.get('resp_t', 0) or 0), + }) + result['success'] = True + result['rows'] = rows + result['summary'] = { + 'project_count': len([x for x in rows + if x['project_id'] and x['project_id'] != '0']), + 'calls': sum(x['calls'] for x in rows), + 'cost': round(sum(x['cost'] for x in rows), 6), + 'charge': round(sum(x['charge'] for x in rows), 6), + } + except Exception as e: + result['message'] = str(e) + return json.dumps(result, ensure_ascii=False, default=str) + + +# ────────────────────── 调用原文追踪：「查看IO」弹窗 ────────────────────── + +_IO_BODY_MAX = 150000 # 弹窗单侧最大展示字符（完整原文仍在盘上，超长提示文件路径） + +_KIND_TEXT = {'chat': '对话', 'gen': '同步生成', 'submit': '异步提交', 'query': '任务轮询'} + + +def _read_trace_file(rel_path): + """读 trace 原文文件。rel_path 为库中 filesroot 相对路径（llm_trace/<日>/_.req.json）。 + + 安全铁律：路径来自数据库，realpath 后必须落在 llm_trace 目录内，逃逸=拒绝访问 + （防路径穿越；realpath 同时防符号链接指向目录外）。返回 (text, err)。 + """ + from .trace import _filesroot, trace_root + if not rel_path: + return '', '无原文文件（该侧落盘失败或已过保留期被清理）' + root = os.path.realpath(trace_root()) + real = os.path.realpath(os.path.join(_filesroot(), str(rel_path))) + if not real.startswith(root + os.sep): + return '', '[非法路径，拒绝访问]' + try: + with open(real, 'r', encoding='utf-8') as f: + return f.read(), '' + except Exception as e: + return '', '[文件读取失败: %s]' % str(e)[:120] + + +def _pretty_or_raw(text): + """落盘 JSON 美化（indent=2）便于阅读；非合法 JSON 原样返回。""" + try: + return json.dumps(json.loads(text), ensure_ascii=False, indent=2) + except Exception: + return text + + +def _msg_widget(message): + return {"widgettype": "Message", + "options": {"title": "提示", "message": message, + "cwidth": 24, "cheight": 8, "timeout": 3}} + + +async def llm_call_trace_io(params_kw, user_org_id='0'): + """「查看IO」弹窗内容组装：TabPanel（上下文摘要/请求原文/响应原文）。 + + dspy 薄代理调本函数（读表+读文件+组装 widget 全在模块侧）。 + 机构隔离：平台业主（org_id='0'）可看全部；普通机构只看本机构或 org_id 为空的行。 + 返回 widget 描述 dict（外层 PopupWindow 壳由列表页 binds popup_options 生成， + 此处再包 PopupWindow 会套娃）。 + """ + pk = params_kw or {} + trace_id = str(pk.get('id') or '').strip() + if not trace_id or len(trace_id) > 32: + return _msg_widget('缺少或非法的 trace id 参数') + + db, dbname = _get_sor() + async with db.sqlorContext(dbname) as sor: + recs = await sor.sqlExe( + "SELECT * FROM llm_call_trace WHERE id=${id}$", {'id': trace_id}) + row = dict(recs[0]) if recs else None + batch_rows, usage_rows = [], [] + if row and row.get('call_id'): + call_id = row['call_id'] + recs2 = await sor.sqlExe( + "SELECT seq, kind, method, status_code, elapsed_ms, url, note " + "FROM llm_call_trace WHERE call_id=${c}$ ORDER BY seq, created_at", + {'c': call_id}) + batch_rows = [dict(r) for r in (recs2 or [])] + recs3 = await sor.sqlExe( + "SELECT u.status, u.accounting_status, u.cost, u.charge, u.task_ref, " + "u.project_id, m.name AS model_name " + "FROM llm_usage u LEFT JOIN llm_model m ON m.id=u.model_id " + "WHERE u.call_id=${c}$", {'c': call_id}) + usage_rows = [dict(r) for r in (recs3 or [])] + await sor.sqlExe("COMMIT", {}) + + if not row: + return _msg_widget('未找到调用追踪记录（可能已过保留期被清理）') + row_org = str(row.get('org_id') or '') + if str(user_org_id or '0') != '0' and row_org not in ('', str(user_org_id)): + return _msg_widget('无权查看其他机构的调用原文') + + # 读文件在 DB 块外（行/批次/usage 已全部拷贝为本地 dict，不再依赖 sor） + req_text, req_err = _read_trace_file(row.get('req_path') or '') + resp_text, resp_err = _read_trace_file(row.get('resp_path') or '') + if req_err: + req_text = req_err + else: + req_text = _pretty_or_raw(req_text) + if len(req_text) > _IO_BODY_MAX: + req_text = req_text[:_IO_BODY_MAX] + \ + '\n\n...［内容过长已截断，完整原文见服务器文件: %s］' % (row.get('req_path') or '') + if resp_err: + resp_text = resp_err + else: + resp_text = _pretty_or_raw(resp_text) + if len(resp_text) > _IO_BODY_MAX: + resp_text = resp_text[:_IO_BODY_MAX] + \ + '\n\n...［内容过长已截断，完整原文见服务器文件: %s］' % (row.get('resp_path') or '') + + # ── 摘要 tab（本行 + 计费流水上下文 + 同批次往返）── + lines = [ + '批次 call_id: %s' % (row.get('call_id') or ''), + '往返 seq/类型: %s / %s' % ( + row.get('seq'), _KIND_TEXT.get(str(row.get('kind') or ''), row.get('kind') or '')), + 'HTTP: %s 状态码: %s 耗时: %s ms' % ( + row.get('method') or '', row.get('status_code'), row.get('elapsed_ms')), + 'URL: %s' % (row.get('url') or ''), + '机构: %s' % (row_org or '(空)'), + '时间: %s' % (row.get('created_at') or ''), + ] + if row.get('note'): + lines.append('错误摘要: %s' % row['note']) + if usage_rows: + lines.append('') + lines.append('—— 计费流水（llm_usage，同 call_id，%d 条）——' % len(usage_rows)) + for u in usage_rows: + lines.append('模型: %s 状态: %s 记账: %s 成本: %s 应付: %s 来源: %s 项目: %s' % ( + u.get('model_name') or '(无)', u.get('status') or '', + u.get('accounting_status') or '-', u.get('cost'), u.get('charge'), + u.get('task_ref') or '-', u.get('project_id') or '-')) + if batch_rows: + lines.append('') + lines.append('—— 同批次往返（%d 次）——' % len(batch_rows)) + for t in batch_rows: + lines.append('seq=%s %s %s 状态:%s %sms %s%s' % ( + t.get('seq'), + _KIND_TEXT.get(str(t.get('kind') or ''), t.get('kind') or ''), + t.get('method') or '', t.get('status_code'), t.get('elapsed_ms'), + (t.get('url') or '')[:100], + (' err: ' + str(t.get('note'))[:80]) if t.get('note') else '')) + summary_text = '\n'.join(lines) + + # 摘要 tab 用 Html widget：Text 的 innerHTML 会吞换行且不转义（正文含上游 + # 响应片段时 < > 会破坏结构/注入），Html + 
 + escape 是唯一安全渲染。
+    summary_html = ('
'
+                    + html.escape(summary_text) + '
')
+
+    # mode:null → CodeMirror 纯文本+行号（仓库 3parties 只有 codemirror.js
+    # 内建 null 模式和 codemirror-python.js；不指定 mode 会去拉不存在的
+    # codemirror-python.js 之外的 mode 文件报 warn，null 是最干净的退化）。
+    def _code_tab(name, label, content):
+        return {"name": name, "label": label,
+                "content": {"widgettype": "CodeEditor",
+                            "options": {"value": content, "mode": "null",
+                                        "readonly": True, "theme": False,
+                                        "width": "100%", "height": "100%"}}}
+
+    return {
+        "widgettype": "TabPanel",
+        "options": {
+            "width": "100%", "height": "100%", "tab_pos": "top",
+            "items": [
+                {"name": "summary", "label": "上下文摘要",
+                 "content": {"widgettype": "VBox",
+                             "options": {"width": "100%", "height": "100%",
+                                         "overflow": "auto", "padding": "12px"},
+                             "subwidgets": [
+                                 {"widgettype": "Html",
+                                  "options": {"width": "100%", "height": "auto",
+                                              "html": summary_html}}]}},
+                _code_tab('req', '请求原文（上行）', req_text),
+                _code_tab('resp', '响应原文（下行）', resp_text),
+            ]
+        }
+    }
+
+
 async def llm_dashboard(params_kw):
     """治理总览（首页卡片数据源）：各机构策略/余额/近期用量。"""
     result = {'success': True}
@@ -529,6 +800,8 @@ def load_pipeline_llm():
     env.delete_llm_model = delete_llm_model
     env.update_llm_org_policy = update_llm_org_policy
     env.llm_usage_query = llm_usage_query
+    env.llm_project_cost_query = llm_project_cost_query
+    env.llm_call_trace_io = llm_call_trace_io
     env.llm_dashboard = llm_dashboard
     # 产品层标准接口（product_management.product_interface 调度）
     from .product_interface import (
diff --git a/pipeline_llm/product_interface.py b/pipeline_llm/product_interface.py
index 8c8a49d..6e62d3d 100644
--- a/pipeline_llm/product_interface.py
+++ b/pipeline_llm/product_interface.py
@@ -109,7 +109,9 @@ async def execute_product_service_stream(resource_ref_id, user_id, user_org_id,
 async def calculate_product_cost(resource_ref_id, usage_data, user_org_id=None):
     """资源层真实定价（原价）。不碰折扣（折扣由产品层按产品精确计算）。
 
-    usage_data: {'prompt_tokens': N, 'completion_tokens': N}（或 req_tokens/resp_tokens）
+    usage_data: {'prompt_tokens': N, 'completion_tokens': N}（或按量因子 dict；
+    键名唯一契约 = 上游 usage 原文结构，2026-09-10 起 req_tokens/resp_tokens 别名
+    随冗余列一并废除）
     """
     m, err = await _resolve_model(resource_ref_id)
     if err:
@@ -121,11 +123,6 @@ async def calculate_product_cost(resource_ref_id, usage_data, user_org_id=None):
     if not hasattr(env, 'buffered_charging'):
         return {'success': False, 'message': '定价引擎（pricing 模块）未加载'}
     usage = dict(usage_data or {})
-    # 兼容两种键名
-    if 'prompt_tokens' not in usage and 'req_tokens' in usage:
-        usage['prompt_tokens'] = usage['req_tokens']
-    if 'completion_tokens' not in usage and 'resp_tokens' in usage:
-        usage['completion_tokens'] = usage['resp_tokens']
     try:
         prices = await env.buffered_charging(ppid, usage)
     except Exception as e:
diff --git a/pipeline_llm/trace.py b/pipeline_llm/trace.py
new file mode 100644
index 0000000..0a1a065
--- /dev/null
+++ b/pipeline_llm/trace.py
@@ -0,0 +1,192 @@
+"""pipeline_llm.trace — 模型调用上行/下行原文追踪（2026-09-10 用户定夺）。
+
+存储结构（唯一收敛点：所有路径的上游 HTTP 往返都经 inference._post_upstream /
+_http_request，在这一层各记一次，全链唯一）：
+  文件：filesroot/llm_trace//_.req.json / .resp.json
+  库：  llm_call_trace 一行一次上游往返（重试/异步轮询的中间轮次都落，1:N）
+
+call_id 关联链：chat_inference 入口生成 → ctx['call_id'] → 每次往返落 trace 行
+→ govern_settle 写 llm_usage.call_id（同值）。按 usage 流水即可反查全部原文。
+
+铁律：
+  1. 脱敏后落盘——Authorization/api-key 等敏感 header 值替换为 [REDACTED]
+     （真 key 不出进程；文件不得成为密钥泄露面）。URL 去 query（可能带签名）。
+  2. 先写文件、后写库行（行含文件路径）；任何失败只记日志绝不阻断调用——
+     审计尽力而为，禁止反噬计费/调用链。record_roundtrip 永不抛异常。
+  3. 保留期：TRACE_RETENTION_DAYS 天，cleanup_expired 删过期日期目录+库行
+     （记账 worker 循环每日调一次）。文件须纳入部署备份（dbackup 只备库，
+     filesroot/llm_trace 需另行同步备份，否则"唯一存储"不成立）。
+  4. 下载走专用端点 wwwroot/api/llm_trace_file.dspy（登录 + 机构隔离 +
+     路径必须落在 llm_trace 目录内）。禁止复用 /idfile——那是登录即可的
+     全平台通用口，会把跨机构会话原文泄露给任意登录用户。
+"""
+
+import json
+import logging
+import os
+import shutil
+from datetime import datetime, timedelta
+
+logger = logging.getLogger("pipeline_llm.trace")
+
+_TRACE_DIR = 'llm_trace'
+TRACE_RETENTION_DAYS = 90
+
+# 敏感 header 关键词（子串匹配，全小写比较）：命中即值替换 [REDACTED]
+_SENSITIVE_HEADER_KEYS = (
+    'authorization', 'api-key', 'apikey', 'x-api-key', 'proxy-authorization',
+    'cookie', 'x-goog-api-key', 'x-auth-token', 'x-dashscope-apikey',
+    'x-aliyun-api-key', 'token',
+)
+
+
+def _filesroot():
+    from appPublic.jsonConfig import getConfig
+    root = None
+    try:
+        root = getConfig().filesroot
+    except Exception:
+        root = None
+    return os.path.abspath(root or __import__('tempfile').gettempdir())
+
+
+def trace_root():
+    """llm_trace 目录绝对路径（下载端点做路径穿越校验用）。"""
+    return os.path.join(_filesroot(), _TRACE_DIR)
+
+
+def _redact_headers(headers):
+    out = {}
+    for k, v in (headers or {}).items():
+        lk = str(k).lower()
+        if any(s in lk for s in _SENSITIVE_HEADER_KEYS):
+            out[k] = '[REDACTED]'
+        else:
+            out[k] = v
+    return out
+
+
+def _dump(obj):
+    try:
+        return json.dumps(obj, ensure_ascii=False, default=str, indent=None)
+    except Exception:
+        return str(obj)[:65535]
+
+
+async def _write_file(rel, content):
+    import aiofiles
+    fp = os.path.join(_filesroot(), rel)
+    os.makedirs(os.path.dirname(fp), exist_ok=True)
+    async with aiofiles.open(fp, 'w', encoding='utf-8') as f:
+        await f.write(content)
+
+
+async def record_roundtrip(ctx, seq, kind, url, method, headers, req_body,
+                           status_code, resp_data, elapsed_ms, err=''):
+    """一次上游 HTTP 往返 → 请求/响应两个文件 + 一行 llm_call_trace。
+
+    ctx       govern_resolve 返回的上下文（须含 call_id；缺 call_id 直接跳过——
+              没有关联键的孤儿 trace 无意义）
+    seq       往返序号（同步重试 0..2；异步提交 0，轮询 1..N）
+    kind      chat / submit / query
+    resp_data 响应体（dict/str 均可；网络异常时 None）
+    err       错误摘要（网络异常/非200 等）
+
+    永不抛出（铁律 2）。
+    """
+    try:
+        ctx = ctx or {}
+        call_id = str(ctx.get('call_id') or '').strip()
+        if not call_id:
+            return
+        day = datetime.now().strftime('%Y%m%d')
+        base = '%s/%s/%s_%03d' % (_TRACE_DIR, day, call_id, int(seq or 0))
+        # URL 去 query（可能带临时签名参数，属敏感信息）
+        safe_url = str(url or '').split('?')[0][:500]
+        req_path = ''
+        resp_path = ''
+        try:
+            req_rel = base + '.req.json'
+            await _write_file(req_rel, _dump({
+                'url': safe_url, 'method': method, 'kind': kind,
+                'headers': _redact_headers(headers),
+                'body': req_body,
+            }))
+            req_path = req_rel
+        except Exception as e:
+            # 请求文件失败不阻断响应文件与库行（部分留痕优于无痕）
+            print("[trace] 请求文件写入失败(call=%s seq=%s): %s" % (call_id, seq, e))
+        try:
+            resp_rel = base + '.resp.json'
+            await _write_file(resp_rel, _dump({
+                'status_code': int(status_code or 0),
+                'elapsed_ms': int(elapsed_ms or 0),
+                'error': err or '',
+                'body': resp_data,
+            }))
+            resp_path = resp_rel
+        except Exception as e:
+            print("[trace] 响应文件写入失败(call=%s seq=%s): %s" % (call_id, seq, e))
+        # 库行（先文件后行：行存在 ⇒ 文件已尝试写入；路径为空=该侧写失败）
+        from appPublic.uniqueID import getID
+        from .gateway import _get_db
+        db, dbname = _get_db()
+        async with db.sqlorContext(dbname) as sor:
+            await sor.C('llm_call_trace', {
+                'id': getID(),
+                'call_id': call_id,
+                'seq': int(seq or 0),
+                'kind': str(kind or '')[:20],
+                'url': safe_url,
+                'method': str(method or '')[:10],
+                'status_code': int(status_code or 0),
+                'elapsed_ms': int(elapsed_ms or 0),
+                'req_path': req_path,
+                'resp_path': resp_path,
+                'org_id': ctx.get('org_id') or '',
+                'note': str(err or '')[:200],
+            })
+            await sor.sqlExe("COMMIT", {})
+    except Exception as e:
+        logger.warning("pipeline_llm.trace: trace 记录失败（不阻断调用）: %s", e)
+
+
+async def cleanup_expired(days=None):
+    """清理过期 trace：按日期目录整删文件 + 删对应库行。返回 (文件目录数, 库行数)。
+
+    目录名即日期（YYYYMMDD），过期判定以目录名为准（不逐文件 stat）；
+    库行按 created_at 删。目录删失败（权限/占用）不阻断库行清理。
+    """
+    days = int(days or TRACE_RETENTION_DAYS)
+    cutoff_day = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
+    cutoff_ts = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d %H:%M:%S')
+    removed_dirs = 0
+    root = trace_root()
+    try:
+        for name in sorted(os.listdir(root)):
+            if not (name.isdigit() and len(name) == 8):
+                continue
+            if name < cutoff_day:
+                try:
+                    shutil.rmtree(os.path.join(root, name))
+                    removed_dirs += 1
+                except Exception as e:
+                    logger.warning("pipeline_llm.trace: 清理目录 %s 失败: %s", name, e)
+    except FileNotFoundError:
+        pass
+    removed_rows = 0
+    try:
+        from .gateway import _get_db
+        db, dbname = _get_db()
+        async with db.sqlorContext(dbname) as sor:
+            await sor.sqlExe(
+                "DELETE FROM llm_call_trace WHERE created_at<${c}$", {"c": cutoff_ts})
+            await sor.sqlExe("COMMIT", {})
+            recs = await sor.sqlExe("SELECT ROW_COUNT() AS n", {})
+            removed_rows = int(getattr(recs[0], 'n', 0)) if recs else 0
+    except Exception as e:
+        logger.warning("pipeline_llm.trace: 清理库行失败: %s", e)
+    if removed_dirs or removed_rows:
+        logger.info("pipeline_llm.trace: 清理过期 trace 目录 %d 个、库行 %d 条（保留 %d 天）",
+                    removed_dirs, removed_rows, days)
+    return removed_dirs, removed_rows
diff --git a/wwwroot/api/llm_call_trace_io.dspy b/wwwroot/api/llm_call_trace_io.dspy
new file mode 100644
index 0000000..8751643
--- /dev/null
+++ b/wwwroot/api/llm_call_trace_io.dspy
@@ -0,0 +1,10 @@
+# llm_call_trace_io.dspy — 「查看IO」弹窗内容（薄代理；业务逻辑在 pipeline_llm.init.llm_call_trace_io）
+# 入参: id（trace 行主键，列表页 toolbar selected_row 自动注入行数据）
+# 返回: TabPanel（上下文摘要/请求原文/响应原文）——外层 PopupWindow 壳由列表页
+#       binds 的 popup_options 生成，此处只返回内容 widget（返回 PopupWindow 会套娃）
+uid = await get_user()
+if not uid:
+    return {"widgettype": "Message", "options": {"title": "提示", "message": "请先登录", "cwidth": 20, "cheight": 8, "timeout": 3}}
+user_org = await get_userorgid() or '0'
+result = await llm_call_trace_io(dict(params_kw or {}), str(user_org))
+return result
diff --git a/wwwroot/api/llm_project_cost_query.dspy b/wwwroot/api/llm_project_cost_query.dspy
new file mode 100644
index 0000000..4247ad0
--- /dev/null
+++ b/wwwroot/api/llm_project_cost_query.dspy
@@ -0,0 +1,11 @@
+# llm_project_cost_query.dspy — 按项目统计费用（2026-09-10 用户定夺）
+# 入参 params_kw: org_id? / date_from? / date_to? / status?
+# 出参: rows[{project_id, project_name, calls, cost, charge, req_tokens, resp_tokens}]
+#       + summary{project_count, calls, cost, charge}
+# project_id 过滤：不传/null = 不区分项目（全部分组都出）
+# 非项目调用（哨兵 project_id='0'）归为「（非项目调用）」一组
+uid = await get_user()
+if not uid:
+    return {"success": False, "message": "未登录"}
+pk = dict(params_kw or {})
+return json.loads(await llm_project_cost_query(pk))
diff --git a/wwwroot/api/v1/chat/completions.dspy b/wwwroot/api/v1/chat/completions.dspy
index 815ef0f..b47c567 100644
--- a/wwwroot/api/v1/chat/completions.dspy
+++ b/wwwroot/api/v1/chat/completions.dspy
@@ -66,7 +66,8 @@ try:
     data = await llm_chat_inference(
         org_id, user_id, payload,
         model_name=(info.get('model_name') or payload.get('model') or ''),
-        task_ref=task_ref)
+        task_ref=task_ref,
+        project_id=info.get('project_id') or '')
     # 回填 token 用量
     try:
         await record_llm_token_usage(info.get('id', ''), data.get('usage') or {})