diff --git a/README.md b/README.md index 6616bc2..c80ee41 100644 --- a/README.md +++ b/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）。 diff --git a/init/data.json b/init/data.json index b6dfe40..f5c84ab 100644 --- a/init/data.json +++ b/init/data.json @@ -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": "充值"} ]} ] diff --git a/models/llm_api_profile.json b/models/llm_api_profile.json index 07da2a0..5f03ad9 100644 --- a/models/llm_api_profile.json +++ b/models/llm_api_profile.json @@ -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)", diff --git a/models/llm_usage.json b/models/llm_usage.json index bb9f89a..5e3863a 100644 --- a/models/llm_usage.json +++ b/models/llm_usage.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", diff --git a/mysql.ddl.sql b/mysql.ddl.sql index 2c279bc..3bcca78 100644 --- a/mysql.ddl.sql +++ b/mysql.ddl.sql @@ -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 '创建时间', diff --git a/pipeline_llm/accounting.py b/pipeline_llm/accounting.py index dc46870..64ccf28 100644 --- a/pipeline_llm/accounting.py +++ b/pipeline_llm/accounting.py @@ -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 [] diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index 12c2afb..b7091dc 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -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) diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index a7f2b7b..4de7316 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -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 {} diff --git a/pipeline_llm/init.py b/pipeline_llm/init.py index 2e98d08..a1e01fe 100644 --- a/pipeline_llm/init.py +++ b/pipeline_llm/init.py @@ -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 diff --git a/wwwroot/api/llm_dashboard_widget.dspy b/wwwroot/api/llm_dashboard_widget.dspy index 4a674de..2915f8b 100644 --- a/wwwroot/api/llm_dashboard_widget.dspy +++ b/wwwroot/api/llm_dashboard_widget.dspy @@ -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