feat(async): 异步模型执行器(提交→轮询query_profile_ids→取结果)+usages用量因子记账;供应商去protocol字段(模式归模型适配模板);修_model_chain漏取profile_id等字段致适配模板从未加载的存量bug
This commit is contained in:
parent
beef2d81a2
commit
c819faef2d
@ -9,12 +9,6 @@
|
||||
"id"
|
||||
],
|
||||
"alters": {
|
||||
"protocol": {
|
||||
"uitype": "code",
|
||||
"dataurl": "{{entire_url('../api/get_llm_protocol_options.dspy')}}",
|
||||
"valueField": "value",
|
||||
"textField": "text"
|
||||
},
|
||||
"status": {
|
||||
"uitype": "code",
|
||||
"dataurl": "{{entire_url('../api/get_llm_status_options.dspy')}}",
|
||||
|
||||
@ -103,6 +103,12 @@
|
||||
"nullable": "no",
|
||||
"default": "'ok'"
|
||||
},
|
||||
{
|
||||
"name": "usages",
|
||||
"title": "用量详情(JSON)",
|
||||
"type": "text",
|
||||
"nullable": "yes"
|
||||
},
|
||||
{
|
||||
"name": "note",
|
||||
"title": "备注/原因",
|
||||
@ -154,4 +160,4 @@
|
||||
"cond": "parentid='llm_usage_status'"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@ -24,14 +24,6 @@
|
||||
"length": 100,
|
||||
"nullable": "no"
|
||||
},
|
||||
{
|
||||
"name": "protocol",
|
||||
"title": "协议类型",
|
||||
"type": "str",
|
||||
"length": 20,
|
||||
"nullable": "no",
|
||||
"default": "'openai_compat'"
|
||||
},
|
||||
{
|
||||
"name": "endpoints",
|
||||
"title": "端点目录(JSON)",
|
||||
@ -80,13 +72,6 @@
|
||||
}
|
||||
],
|
||||
"codes": [
|
||||
{
|
||||
"field": "protocol",
|
||||
"table": "appcodes_kv",
|
||||
"valuefield": "k",
|
||||
"textfield": "v",
|
||||
"cond": "parentid='llm_protocol'"
|
||||
},
|
||||
{
|
||||
"field": "status",
|
||||
"table": "appcodes_kv",
|
||||
@ -95,4 +80,4 @@
|
||||
"cond": "parentid='llm_status'"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@ -5,7 +5,6 @@
|
||||
CREATE TABLE IF NOT EXISTS llm_vendor (
|
||||
`id` varchar(32) NOT NULL comment '主键ID',
|
||||
`name` varchar(100) NOT NULL comment '供应商名称',
|
||||
`protocol` varchar(20) NOT NULL DEFAULT 'openai_compat' comment '协议类型',
|
||||
`endpoints` text comment '端点目录(JSON): [{region,base_url,proxy,timeout,note}]',
|
||||
`description` text comment '描述',
|
||||
`status` varchar(20) NOT NULL DEFAULT 'active' comment '状态',
|
||||
@ -126,6 +125,7 @@ CREATE TABLE IF NOT EXISTS llm_usage (
|
||||
`ppid` varchar(32) comment '定价项目ID',
|
||||
`task_ref` varchar(100) comment '调用来源',
|
||||
`status` varchar(20) NOT NULL DEFAULT 'ok' comment '状态(ok/failed/recharge)',
|
||||
`usages` text comment '非token用量因子JSON(时长/分辨率等,异步出账读入定价引擎)',
|
||||
`note` varchar(200) comment '备注/原因',
|
||||
`created_at` datetime NOT NULL DEFAULT current_timestamp() comment '创建时间',
|
||||
PRIMARY KEY (`id`),
|
||||
|
||||
@ -12,6 +12,7 @@
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
|
||||
@ -29,7 +30,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 "
|
||||
"SELECT id, org_id, user_id, model_id, req_tokens, resp_tokens, cost, ppid, usages "
|
||||
"FROM llm_usage WHERE accounting_status='pending' AND status='ok' "
|
||||
"ORDER BY created_at LIMIT %d" % _BATCH, {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
@ -63,6 +64,16 @@ async def _settle_one(sor, row):
|
||||
'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_raw = getattr(row, 'usages', '') or ''
|
||||
if usages_raw:
|
||||
try:
|
||||
extra = json.loads(usages_raw)
|
||||
if isinstance(extra, dict):
|
||||
usage_data.update(extra)
|
||||
except Exception:
|
||||
pass
|
||||
fn = getattr(env, 'product_accounting_generic', None)
|
||||
if fn is None:
|
||||
# 产品模块未加载 → 保留 pending 下轮再试
|
||||
|
||||
@ -199,7 +199,7 @@ async def _model_chain(sor, org_id: str, model_name: str, purpose: str = ''):
|
||||
for mid in chain_ids:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT id, name, vendor_id, vendor_model_id, capability, default_params, "
|
||||
"ppid, org_id "
|
||||
"ppid, org_id, profile_id, sync_mode, query_profile_ids "
|
||||
"FROM llm_model WHERE id=${i}$ AND status='active' LIMIT 1", {"i": mid})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if recs:
|
||||
@ -208,15 +208,18 @@ async def _model_chain(sor, org_id: str, model_name: str, purpose: str = ''):
|
||||
|
||||
|
||||
async def _vendor_endpoints(sor, vendor_id: str):
|
||||
"""供应商端点目录。返回 (protocol, [endpoint dict])。"""
|
||||
"""供应商端点目录。返回 [endpoint dict]。
|
||||
|
||||
2026-09-05：供应商表不再有 protocol 字段——请求形态（协议/模式）由模型挂的
|
||||
适配模板（llm_api_profile）决定，供应商只提供端点目录（账号/区域/超时）。
|
||||
"""
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT protocol, endpoints FROM llm_vendor WHERE id=${i}$ AND status='active' LIMIT 1",
|
||||
"SELECT endpoints FROM llm_vendor WHERE id=${i}$ AND status='active' LIMIT 1",
|
||||
{"i": vendor_id})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not recs:
|
||||
return '', []
|
||||
r = recs[0]
|
||||
return getattr(r, 'protocol', '') or '', _parse_json(getattr(r, 'endpoints', ''), [])
|
||||
return []
|
||||
return _parse_json(getattr(recs[0], 'endpoints', ''), [])
|
||||
|
||||
|
||||
async def _account_candidates(sor, model: dict):
|
||||
@ -255,7 +258,7 @@ async def _pick_candidate(sor, model: dict, pref: str):
|
||||
vendor_id = model.get('vendor_id', '')
|
||||
if not vendor_id:
|
||||
return False, '模型「%s」未绑定供应商，无法选择账号' % model.get('name', '')
|
||||
protocol, endpoints = await _vendor_endpoints(sor, vendor_id)
|
||||
endpoints = await _vendor_endpoints(sor, vendor_id)
|
||||
if not endpoints:
|
||||
return False, '供应商端点目录为空（模型治理→供应商→端点目录），请配置端点'
|
||||
accounts = await _account_candidates(sor, model)
|
||||
@ -478,11 +481,14 @@ async def govern_resolve(org_id: str, user_id: str = '', model_name: str = '',
|
||||
|
||||
|
||||
async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_tokens: int = 0,
|
||||
note: str = ''):
|
||||
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 选用）。
|
||||
|
||||
记账分流（2026-09）：
|
||||
本机构模型（含 owner 自用）：只记成本侧——供应商账号扣减，
|
||||
@ -521,6 +527,12 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke
|
||||
if ok_call and is_owner_model:
|
||||
accounting_status = 'pending'
|
||||
from appPublic.uniqueID import getID
|
||||
usages_str = ''
|
||||
if usages:
|
||||
try:
|
||||
usages_str = json.dumps(usages, ensure_ascii=False, default=str)[:2000]
|
||||
except Exception:
|
||||
usages_str = ''
|
||||
await sor.C('llm_usage', {
|
||||
'id': getID(),
|
||||
'org_id': caller_org,
|
||||
@ -536,6 +548,7 @@ async def govern_settle(ctx: dict, ok_call: bool, req_tokens: int = 0, resp_toke
|
||||
'task_ref': ctx.get('task_ref', ''),
|
||||
'status': 'ok' if ok_call else 'failed',
|
||||
'accounting_status': accounting_status,
|
||||
'usages': usages_str,
|
||||
'note': (note or '')[:200],
|
||||
})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
|
||||
@ -16,6 +16,7 @@ path/headers 决定上游请求形态（缺省按 OpenAI 兼容）；非 t2t 形
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
|
||||
logger = logging.getLogger("pipeline_llm.inference")
|
||||
|
||||
@ -135,6 +136,33 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability,
|
||||
ctx['default_params'] = _parse_json(model_row.get('default_params'), {})
|
||||
ctx['profile'] = profile
|
||||
ctx['capability'] = cap
|
||||
# 异步模型的后续步骤模板（提交→查询→[下载]）：按 query_profile_ids 顺序加载
|
||||
qpids = _parse_json(model_row.get('query_profile_ids'), [])
|
||||
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})
|
||||
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
|
||||
qps.append({
|
||||
'path': getattr(r, 'path', '') or '',
|
||||
'headers': getattr(r, 'headers', '') or '',
|
||||
'method': qmethod,
|
||||
'request_template': getattr(r, 'request_template', '') or '',
|
||||
'response_template': getattr(r, 'response_template', '') or '',
|
||||
})
|
||||
ctx['query_profiles'] = qps
|
||||
return ctx
|
||||
|
||||
|
||||
@ -331,14 +359,227 @@ async def _call_upstream_chat(ctx, payload):
|
||||
raise last if last else ValueError('上游调用失败')
|
||||
|
||||
|
||||
# ────────────────────────── 异步执行器（提交→轮询→取结果） ──────────────────────────
|
||||
# 异步模型（sync_mode='async'，如 DashScope 视频/图像生成）：
|
||||
# 1. 提交：POST 提交模板（模型挂的 profile）→ 返回 output.task_id
|
||||
# 2. 轮询：GET 查询步骤模板（query_profiles[0]）直到终态
|
||||
# 3. 结算：用量因子（时长/分辨率等）落 llm_usage.usages，异步出账读入定价引擎
|
||||
#
|
||||
# 模板驱动：提交/查询的 path/headers/body 全来自 llm_api_profile（Jinja2 渲染），
|
||||
# 新增异步形态只加 profile 记录，不改代码。DashScope 约定（官方文档核实，
|
||||
# 2026-09-05）：提交响应 output.task_id；查询响应 output.task_status ∈
|
||||
# PENDING/RUNNING/SUCCEEDED/FAILED/CANCELED/UNKNOWN，成功时带结果产物。
|
||||
_ASYNC_POLL_START = 5.0 # 轮询起始间隔（秒）
|
||||
_ASYNC_POLL_MAX = 15.0 # 轮询最大间隔（秒）
|
||||
_ASYNC_TOTAL_TIMEOUT = 600 # 异步任务总等待预算（秒，可被 _timeout 覆盖，上限 900）
|
||||
_ASYNC_TASK_TERMINAL = ('SUCCEEDED', 'FAILED', 'CANCELED', 'UNKNOWN')
|
||||
|
||||
|
||||
def _last_prompt_text(messages):
|
||||
"""从 messages 提取提示词文本（取末条非空 content；多模态列表拼文本部分）。"""
|
||||
for m in reversed(messages or []):
|
||||
if not isinstance(m, dict):
|
||||
continue
|
||||
c = m.get('content', '')
|
||||
if isinstance(c, str) and c.strip():
|
||||
return c
|
||||
if isinstance(c, list):
|
||||
parts = []
|
||||
for seg in c:
|
||||
if isinstance(seg, dict):
|
||||
t = seg.get('text') or ''
|
||||
if t:
|
||||
parts.append(str(t))
|
||||
if parts:
|
||||
return '\n'.join(parts)
|
||||
return ''
|
||||
|
||||
|
||||
async def _build_async_body(ctx, payload):
|
||||
"""异步提交请求体渲染（同 _build_upstream_body，另注入 prompt 便利变量）。"""
|
||||
profile = ctx.get('profile') or {}
|
||||
tmpl = (profile.get('request_template') or '').strip()
|
||||
if not tmpl:
|
||||
raise GovernError('异步模型的提交适配模板缺 request_template——请完善模型挂的适配模板')
|
||||
upstream = dict(payload or {})
|
||||
upstream.pop('stream', None)
|
||||
upstream['model'] = ctx.get('model_id') or upstream.get('model') or ''
|
||||
for k, v in (ctx.get('default_params') or {}).items():
|
||||
upstream.setdefault(k, v)
|
||||
messages = upstream.pop('messages', [])
|
||||
model = upstream.pop('model', '')
|
||||
ns = {
|
||||
'model': model, 'messages': messages,
|
||||
'prompt': _last_prompt_text(messages),
|
||||
'params': upstream, # 业务参数（resolution/duration/image_url 等）
|
||||
'api_key': ctx.get('api_key') or '',
|
||||
'org_id': ctx.get('org_id') or '',
|
||||
}
|
||||
s = await _render_tmpl(tmpl, ns)
|
||||
try:
|
||||
body = json.loads(s)
|
||||
except Exception as e:
|
||||
raise GovernError('提交请求体模板渲染结果不是合法 JSON：%s（模板：%s）' % (
|
||||
str(e)[:120], tmpl[:120]))
|
||||
if not isinstance(body, dict):
|
||||
raise GovernError('提交请求体模板必须渲染成 JSON 对象')
|
||||
return body
|
||||
|
||||
|
||||
async def _http_request(method, url, headers, body, timeout):
|
||||
"""通用上游 HTTP 请求（异步执行器用：提交/查询）。
|
||||
|
||||
返回 (status, data)；data 尽量 json.loads，失败返回原文。
|
||||
超时/连接错误抛 ValueError（消息真实可行动，含异常类型）。
|
||||
"""
|
||||
import aiohttp
|
||||
import asyncio
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.request(
|
||||
method.upper(), url, headers=headers,
|
||||
json=body if body is not None else None,
|
||||
timeout=aiohttp.ClientTimeout(total=timeout, connect=_CONNECT_TIMEOUT),
|
||||
) as resp:
|
||||
text = await resp.text()
|
||||
try:
|
||||
data = json.loads(text) if text else {}
|
||||
except Exception:
|
||||
data = text
|
||||
return resp.status, data
|
||||
except (asyncio.TimeoutError, aiohttp.ClientError) as e:
|
||||
raise ValueError('上游请求失败（%s）：%s' % (type(e).__name__, repr(e)))
|
||||
|
||||
|
||||
async def _async_query_once(ctx, qprofile, task_id, ns, timeout):
|
||||
"""按查询步骤模板请求一次任务状态。path 支持 {task_id} 占位符。"""
|
||||
path = await _render_tmpl((qprofile.get('path') or '').strip(), ns)
|
||||
path = path.replace('{task_id}', task_id)
|
||||
if path.startswith('http://') or path.startswith('https://'):
|
||||
url = path
|
||||
else:
|
||||
url = (ctx.get('api_base') or '').rstrip('/') + 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)
|
||||
if status != 200:
|
||||
raise ValueError('任务查询失败 HTTP %d: %s' % (
|
||||
status, str(data)[:200]))
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError('任务查询返回非 JSON：%s' % str(data)[:200])
|
||||
return data
|
||||
|
||||
|
||||
async def _async_inference(ctx, payload, req_timeout):
|
||||
"""异步模型执行：提交任务 → 轮询至终态 → 组装结果 → 结算（usages 记账）。
|
||||
|
||||
返回 OpenAI 兼容形态（choices[0].message.content = 产物地址/文本），
|
||||
另带 output（上游原始输出）与 task_id。
|
||||
任何失败：先 govern_settle 记 failed 流水，再抛 GovernError（消息真实可行动）。
|
||||
"""
|
||||
import asyncio
|
||||
from .gateway import govern_settle
|
||||
|
||||
async def _fail(msg):
|
||||
await govern_settle(ctx, False, 0, 0, msg)
|
||||
raise GovernError(msg)
|
||||
|
||||
qprofiles = ctx.get('query_profiles') or []
|
||||
if not qprofiles:
|
||||
await _fail('模型「%s」为异步模型但未登记查询步骤模板（query_profile_ids），'
|
||||
'无法获取任务结果——请在模型治理补录查询步骤适配模板' % ctx.get('name', ''))
|
||||
return {}
|
||||
profile = ctx.get('profile') or {}
|
||||
ns = {'api_key': ctx.get('api_key') or '', 'org_id': ctx.get('org_id') or ''}
|
||||
step_timeout = int(_fnum(ctx.get('timeout')) or 60)
|
||||
|
||||
# ── 1. 提交任务 ──
|
||||
body = await _build_async_body(ctx, payload)
|
||||
path = await _render_tmpl((profile.get('path') or '').strip(), ns)
|
||||
if path.startswith('http://') or path.startswith('https://'):
|
||||
url = path
|
||||
else:
|
||||
url = (ctx.get('api_base') or '').rstrip('/') + path
|
||||
headers = await _render_headers(profile.get('headers', ''), ns)
|
||||
try:
|
||||
status, data = await _http_request('POST', url, headers, body, step_timeout)
|
||||
except ValueError as e:
|
||||
await _fail('异步任务提交失败：%s' % e)
|
||||
return {}
|
||||
if status != 200 or not isinstance(data, dict):
|
||||
await _fail('异步任务提交失败 HTTP %d: %s' % (status, str(data)[:300]))
|
||||
return {}
|
||||
task_id = ((data.get('output') or {}).get('task_id') or data.get('task_id') or '').strip()
|
||||
if not task_id:
|
||||
await _fail('异步任务提交响应缺 task_id：%s' % (
|
||||
json.dumps(data, ensure_ascii=False, default=str)[:300]))
|
||||
return {}
|
||||
print("[inference] 异步任务已提交 task_id=%s model=%s" % (task_id, ctx.get('model_id', '')))
|
||||
|
||||
# ── 2. 轮询至终态 ──
|
||||
total_budget = req_timeout if req_timeout > 0 else _ASYNC_TOTAL_TIMEOUT
|
||||
deadline = time.time() + total_budget
|
||||
interval = _ASYNC_POLL_START
|
||||
last = data
|
||||
finished = ''
|
||||
while time.time() < deadline:
|
||||
await asyncio.sleep(interval)
|
||||
interval = min(interval * 1.5, _ASYNC_POLL_MAX)
|
||||
try:
|
||||
last = await _async_query_once(ctx, qprofiles[0], task_id, ns, step_timeout)
|
||||
except ValueError as e:
|
||||
# 单次查询失败不立即终止（可能瞬时抖动），预算内继续
|
||||
print("[inference] 异步查询瞬时失败（预算内继续）: %s" % e)
|
||||
continue
|
||||
finished = str((last.get('output') or {}).get('task_status') or '').upper()
|
||||
if finished in _ASYNC_TASK_TERMINAL:
|
||||
break
|
||||
if finished not in _ASYNC_TASK_TERMINAL:
|
||||
await _fail('异步任务等待超时（%d 秒）：最后状态 %s，task_id=%s——'
|
||||
'可用 _timeout 延长等待（上限 900 秒）' % (
|
||||
total_budget, finished or 'UNKNOWN', task_id))
|
||||
if finished != 'SUCCEEDED':
|
||||
out = last.get('output') or {}
|
||||
await _fail('异步任务执行失败：%s %s（task_id=%s）' % (
|
||||
out.get('code') or finished, str(out.get('message') or '')[:200], task_id))
|
||||
|
||||
# ── 3. 组装结果 + 用量因子记账 ──
|
||||
out = last.get('output') or {}
|
||||
content = (out.get('video_url') or out.get('image_url') or out.get('result_url')
|
||||
or out.get('text') or '')
|
||||
usage = last.get('usage') or {}
|
||||
usages = dict(usage) if isinstance(usage, dict) else {}
|
||||
usages['model'] = ctx.get('model_id') or ''
|
||||
# 计价过滤维度用请求值（与文档计价档位原文一致，如 resolution=720P）；
|
||||
# 时长以上游计费返回为准，缺则回请求值
|
||||
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)
|
||||
return {
|
||||
"choices": [{"message": {"content": content, "role": "assistant"},
|
||||
"finish_reason": "stop"}],
|
||||
"usage": usage,
|
||||
"output": out,
|
||||
"task_id": task_id,
|
||||
}
|
||||
|
||||
|
||||
# ────────────────────────── 对外主入口 ──────────────────────────
|
||||
|
||||
async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''):
|
||||
"""t2t 推理：门禁链 → 上游调用 → 结算。返回上游响应 dict（OpenAI 兼容）。
|
||||
"""统一推理：门禁链 → 上游调用 → 结算。返回上游响应 dict（OpenAI 兼容形态）。
|
||||
|
||||
payload: 客户端请求体（messages 必需；tools/temperature/max_tokens 透传）。
|
||||
payload: 客户端请求体（messages 必需；tools/temperature/max_tokens 透传；
|
||||
异步模型另收业务参数 resolution/duration/image_url 等）。
|
||||
model_name 空 → 机构策略缺省模型（主→备链第一个）。
|
||||
失败抛 GovernError（消息真实可行动）。
|
||||
|
||||
同步/异步分流（2026-09-05）：按模型 sync_mode——sync 走 chat/completions
|
||||
一次往返；async 走异步执行器（提交任务→按 query_profile_ids 轮询→取结果），
|
||||
用量因子（视频时长/分辨率等）落 llm_usage.usages 供定价引擎计费。
|
||||
本入口不再限定能力类型——统一入口按模型实际形态执行，能力分栏在
|
||||
list_models/选择层体现。
|
||||
"""
|
||||
from .gateway import govern_settle, _get_db
|
||||
|
||||
@ -357,9 +598,20 @@ 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, 't2t', est, task_ref, purpose)
|
||||
sor, org_id, user_id or '', model_name, '', est, task_ref, purpose)
|
||||
if req_timeout > 0:
|
||||
ctx['timeout'] = min(req_timeout, 900)
|
||||
# 异步模型：提交→轮询→取结果（用量因子走 usages 记账）
|
||||
model_row = ctx.get('model_row') or {}
|
||||
if (model_row.get('sync_mode') or 'sync') == 'async':
|
||||
return await _async_inference(ctx, payload, req_timeout)
|
||||
# 同步模型守能力边界：聊天入口只认 t2t（embedding/rerank 等另有入口）
|
||||
cap = (model_row.get('capability') or 't2t').strip().lower()
|
||||
if cap != 't2t':
|
||||
from .gateway import govern_settle
|
||||
await govern_settle(ctx, False, 0, 0, 'capability mismatch: %s != t2t' % cap)
|
||||
raise GovernError('模型「%s」能力为 %s，聊天入口仅支持 t2t——请按能力分类选择模型' % (
|
||||
model_row.get('name', ''), cap))
|
||||
try:
|
||||
data = await _call_upstream_chat(ctx, payload)
|
||||
except Exception as e:
|
||||
|
||||
@ -83,7 +83,7 @@ async def _get_or_create_vendor(sor, provider, base_url, dry):
|
||||
eps = [{'base_url': base_url, 'region': 'domestic', 'timeout': 60}]
|
||||
if not dry:
|
||||
await sor.C('llm_vendor', {
|
||||
'id': vid, 'name': name, 'protocol': 'openai_compat',
|
||||
'id': vid, 'name': name,
|
||||
'endpoints': json.dumps(eps, ensure_ascii=False),
|
||||
'description': '由旧 llm 表迁移生成', 'status': 'active', 'org_id': '0',
|
||||
})
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user