feat(llm): status统一枚举SUCCEEDED/FAILED/PENDING/RUNNING+非SUCCEEDED记账状态NULL;llm_api_profile加method独立列(2026-09-06用户定夺)
- llm_usage.status 各家归一枚举(其他=recharge对账行);accounting_status仅SUCCEEDED行写created,其余NULL不冒充待记账 - govern_settle失败行不再挂created;充值行不写accounting_status - 出账循环拾取条件 status='SUCCEEDED';面板/概览统计同步改 - llm_api_profile.method GET/POST独立列(默认POST),运行时读列决定动词;查询步骤存量兼容回退解析request_template.$.method - 字典llm_usage_status更新五枚举;README同步
This commit is contained in:
parent
825dd2e130
commit
671d320d59
24
README.md
24
README.md
@ -19,11 +19,11 @@
|
||||
| `llm_vendor` | 供应商（端点目录 JSON；2026-09-05 起不存协议字段，请求形态归模型挂的适配模板） |
|
||||
| `llm_account` | 供应商账号（钱包：余额/充值/选端点，api_key AES 加密；模型可绑定账号 account_id 收窄候选池） |
|
||||
| `llm_model` | 模型注册（能力分类/挂定价 ppid/sync_mode/query_profile_ids，name 唯一；2026-09 起不存单价，定价走 ppid） |
|
||||
| `llm_api_profile` | 适配模板（协议×能力形态去重；data/response 模板全 Jinja2，含媒体转换函数） |
|
||||
| `llm_api_profile` | 适配模板（协议×能力形态去重；data/response 模板全 Jinja2，含媒体转换函数；method 独立列，2026-09-06 起不塞 request_template） |
|
||||
| `llm_org_policy` | 组织容错策略（主备模型 + 辅助模型 utility_model_id + 端点偏好） |
|
||||
| `llm_org_quota` | 组织限额限流（池余额 + QPM） |
|
||||
| `llm_user_quota` | 个人限额限流（额度上限 + QPM） |
|
||||
| `llm_usage` | 用量流水（双维度记账事实表；usages 字段存非 token 计价因子 JSON） |
|
||||
| `llm_usage` | 用量流水（双维度记账事实表；status 统一枚举 SUCCEEDED/FAILED/PENDING/RUNNING/recharge；accounting_status 仅 SUCCEEDED 行有意义否则 NULL；usages 字段存非 token 计价因子 JSON） |
|
||||
|
||||
## 调用门禁链
|
||||
|
||||
@ -60,17 +60,23 @@
|
||||
- **限流用 Redis 分钟窗口原子计数**（db4，与会话 db3 隔离）；Redis 故障时限流放行（fail-open），
|
||||
余额仍由 DB 条件 UPDATE 守住。
|
||||
- **429 → (账号,端点)对冷却 60s**；账号级冷却用 `*` 通配键。
|
||||
- **记账三态（2026-09-05 用户定夺）**：`accounting_status` 只有三个值——
|
||||
`created`（已创建待记账）→ `accounted`（记账成功）/ `failed`（记账失败）。
|
||||
所有成功调用统一 created 进出账队列；自用 vs 跨机构分流由产品层
|
||||
`product_accounting_generic` 的 is_self_use 决定（自用只记 PAY* 采购成本，
|
||||
- **status 统一枚举 + 记账三态（2026-09-05 定夺；2026-09-06 补 NULL 语义，m0012）**：
|
||||
`llm_usage.status` 各家模型归一为 SUCCEEDED/FAILED/PENDING/RUNNING（其他=recharge 充值对账行）。
|
||||
`accounting_status` 只有三个值——`created`（待记账）→ `accounted`（记账成功）/ `failed`（记账失败）；
|
||||
**只有 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 并写明原因（不静默）。
|
||||
- **媒体转换铁律（2026-09-05）**：生成类模型适配模板 data 用
|
||||
`b64media2url(request, xxx_file)`（上传转公网 URL）、response 用
|
||||
- **method 独立列（2026-09-06 用户定夺，m0012）**：`llm_api_profile.method`（GET/POST，
|
||||
默认 POST）不塞 request_template JSON；运行时读该列决定 HTTP 动词，存量行由迁移
|
||||
从 request_template 的 $.method 回填，查询步骤兼容回退解析。
|
||||
- **媒体转换铁律（2026-09-06 三数组契约，对齐 sage/llmage）**：生成类模型适配模板
|
||||
data 用 `b64media2url(request, _f)` 逐元素转公网 URL、response 用
|
||||
`downloadfile2url(request, url)`（生成物落地本地，上游 URL 仅 24 小时）；
|
||||
同类能力对外契约统一（i2v：image_file 进 → video 出）。
|
||||
上行媒体统一三数组参数 `image_files`/`audio_files`/`video_files`（字符串或数组
|
||||
均可，Jinja `is string` 动态判断），变长输入用 `{% for %}` 按文档示例组装。
|
||||
- **异步模型执行器**：sync_mode=async 时走提交→按 query_profile_ids 轮询→
|
||||
取结果；等待预算默认 600 秒，`_timeout` 可覆盖（上限 900）。
|
||||
|
||||
|
||||
@ -34,8 +34,10 @@
|
||||
{"k": "must_intl", "v": "仅国际(合规)"}
|
||||
]},
|
||||
{"parentid": "llm_usage_status", "parentname": "用量状态", "items": [
|
||||
{"k": "ok", "v": "成功"},
|
||||
{"k": "failed", "v": "失败"},
|
||||
{"k": "SUCCEEDED", "v": "成功"},
|
||||
{"k": "FAILED", "v": "失败"},
|
||||
{"k": "PENDING", "v": "等待"},
|
||||
{"k": "RUNNING", "v": "运行中"},
|
||||
{"k": "recharge", "v": "充值"}
|
||||
]}
|
||||
]
|
||||
|
||||
@ -45,6 +45,14 @@
|
||||
"type": "str",
|
||||
"length": 200
|
||||
},
|
||||
{
|
||||
"name": "method",
|
||||
"title": "HTTP方法(GET/POST)",
|
||||
"type": "str",
|
||||
"length": 10,
|
||||
"nullable": "no",
|
||||
"default": "'POST'"
|
||||
},
|
||||
{
|
||||
"name": "headers",
|
||||
"title": "请求头(JSON)",
|
||||
|
||||
@ -77,11 +77,10 @@
|
||||
},
|
||||
{
|
||||
"name": "accounting_status",
|
||||
"title": "记账状态(created已创建待记账/accounted记账成功/failed记账失败)",
|
||||
"title": "记账状态(created待记账/accounted记账成功/failed记账失败;仅status=SUCCEEDED时有意义,否则NULL)",
|
||||
"type": "str",
|
||||
"length": 20,
|
||||
"nullable": "no",
|
||||
"default": "'created'"
|
||||
"nullable": "yes"
|
||||
},
|
||||
{
|
||||
"name": "ppid",
|
||||
@ -97,11 +96,11 @@
|
||||
},
|
||||
{
|
||||
"name": "status",
|
||||
"title": "状态",
|
||||
"title": "调用状态(SUCCEEDED/FAILED/PENDING/RUNNING/其他如recharge)",
|
||||
"type": "str",
|
||||
"length": 20,
|
||||
"nullable": "no",
|
||||
"default": "'ok'"
|
||||
"default": "'SUCCEEDED'"
|
||||
},
|
||||
{
|
||||
"name": "usages",
|
||||
|
||||
@ -59,6 +59,7 @@ CREATE TABLE IF NOT EXISTS llm_api_profile (
|
||||
`protocol` varchar(20) NOT NULL comment '协议类型',
|
||||
`capability` varchar(20) NOT NULL DEFAULT 't2t' comment '能力类型',
|
||||
`path` varchar(200) comment '请求路径(相对端点base_url,如/chat/completions)',
|
||||
`method` varchar(10) NOT NULL DEFAULT 'POST' comment 'HTTP方法(GET/POST,独立列不再塞request_template)',
|
||||
`headers` text comment '请求头(JSON,含认证头模板)',
|
||||
`request_template` text comment '请求模板(Jinja2)',
|
||||
`response_template` text comment '响应模板(Jinja2)',
|
||||
@ -121,10 +122,10 @@ CREATE TABLE IF NOT EXISTS llm_usage (
|
||||
`resp_tokens` int NOT NULL DEFAULT 0 comment '响应token',
|
||||
`cost` decimal(12,6) NOT NULL DEFAULT 0 comment '成本金额(账号侧)',
|
||||
`charge` decimal(12,6) NOT NULL DEFAULT 0 comment '客户应付(异步三方账计算,本机构模型恒0)',
|
||||
`accounting_status` varchar(20) NOT NULL DEFAULT 'created' comment '记账状态(三态:created已创建待记账/accounted记账成功/failed记账失败)',
|
||||
`accounting_status` varchar(20) comment '记账状态(三态:created待记账/accounted记账成功/failed记账失败;仅status=SUCCEEDED的行有意义,其余NULL)',
|
||||
`ppid` varchar(32) comment '定价项目ID',
|
||||
`task_ref` varchar(100) comment '调用来源',
|
||||
`status` varchar(20) NOT NULL DEFAULT 'ok' comment '状态(ok/failed/recharge)',
|
||||
`status` varchar(20) NOT NULL DEFAULT 'SUCCEEDED' comment '调用状态(统一枚举:SUCCEEDED成功/FAILED失败/PENDING等待/RUNNING运行;其他如recharge)',
|
||||
`usages` text comment '非token用量因子JSON(时长/分辨率等,异步出账读入定价引擎)',
|
||||
`note` varchar(200) comment '备注/原因',
|
||||
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
|
||||
|
||||
@ -1,9 +1,11 @@
|
||||
"""pipeline_llm.accounting — 异步出账循环（owner 模型三方账）。
|
||||
|
||||
扫 llm_usage 中 accounting_status='created' 且 status='ok' 的用量流水，
|
||||
扫 llm_usage 中 accounting_status='created' 且 status='SUCCEEDED' 的用量流水，
|
||||
逐条映射模型→产品，调产品层 product_accounting_generic 记账
|
||||
（产品层内部按 is_self_use 分流：自用只记 PAY* 采购成本；跨机构另记 PAY 客户应付，
|
||||
折扣精确到产品），成功置 'accounted'，失败置 'failed'（三态：created/accounted/failed）。
|
||||
2026-09-06 用户定夺：status 统一枚举 SUCCEEDED/FAILED/PENDING/RUNNING；
|
||||
只有 SUCCEEDED 的行才写 accounting_status（其余 NULL），出账循环自然只拾取 SUCCEEDED。
|
||||
|
||||
与账号/存储的同步记账分工明确（2026-09 用户定夺）：
|
||||
- 账号/存储：purchase_realtime 同步记账
|
||||
@ -32,7 +34,7 @@ _LOCK_TTL = 90 # 锁 TTL（略大于单批预算耗时）
|
||||
async def _get_pending(sor):
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT id, org_id, user_id, model_id, req_tokens, resp_tokens, cost, ppid, usages "
|
||||
"FROM llm_usage WHERE accounting_status='created' AND status='ok' "
|
||||
"FROM llm_usage WHERE accounting_status='created' AND status='SUCCEEDED' "
|
||||
"ORDER BY created_at LIMIT %d" % _BATCH, {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return recs or []
|
||||
|
||||
@ -539,15 +539,16 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke
|
||||
图像张数等按量计费模型的计价因子，由异步出账读入定价引擎
|
||||
（与 req_tokens/resp_tokens 并列，定价引擎按 pricing_data 选用）。
|
||||
|
||||
accounting_status 三态（2026-09-05 用户定夺）：
|
||||
created —— 已创建、待记账（status='ok' 的行由异步出账循环拾取）
|
||||
accounting_status 三态（2026-09-05 用户定夺；2026-09-06 补 NULL 语义）：
|
||||
created —— 已创建、待记账（仅 status='SUCCEEDED' 的行，出账循环拾取）
|
||||
accounted —— 记账成功
|
||||
failed —— 记账失败
|
||||
NULL —— 无记账意义：status≠SUCCEEDED 的行（FAILED/PENDING/RUNNING/recharge）
|
||||
一律保持 NULL，不写 created（用户定夺：只有 SUCCEEDED 才有记账的意义）
|
||||
status 统一枚举（2026-09-06 用户定夺）：SUCCEEDED/FAILED/PENDING/RUNNING（其他=recharge）。
|
||||
自用 vs 跨机构的分流不在这里——product_accounting_generic 内部按
|
||||
is_self_use 决定记账项（自用只记 PAY* 采购成本，跨机构另记 PAY 客户应付），
|
||||
所有成功调用统一 created 进出账队列，成本账不遗漏。
|
||||
status='failed'（调用失败）/'recharge'（充值）的行也置 created，但出账循环
|
||||
只拾取 status='ok'，故它们不会被记账（无可计费用量）。
|
||||
"""
|
||||
if not ctx or not isinstance(ctx, dict):
|
||||
return
|
||||
@ -578,7 +579,10 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke
|
||||
usages_str = json.dumps(usages, ensure_ascii=False, default=str)[:2000]
|
||||
except Exception:
|
||||
usages_str = ''
|
||||
await sor.C('llm_usage', {
|
||||
# status 统一枚举（2026-09-06 用户定夺）：SUCCEEDED/FAILED；
|
||||
# accounting_status 只在 SUCCEEDED 时写 created，否则缺省 NULL
|
||||
# （只有 SUCCEEDED 才有记账意义——失败/充值行挂 created 曾被误报"待记账"）
|
||||
usage_row = {
|
||||
'id': getID(),
|
||||
'org_id': caller_org,
|
||||
'user_id': ctx.get('user_id', ''),
|
||||
@ -591,11 +595,13 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke
|
||||
'charge': 0, # 客户应付由异步出账计算（自用为 0，跨机构按折扣）
|
||||
'ppid': model.get('ppid', '') or '',
|
||||
'task_ref': ctx.get('task_ref', ''),
|
||||
'status': 'ok' if ok_call else 'failed',
|
||||
'accounting_status': 'created',
|
||||
'status': 'SUCCEEDED' if ok_call else 'FAILED',
|
||||
'usages': usages_str,
|
||||
'note': (note or '')[:200],
|
||||
})
|
||||
}
|
||||
if ok_call:
|
||||
usage_row['accounting_status'] = 'created'
|
||||
await sor.C('llm_usage', usage_row)
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
except Exception as e:
|
||||
logger.warning("pipeline_llm: 结算失败（不阻断调用方）: %s", e)
|
||||
|
||||
@ -182,18 +182,19 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability,
|
||||
raise GovernError(
|
||||
'模型「%s」能力为 %s，与请求能力 %s 不匹配。请按能力分类选择模型' % (
|
||||
model_row.get('name', ''), cap, capability))
|
||||
# 适配模板（组包：path/headers/request_template，全 Jinja2，借鉴 uapi）
|
||||
# 适配模板（组包：path/method/headers/request_template，全 Jinja2，借鉴 uapi）
|
||||
profile = {}
|
||||
pid = model_row.get('profile_id') or ''
|
||||
if pid:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT path, headers, request_template, response_template FROM llm_api_profile "
|
||||
"WHERE id=${i}$ AND status='active' LIMIT 1", {"i": pid})
|
||||
"SELECT path, method, headers, request_template, response_template "
|
||||
"FROM llm_api_profile WHERE id=${i}$ AND status='active' LIMIT 1", {"i": pid})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if recs:
|
||||
r = recs[0]
|
||||
profile = {
|
||||
'path': getattr(r, 'path', '') or '',
|
||||
'method': (getattr(r, 'method', '') or 'POST').upper(),
|
||||
'headers': getattr(r, 'headers', '') or '',
|
||||
'request_template': getattr(r, 'request_template', '') or '',
|
||||
'response_template': getattr(r, 'response_template', '') or '',
|
||||
@ -207,20 +208,22 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability,
|
||||
qps = []
|
||||
for qid in qpids or []:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT path, headers, request_template, response_template FROM llm_api_profile "
|
||||
"WHERE id=${i}$ AND status='active' LIMIT 1", {"i": qid})
|
||||
"SELECT path, method, headers, request_template, response_template "
|
||||
"FROM llm_api_profile WHERE id=${i}$ AND status='active' LIMIT 1", {"i": qid})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not recs:
|
||||
continue
|
||||
r = recs[0]
|
||||
# method 无独立列：从 request_template JSON 的 method 键取（缺省 GET）
|
||||
qmethod = 'GET'
|
||||
try:
|
||||
rt = json.loads(getattr(r, 'request_template', '') or '{}')
|
||||
if isinstance(rt, dict) and rt.get('method'):
|
||||
qmethod = str(rt.get('method')).upper()
|
||||
except Exception:
|
||||
pass
|
||||
# method 独立列（2026-09-06 用户定夺：不塞 request_template）；
|
||||
# 存量未迁移行兼容：列空时回退解析 request_template JSON 的 method 键
|
||||
qmethod = (getattr(r, 'method', '') or '').strip().upper() or 'GET'
|
||||
if qmethod == 'GET':
|
||||
try:
|
||||
rt = json.loads(getattr(r, 'request_template', '') or '{}')
|
||||
if isinstance(rt, dict) and str(rt.get('method', '')).upper() in ('GET', 'POST'):
|
||||
qmethod = str(rt['method']).upper()
|
||||
except Exception:
|
||||
pass
|
||||
qps.append({
|
||||
'path': getattr(r, 'path', '') or '',
|
||||
'headers': getattr(r, 'headers', '') or '',
|
||||
@ -395,6 +398,8 @@ async def _call_upstream_chat(ctx, payload):
|
||||
url = _join_url(ctx.get('api_base'), path)
|
||||
headers = await _render_headers(profile.get('headers', ''), ns)
|
||||
body = await _build_upstream_body(ctx, payload)
|
||||
# method 独立列（2026-09-06 用户定夺）：默认 POST，profile 声明优先
|
||||
method = (profile.get('method') or 'POST').upper()
|
||||
timeout = int(_fnum(ctx.get('timeout')) or _TOTAL_TIMEOUT)
|
||||
# 可诊断性：只记长度不记值（真 key 不落日志）。print 确保进应用日志
|
||||
# （模块 logging.getLogger 的 handler 未接应用日志管道，2026-09-04 实测）
|
||||
@ -406,8 +411,8 @@ async def _call_upstream_chat(ctx, payload):
|
||||
for attempt in range(_MAX_ATTEMPTS):
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(
|
||||
url, headers=headers, json=body,
|
||||
async with session.request(
|
||||
method, url, headers=headers, json=body,
|
||||
timeout=aiohttp.ClientTimeout(total=timeout, connect=_CONNECT_TIMEOUT),
|
||||
) as resp:
|
||||
if resp.status != 200:
|
||||
@ -587,8 +592,9 @@ async def _async_inference(ctx, payload, req_timeout):
|
||||
else:
|
||||
url = _join_url(ctx.get('api_base'), path)
|
||||
headers = await _render_headers(profile.get('headers', ''), ns)
|
||||
submit_method = (profile.get('method') or 'POST').upper()
|
||||
try:
|
||||
status, data = await _http_request('POST', url, headers, body, step_timeout)
|
||||
status, data = await _http_request(submit_method, url, headers, body, step_timeout)
|
||||
except ValueError as e:
|
||||
await _fail('异步任务提交失败：%s' % e)
|
||||
return {}
|
||||
|
||||
@ -478,7 +478,7 @@ async def llm_dashboard(params_kw):
|
||||
models = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_model WHERE status='active'", {})
|
||||
orgs = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_org_policy WHERE status='active'", {})
|
||||
usage = await sor.sqlExe(
|
||||
"SELECT COUNT(*) AS c, COALESCE(SUM(charge),0) AS ch FROM llm_usage WHERE status='ok'", {})
|
||||
"SELECT COUNT(*) AS c, COALESCE(SUM(charge),0) AS ch FROM llm_usage WHERE status='SUCCEEDED'", {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
result['vendors'] = int(getattr(vendors[0], 'c', 0)) if vendors else 0
|
||||
result['accounts'] = int(getattr(accounts[0], 'c', 0)) if accounts else 0
|
||||
|
||||
@ -10,7 +10,7 @@ try:
|
||||
models = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_model WHERE status='active'", {})
|
||||
orgs = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_org_policy WHERE status='active'", {})
|
||||
usage = await sor.sqlExe(
|
||||
"SELECT COUNT(*) AS c, COALESCE(SUM(charge),0) AS ch FROM llm_usage WHERE status='ok'", {})
|
||||
"SELECT COUNT(*) AS c, COALESCE(SUM(charge),0) AS ch FROM llm_usage WHERE status='SUCCEEDED'", {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
data['vendors'] = int(getattr(vendors[0], 'c', 0)) if vendors else 0
|
||||
data['accounts'] = int(getattr(accounts[0], 'c', 0)) if accounts else 0
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user