yumoqing/pipeline-llm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(usage): usages全量不截断为计费唯一事实源(m0021删token冗余列)+llm_call_trace上下行原文追踪+project_id按项目统计费用(2026-09-10)

Browse Source
This commit is contained in:
yumoqing 2026-09-11 12:51:31 +08:00
parent d086c4f7d2
commit 427cd7cbc1
16 changed files with 1005 additions and 243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
README.md
View File

@ -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 回填，查询步骤兼容回退解析。

6
docs/design-spec.md
View File

@ -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) | |

76
json/llm_call_trace_list.json Normal file
View File

@ -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}$"
}
}
]
}
}

14
json/llm_usage_list.json
View File

@ -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
}
}
]
}
}

111
models/llm_call_trace.json Normal file
View File

@ -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": []
}

337
models/llm_usage.json
View File

@ -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'"
}
]
}

9
mysql.ddl.sql
View File

@ -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='用量流水(双维度记账)';

38
pipeline_llm/accounting.py
View File

@ -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)

35
pipeline_llm/gateway.py
View File

@ -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': '',

93
pipeline_llm/inference.py
View File

@ -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

293
pipeline_llm/init.py
View File

@ -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/<日>/<call>_<seq>.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 + <pre> + escape 是唯一安全渲染。
summary_html = ('<pre style="white-space:pre-wrap;word-break:break-all;'
'font-size:13px;margin:0;font-family:inherit;">'
+ html.escape(summary_text) + '</pre>')
# 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 (

9
pipeline_llm/product_interface.py
View File

@ -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:

192
pipeline_llm/trace.py Normal file
View File

@ -0,0 +1,192 @@
"""pipeline_llm.trace — 模型调用上行/下行原文追踪（2026-09-10 用户定夺）。
存储结构（唯一收敛点：所有路径的上游 HTTP 往返都经 inference._post_upstream /
_http_request，在这一层各记一次，全链唯一）：
文件：filesroot/llm_trace/<YYYYMMDD>/<call_id>_<seq>.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

10
wwwroot/api/llm_call_trace_io.dspy Normal file
View File

@ -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

11
wwwroot/api/llm_project_cost_query.dspy Normal file
View File

@ -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))

3
wwwroot/api/v1/chat/completions.dspy
View File

@ -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 {})