987 lines
47 KiB
Python
987 lines
47 KiB
Python
"""pipeline_llm.inference — 统一推理入口（分类照 llmage：按能力类型 catelogid 路由）。
|
||
|
||
设计（2026-09-04）：产线平台所有模型调用收敛到本模块：
|
||
- 进程内：agent/引擎经 pipeline_service.llm_bridge 走 HTTP 自调用 →
|
||
/pipeline_llm/api/v1/chat/completions（Bearer 短期 token）→ chat_inference()
|
||
- 运行环境：短期 token 代理（真 key 永不出进程，沿用原铁律）同一入口
|
||
- 旧运行环境入口 /pipeline_core/api/llm_v1 委托同一实现（收敛唯一实现）
|
||
|
||
门禁链复用 gateway.govern_resolve/govern_settle（限流/限额/主备容错/端点轮转/
|
||
预授权/双维度记账）。治理真实失败抛 GovernError，消息真实可行动，禁止静默回退。
|
||
|
||
一期形态：t2t（OpenAI 兼容 chat/completions）。适配模板 llm_api_profile 的
|
||
path/headers 决定上游请求形态（缺省按 OpenAI 兼容）；非 t2t 形态（图/视频/语音）
|
||
进来时按 capability 增加入口（对齐 llmage v1 端点分栏），不改本层结构。
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
|
||
logger = logging.getLogger("pipeline_llm.inference")
|
||
|
||
# OpenAI 兼容缺省适配（模型未挂 profile / profile 缺字段时的兜底）——
|
||
# 组包方式借鉴 uapi 模块（llmage 即调 uapi 组包）：path/headers/body 全部
|
||
# Jinja2 模板，经 tmpl_engine 渲染后 json.loads；api_key 作模板变量注入。
|
||
_DEFAULT_CHAT_PATH = "/chat/completions"
|
||
# ⚠️ 模板里的 {{ api_key }} 占位符必须用拼接构造——完整模式写在源码字面量里
|
||
# 会被脱敏工具替换成 ***（2026-09-04 实测 401 根因：渲染出字面 *** 当 key）
|
||
_API_KEY_VAR = '{{' + ' api_key ' + '}}'
|
||
_BUILTIN_HEADERS_TMPL = '{"Authorization": "Bearer ' + _API_KEY_VAR + '", "Content-Type": "application/json"}'
|
||
# 缺省请求体模板：model+messages 必填，其余调用方参数经 extra 透传
|
||
_BUILTIN_REQUEST_TMPL = (
|
||
'{"model": {{ model|tojson }}, "messages": {{ messages|tojson }}'
|
||
'{% for k, v in extra.items() %}, "{{ k }}": {{ v|tojson }}{% endfor %}}'
|
||
)
|
||
# 请求体透传字段白名单（OpenAI 兼容常用参数，除 model/messages 外）
|
||
_PASSTHROUGH_FIELDS = ('temperature', 'top_p', 'max_tokens', 'tools',
|
||
'tool_choice', 'presence_penalty', 'frequency_penalty',
|
||
'stop', 'response_format', 'seed')
|
||
_RETRYABLE_STATUS = (429, 500, 502, 503, 504)
|
||
_MAX_ATTEMPTS = 3
|
||
_TOTAL_TIMEOUT = 300
|
||
_CONNECT_TIMEOUT = 30
|
||
|
||
EST_TOKENS_PER_CHAR = 2 # 预估用量：中文约 1 字符≈0.5 token，保守按 2 字符 1 token
|
||
|
||
from .gateway import GovernError # 复用治理层错误类型（消息真实可行动）
|
||
|
||
|
||
def _parse_json(s, default):
|
||
if not s:
|
||
return default
|
||
try:
|
||
v = json.loads(s)
|
||
return v if v is not None else default
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def _fnum(v):
|
||
try:
|
||
return float(v or 0)
|
||
except (TypeError, ValueError):
|
||
return 0.0
|
||
|
||
|
||
def _media_ns_extra():
|
||
"""模板命名空间扩展：媒体文件转换函数（对齐 uapi 约定）+ request 占位。
|
||
|
||
生成类模型的两条铁律（2026-09-05 用户定夺）：
|
||
上行（data 模板）：上传/本地媒体用 {{b64media2url(request, xxx_file)}}
|
||
转成本地公网 URL 再传上游（base64 落盘→/idfile 公网地址）；
|
||
下行（response 模板）：上游生成物 URL 用
|
||
{{downloadfile2url(request, output.video_url)}} 落地为本地持久
|
||
URL 再给调用者（上游 URL 有效期短，DashScope 视频仅 24 小时）。
|
||
request=None：内部执行器无 HTTP 请求上下文，函数走 BASEURL 环境变量
|
||
兜底拼公网地址（部署机须配置）。导入失败退回透传函数（不阻断调用）。
|
||
"""
|
||
ns = {'request': None, 'json': json}
|
||
try:
|
||
from ahserver.filestorage import downloadfile2url, b64media2url
|
||
ns['downloadfile2url'] = downloadfile2url
|
||
ns['b64media2url'] = b64media2url
|
||
except Exception:
|
||
async def _d2u(request, url, **kw):
|
||
return url
|
||
async def _b2u(request, media):
|
||
return media
|
||
ns['downloadfile2url'] = _d2u
|
||
ns['b64media2url'] = _b2u
|
||
return ns
|
||
|
||
|
||
def _normalize_media_aliases(upstream):
|
||
"""同类接口对外契约统一（2026-09-06 用户定夺，对齐 sage/llmage）：
|
||
上传媒体一律三数组参数 image_files / audio_files / video_files，
|
||
值为字符串或数组均可（模板 Jinja 动态判断两种形态）。
|
||
|
||
旧别名归一：xxx_file / xxx_url（单值）→ xxx_files（数组）；
|
||
DashScope 形态的 media 数组（[{"type": "reference_image", "url": ...}]
|
||
或纯 URL 列表）→ 按 type 分组进三数组（r2v 实测调用方直接传 media）。
|
||
"""
|
||
for media in ('image', 'video', 'audio'):
|
||
files_k = media + '_files'
|
||
if not upstream.get(files_k):
|
||
for alt in (media + '_file', media + '_url'):
|
||
v = upstream.get(alt)
|
||
if v:
|
||
upstream[files_k] = [v] if isinstance(v, str) else list(v)
|
||
break
|
||
m = upstream.get('media')
|
||
if m and not any(upstream.get(x + '_files') for x in ('image', 'video', 'audio')):
|
||
groups = {'image': [], 'video': [], 'audio': []}
|
||
items = m if isinstance(m, list) else [m]
|
||
for it in items:
|
||
if isinstance(it, str) and it.strip():
|
||
groups['image'].append(it)
|
||
elif isinstance(it, dict):
|
||
url = next((v for v in it.values()
|
||
if isinstance(v, str) and v.strip().lower().startswith(
|
||
('http://', 'https://', 'data:', 'asset://', '/'))), '')
|
||
if not url:
|
||
continue
|
||
t = str(it.get('type') or '').lower()
|
||
g = 'video' if 'video' in t else ('audio' if 'audio' in t else 'image')
|
||
groups[g].append(url)
|
||
for media, lst in groups.items():
|
||
if lst:
|
||
upstream[media + '_files'] = lst
|
||
upstream.pop('media', None)
|
||
|
||
|
||
def _texts_len(messages):
|
||
"""messages 内容总长（估预授权用量用）。"""
|
||
try:
|
||
return sum(len(str(m.get('content', ''))) for m in (messages or [])
|
||
if isinstance(m, dict))
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
# ────────────────────────── 解析链 ──────────────────────────
|
||
|
||
async def _resolve_call(sor, org_id, user_id, model_name, capability,
|
||
est_tokens, task_ref, purpose='', project_id='',
|
||
session_id=''):
|
||
"""门禁链 ①-⑥ + 加载模型/适配模板。
|
||
|
||
purpose='utility'：辅助任务（分类/选择/摘要），模型链为 辅助→主→备
|
||
（需机构策略配置 utility_model_id；未配则等同普通链）。
|
||
|
||
session_id：会话粘性（2026-09-11 用户定夺：同会话固定账号省上游缓存钱，
|
||
跨会话轮询）；空=纯轮询。
|
||
|
||
返回 dict（govern ctx + payload 所需全部信息）。任何一级不过抛 GovernError。
|
||
⚠️ 本层无 __LEGACY__ 兜底——新接口只认新表；机构未配策略直接报可行动错误。
|
||
"""
|
||
from .gateway import govern_resolve
|
||
|
||
if not org_id:
|
||
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 '',
|
||
project_id=project_id or '', session_id=session_id or '')
|
||
if not ok:
|
||
if res == '__LEGACY__':
|
||
# 全平台无策略/指定模型不在新表：不再有旧表兜底，报可行动错误
|
||
if model_name:
|
||
raise GovernError(
|
||
'模型「%s」未注册到模型治理（模型注册表无此名或已停用）。'
|
||
'请在模型治理→模型注册中添加，或改用已注册模型' % model_name)
|
||
raise GovernError(
|
||
'无可用模型策略：机构 %s 未配置且平台也未配置组织容错策略。'
|
||
'请在模型治理→组织容错策略配置主/备模型' % org_id)
|
||
raise GovernError(res)
|
||
if not isinstance(res, dict):
|
||
raise GovernError('治理引擎返回异常结果: %s' % str(res)[:200])
|
||
ctx = res
|
||
# 能力类型校验（防 t2t 入口调到非 t2t 模型）
|
||
model_row = ctx.get('model_row') or {}
|
||
cap = (model_row.get('capability') or 't2t').strip().lower()
|
||
if capability and cap != capability:
|
||
# 释放预授权再报错（调用不会发生）
|
||
from .gateway import govern_settle
|
||
await govern_settle(ctx, False, 'capability mismatch: %s!=%s' % (cap, capability))
|
||
raise GovernError(
|
||
'模型「%s」能力为 %s，与请求能力 %s 不匹配。请按能力分类选择模型' % (
|
||
model_row.get('name', ''), cap, capability))
|
||
# 适配模板（组包：path/method/headers/request_template，全 Jinja2，借鉴 uapi）
|
||
profile = {}
|
||
pid = model_row.get('profile_id') or ''
|
||
if pid:
|
||
recs = await sor.sqlExe(
|
||
"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 '',
|
||
}
|
||
# default_params 合并（调用方显式参数优先）
|
||
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, 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 独立列（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 '',
|
||
'method': qmethod,
|
||
'request_template': getattr(r, 'request_template', '') or '',
|
||
'response_template': getattr(r, 'response_template', '') or '',
|
||
})
|
||
ctx['query_profiles'] = qps
|
||
return ctx
|
||
|
||
|
||
async def _pick_default_model_name(org_id, capability='t2t'):
|
||
"""机构缺省模型：组织策略主模型 → 备链第一个；
|
||
本机构无策略 → 平台 owner('0') 策略（与 govern_resolve 兜底一致）。都没有返回 ''。"""
|
||
from .gateway import _get_db
|
||
db, dbname = _get_db()
|
||
async with db.sqlorContext(dbname) as sor:
|
||
policy_orgs = [org_id] if (org_id and org_id != '0') else []
|
||
policy_orgs.append('0')
|
||
recs = []
|
||
for po in policy_orgs:
|
||
recs = await sor.sqlExe(
|
||
"SELECT primary_model_id, backup_model_ids FROM llm_org_policy "
|
||
"WHERE org_id=${o}$ AND status='active' LIMIT 1", {"o": po})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if recs:
|
||
break
|
||
if not recs:
|
||
return ''
|
||
mids = []
|
||
pm = getattr(recs[0], 'primary_model_id', '') or ''
|
||
if pm:
|
||
mids.append(pm)
|
||
for b in _parse_json(getattr(recs[0], 'backup_model_ids', ''), []):
|
||
if b and b not in mids:
|
||
mids.append(b)
|
||
for mid in mids:
|
||
r2 = await sor.sqlExe(
|
||
"SELECT name FROM llm_model WHERE id=${i}$ AND status='active' LIMIT 1",
|
||
{"i": mid})
|
||
await sor.sqlExe("COMMIT", {})
|
||
if r2:
|
||
n = getattr(r2[0], 'name', '') or ''
|
||
if n:
|
||
return n
|
||
return ''
|
||
|
||
|
||
# ────────────────────────── 上游调用（组包借鉴 uapi） ──────────────────────────
|
||
# uapi 模块（llmage 即调它组包）的做法：path/headers/body 全是 Jinja2 模板，
|
||
# tmpl_engine.renders() 渲染后 json.loads；api_key 等作模板变量注入命名空间。
|
||
# 本层同款实现：模板来自 llm_api_profile（模型挂的适配模板），缺省用内置
|
||
# openai-chat 模板——新增供应商形态只加 profile 记录，不改代码。
|
||
|
||
_tmpl_engine_cache = None
|
||
|
||
|
||
async def _render_tmpl(tmplstr, ns):
|
||
"""渲染一段模板字符串（uapi rendertmpl 同款）。
|
||
|
||
服务进程用 ahserver 的 tmpl_engine（启动时 setupTemplateEngine 注入，异步）；
|
||
独立脚本环境没有引擎时用 jinja2.Environment 兜底（同样异步渲染）。
|
||
"""
|
||
global _tmpl_engine_cache
|
||
te = _tmpl_engine_cache
|
||
if te is None:
|
||
try:
|
||
from ahserver.serverenv import ServerEnv
|
||
te = getattr(ServerEnv(), 'tmpl_engine', None)
|
||
except Exception:
|
||
te = None
|
||
if te is not None and callable(getattr(te, 'renders', None)):
|
||
_tmpl_engine_cache = te
|
||
return await te.renders(tmplstr, ns)
|
||
# 兜底：原生 jinja2.Environment 没有 renders 方法，用 from_string+render_async
|
||
try:
|
||
from jinja2 import Environment
|
||
env = Environment(enable_async=True)
|
||
_tmpl_engine_cache = None # 原生引擎不缓存（避免掩盖后续真正的引擎注入）
|
||
t = env.from_string(tmplstr)
|
||
return await t.render_async(ns)
|
||
except Exception as e:
|
||
raise GovernError('模板渲染失败（组包不可用）：%s' % str(e)[:200])
|
||
|
||
|
||
async def _render_headers(tmpl, ns):
|
||
"""headers 模板 → dict：Jinja2 渲染后 json.loads（同 uapi）。
|
||
|
||
模板空用内置 Bearer 模板；渲染结果缺 Authorization 头时兜底补标准 Bearer。
|
||
"""
|
||
s = await _render_tmpl((tmpl or '').strip() or _BUILTIN_HEADERS_TMPL, ns)
|
||
try:
|
||
out = json.loads(s)
|
||
except Exception as e:
|
||
raise GovernError('请求头模板渲染结果不是合法 JSON：%s（模板：%s）' % (
|
||
str(e)[:120], (tmpl or '(内置)')[:120]))
|
||
if not isinstance(out, dict):
|
||
raise GovernError('请求头模板必须渲染成 JSON 对象')
|
||
if not any(str(k).lower() == 'authorization' for k in out):
|
||
out['Authorization'] = 'Bearer %s' % (ns.get('api_key') or '')
|
||
out.setdefault('Content-Type', 'application/json')
|
||
return out
|
||
|
||
|
||
def _join_url(api_base, path):
|
||
"""api_base + path 拼接（2026-09-05 404 教训）：模板 path 若把 base_url
|
||
已有的路径前缀带上（提取 LLM 给 /api/v1/tasks/...，端点又是 .../api/v1），
|
||
朴素字符串相加会双前缀 404——重叠段去重后再拼。"""
|
||
base = (api_base or '').rstrip('/')
|
||
p = (path or '').strip()
|
||
if not base:
|
||
return p
|
||
if not p.startswith('/'):
|
||
p = '/' + p
|
||
from urllib.parse import urlparse
|
||
bp = urlparse(base).path.rstrip('/')
|
||
if bp and p.startswith(bp + '/'):
|
||
p = p[len(bp):]
|
||
return base + p
|
||
|
||
|
||
async def _build_upstream_body(ctx, payload):
|
||
"""请求体模板渲染（uapi data 模板同款）。
|
||
|
||
合并顺序：模型 default_params 打底 → 调用方显式参数覆盖 → 模板渲染。
|
||
模板取 profile.request_template，缺省内置 openai-chat 模板。
|
||
模板命名空间：model / messages / prompt / extra(白名单透传参数) /
|
||
params(全部) / api_key / org_id。
|
||
⚠️ prompt 便利变量必须注入（2026-09-08 实测根因）：提取生成器为
|
||
DashScope 生成类模型产的 request_template 用 {{ prompt }}（messages
|
||
末条文本），此前只有异步路径 _build_async_body 注入——同步生成路径
|
||
复用本函数时 StrictUndefined 渲染崩「Object of type Undefined」。
|
||
两路径命名空间必须一致。
|
||
"""
|
||
profile = ctx.get('profile') or {}
|
||
tmpl = (profile.get('request_template') or '').strip() or _BUILTIN_REQUEST_TMPL
|
||
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', '')
|
||
extra = {k: upstream[k] for k in upstream if k in _PASSTHROUGH_FIELDS}
|
||
ns = {
|
||
'model': model, 'messages': messages, 'extra': extra,
|
||
'prompt': _last_prompt_text(messages),
|
||
'params': upstream,
|
||
'api_key': ctx.get('api_key') or '',
|
||
'org_id': ctx.get('org_id') or '',
|
||
}
|
||
ns.update(_media_ns_extra())
|
||
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 _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 全模板组包），带瞬时错误重试。
|
||
|
||
require_choices=True：chat 语义（顶层必须有 choices）；
|
||
False：生成类语义（任意 JSON dict 即有效，产物由 response 模板/搜索提取）。
|
||
返回上游响应 dict。失败抛 ValueError（消息真实可行动）。
|
||
"""
|
||
import aiohttp
|
||
import asyncio
|
||
|
||
profile = ctx.get('profile') or {}
|
||
ns = {'api_key': ctx.get('api_key') or '', 'org_id': ctx.get('org_id') or ''}
|
||
path = await _render_tmpl((profile.get('path') or '').strip() or _DEFAULT_CHAT_PATH, ns)
|
||
if path.startswith('http://') or path.startswith('https://'):
|
||
url = path
|
||
else:
|
||
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)
|
||
# 可诊断性：只记长度不记值（真 key 不落日志）。print 确保进应用日志
|
||
# （模块 logging.getLogger 的 handler 未接应用日志管道，2026-09-04 实测）
|
||
print("[inference] 组包 url=%s auth头长度=%d body字段=%s" % (
|
||
url.split('?')[0], len(headers.get('Authorization', '') or ''),
|
||
sorted(body.keys())))
|
||
|
||
last = None
|
||
for attempt in range(_MAX_ATTEMPTS):
|
||
_t0 = time.time()
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.request(
|
||
method, url, headers=headers, json=body,
|
||
timeout=aiohttp.ClientTimeout(total=timeout, connect=_CONNECT_TIMEOUT),
|
||
) 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
|
||
await asyncio.sleep(2 * (attempt + 1))
|
||
continue
|
||
raise err
|
||
data = await resp.json(content_type=None)
|
||
if not isinstance(data, dict) or (require_choices and 'choices' not in data):
|
||
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]))
|
||
if attempt < _MAX_ATTEMPTS - 1:
|
||
await asyncio.sleep(2 * (attempt + 1))
|
||
continue
|
||
raise ValueError('上游调用失败（重试 %d 次仍失败，%s）: %s' % (
|
||
_MAX_ATTEMPTS, type(e).__name__, detail))
|
||
raise last if last else ValueError('上游调用失败')
|
||
|
||
|
||
async def _call_upstream_chat(ctx, payload):
|
||
"""POST 上游 chat/completions（OpenAI 兼容，顶层必须有 choices）。"""
|
||
return await _post_upstream(ctx, payload, require_choices=True)
|
||
|
||
|
||
def _find_artifact_url(out):
|
||
"""从上游 output 搜索生成产物 URL（response 模板缺失/渲染失败时的兜底）。
|
||
|
||
DashScope 同步生成实测形态（2026-09-08）：
|
||
output.choices[0].message.content[0].{image|video|audio}
|
||
旧形态：output.results[0].url。诚实搜索，找不到返回空串。
|
||
"""
|
||
try:
|
||
for ch in (out.get('choices') or []):
|
||
content = ((ch.get('message') or {}).get('content'))
|
||
segs = (content if isinstance(content, list)
|
||
else ([{'text': content}] if isinstance(content, str) else []))
|
||
for seg in segs:
|
||
if isinstance(seg, dict):
|
||
for k in ('image', 'video', 'audio'):
|
||
v = seg.get(k)
|
||
if isinstance(v, str) and v.strip():
|
||
return v
|
||
for r in (out.get('results') or []):
|
||
if isinstance(r, dict):
|
||
for k in ('url', 'image_url', 'video_url'):
|
||
if r.get(k):
|
||
return str(r[k])
|
||
except Exception:
|
||
pass
|
||
return ''
|
||
|
||
|
||
async def _sync_generation_inference(ctx, payload):
|
||
"""同步生成模型执行路径（2026-09-08 用户定夺补全的第三种交互形态）。
|
||
|
||
交互形态三种：流式（chat stream，前端 NDJSON 交付层）/ 异步任务（提交→
|
||
轮询，_async_inference）/ **同步生成（客户提交后模型即时给出产物）**——
|
||
此前 sync 路径只认 OpenAI chat（choices+CHAT_CAPS 门禁），DashScope
|
||
MultiModalConversation 类同步生成模型（qwen-image-plus 实测 403/无路径）
|
||
没有执行路径。本路径 = 一次 POST 拿结果，复用异步路径的 response 模板
|
||
渲染 + downloadfile2url 落地 + usages 记账，模板失效时按形态搜索产物。
|
||
"""
|
||
from .gateway import govern_settle
|
||
|
||
profile = ctx.get('profile') or {}
|
||
try:
|
||
data = await _post_upstream(ctx, payload, require_choices=False)
|
||
except Exception as e:
|
||
note = str(e)[:200]
|
||
await govern_settle(ctx, False,
|
||
'429 upstream: ' + note if ' 429' in note else note)
|
||
raise GovernError(str(e))
|
||
# 上游业务错误（DashScope：顶层 code/message，无 output）——如实报错不粉饰
|
||
if data.get('code') and not (data.get('output') or data.get('choices')):
|
||
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, msg[:200])
|
||
raise GovernError(msg)
|
||
|
||
out = data.get('output') or {}
|
||
usage = data.get('usage') or {}
|
||
# response 模板渲染（同异步路径）：产物 URL 经 downloadfile2url 落地
|
||
rendered = {}
|
||
resp_tmpl = (profile.get('response_template') or '').strip()
|
||
if resp_tmpl and '__note__' not in resp_tmpl:
|
||
ns = dict(data)
|
||
ns['output'] = out
|
||
ns['usage'] = usage
|
||
ns['request_id'] = data.get('request_id') or ''
|
||
ns.update(_media_ns_extra())
|
||
try:
|
||
s = await _render_tmpl(resp_tmpl, ns)
|
||
parsed = json.loads(s)
|
||
if isinstance(parsed, dict):
|
||
rendered = parsed
|
||
except Exception as e:
|
||
print("[inference] 同步生成 response模板渲染失败（回退产物搜索）: %s"
|
||
% str(e)[:150])
|
||
content = (rendered.get('video') or rendered.get('image')
|
||
or rendered.get('audio') or rendered.get('glb')
|
||
or rendered.get('3dmodel') or _find_artifact_url(out) or '')
|
||
if not content:
|
||
msg = ('同步生成调用成功但未在响应中找到产物（模板渲染与形态搜索均未命中）。'
|
||
'上游响应：%s' % json.dumps(data, ensure_ascii=False, default=str)[:300])
|
||
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, '' if usage else 'no-usage', usages)
|
||
result = {
|
||
"choices": [{"message": {"content": content, "role": "assistant"},
|
||
"finish_reason": "stop"}],
|
||
"usage": usage,
|
||
"output": out,
|
||
}
|
||
if rendered:
|
||
result["media"] = rendered # 统一出参：本地持久 URL（image/video/...）
|
||
return result
|
||
|
||
|
||
# ────────────────────────── 异步执行器（提交→轮询→取结果） ──────────────────────────
|
||
# 异步模型（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)
|
||
_normalize_media_aliases(upstream) # 上传媒体统一 xxx_file 契约
|
||
messages = upstream.pop('messages', [])
|
||
model = upstream.pop('model', '')
|
||
ns = {
|
||
'model': model, 'messages': messages,
|
||
'prompt': _last_prompt_text(messages),
|
||
'params': upstream, # 业务参数（resolution/duration/image_file 等）
|
||
'api_key': ctx.get('api_key') or '',
|
||
'org_id': ctx.get('org_id') or '',
|
||
}
|
||
ns.update(_media_ns_extra())
|
||
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, seq=0):
|
||
"""按查询步骤模板请求一次任务状态。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 = _join_url(ctx.get('api_base'), path)
|
||
headers = await _render_headers(qprofile.get('headers', ''), ns)
|
||
method = (qprofile.get('method') or 'GET').upper()
|
||
_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]))
|
||
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, 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 = _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 {}
|
||
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 = ''
|
||
_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, seq=_poll_seq)
|
||
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 {}
|
||
usage = last.get('usage') or {}
|
||
# response 模板渲染（对齐 uapi）：生成物 URL 必须经 downloadfile2url
|
||
# 落地为本地持久 URL 再给调用者（上游 URL 有效期短，视频仅 24 小时）。
|
||
# 同类能力出参字段统一（i2v/t2v → video，t2i → image）。
|
||
rendered = {}
|
||
resp_tmpl = (profile.get('response_template') or '').strip()
|
||
if resp_tmpl and '__note__' not in resp_tmpl:
|
||
ns = dict(out)
|
||
ns['output'] = out
|
||
ns['usage'] = usage
|
||
ns['task_id'] = task_id
|
||
ns.update(_media_ns_extra())
|
||
try:
|
||
s = await _render_tmpl(resp_tmpl, ns)
|
||
parsed = json.loads(s)
|
||
if isinstance(parsed, dict):
|
||
rendered = parsed
|
||
except Exception as e:
|
||
print("[inference] response模板渲染失败（回退原始URL，生成物未落地）: %s"
|
||
% str(e)[:150])
|
||
content = (rendered.get('video') or rendered.get('image')
|
||
or rendered.get('glb') or rendered.get('3dmodel')
|
||
or out.get('video_url') or out.get('image_url')
|
||
or out.get('result_url') or out.get('text') 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, '' if usage else 'no-usage', usages)
|
||
result = {
|
||
"choices": [{"message": {"content": content, "role": "assistant"},
|
||
"finish_reason": "stop"}],
|
||
"usage": usage,
|
||
"output": out,
|
||
"task_id": task_id,
|
||
}
|
||
if rendered:
|
||
result["media"] = rendered # 统一出参：本地持久 URL（video/image/...）
|
||
return result
|
||
|
||
|
||
# ────────────────────────── 对外主入口 ──────────────────────────
|
||
|
||
async def chat_inference(org_id, user_id, payload, model_name='', task_ref='',
|
||
project_id='', session_id=''):
|
||
# project_id（2026-09-10 用户定夺）：token 绑定的项目 ID 透传落 llm_usage.project_id，
|
||
# 非项目调用不传（govern_settle 兜底哨兵'0'），支撑按项目统计费用。
|
||
# 调用方（llm_proxy/v1 dspy）负责传入。
|
||
# session_id（2026-09-11 用户定夺）：会话粘性账号键——同会话固定账号省上游
|
||
# 缓存钱；调用方经 payload._session_id 或显式参数传入（llm_bridge 透传）。
|
||
"""统一推理：门禁链 → 上游调用 → 结算。返回上游响应 dict（OpenAI 兼容形态）。
|
||
|
||
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
|
||
|
||
if not isinstance(payload, dict) or not payload.get('messages'):
|
||
raise GovernError('请求体缺 messages')
|
||
model_name = (model_name or payload.get('model') or '').strip()
|
||
if not model_name:
|
||
model_name = await _pick_default_model_name(org_id, 't2t')
|
||
|
||
# 用途标记（辅助任务走辅助模型链）：取出后从 payload 剥离，不透传给上游
|
||
purpose = (payload.pop('_purpose', '') or '')
|
||
# 单次调用超时覆盖（长文本提取等慢任务）：剥离不透传，上限 900 秒
|
||
req_timeout = int(_fnum(payload.pop('_timeout', 0)) or 0)
|
||
# 会话粘性键（2026-09-11 用户定夺）：payload._session_id 优先于显式参数
|
||
# （llm_bridge 经 payload 透传，与 _purpose/_timeout 同通道）；剥离不上行
|
||
_sid = (payload.pop('_session_id', '') or '')
|
||
session_id = str(_sid or session_id or '')
|
||
|
||
est = max(_texts_len(payload.get('messages')) // EST_TOKENS_PER_CHAR, 200)
|
||
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,
|
||
project_id=project_id or '', session_id=session_id)
|
||
# 调用批次 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 记账）
|
||
model_row = ctx.get('model_row') or {}
|
||
if (model_row.get('sync_mode') or 'sync') == 'async':
|
||
return await _async_inference(ctx, payload, req_timeout)
|
||
# 同步模型按能力三分支（2026-09-08 用户定夺：交互形态第三种——同步生成）：
|
||
# ① 对话能力（t2t/i2t/m2t）→ OpenAI chat 语义（choices 结构）
|
||
# ② 生成类能力（t2i/tts 等）→ 一次 POST 即时拿产物（DashScope
|
||
# MultiModalConversation 类），走 response 模板/形态搜索提取 + 落地记账
|
||
# ③ 其余（embedding/rerank 等）→ 无执行路径，如实报错（禁静默）
|
||
from .selection import CHAT_CAPS
|
||
cap = (model_row.get('capability') or 't2t').strip().lower()
|
||
_GEN_OK = ('t2i', 'i2i', 't2v', 'i2v', 'r2v', 'tts', 'asr')
|
||
if cap in _GEN_OK:
|
||
return await _sync_generation_inference(ctx, payload)
|
||
if cap not in CHAT_CAPS:
|
||
from .gateway import govern_settle
|
||
await govern_settle(ctx, False, 'capability %s 无执行路径' % cap)
|
||
raise GovernError('模型「%s」能力为 %s：chat 入口支持对话（%s）与生成（%s）——'
|
||
'%s 请走其专属入口' % (
|
||
model_row.get('name', ''), cap, '/'.join(CHAT_CAPS),
|
||
'/'.join(_GEN_OK), cap))
|
||
try:
|
||
data = await _call_upstream_chat(ctx, payload)
|
||
except Exception as e:
|
||
# 失败结算：释放预授权 + failed 流水；上游 429 触发账号冷却
|
||
note = str(e)[:200]
|
||
await govern_settle(ctx, False,
|
||
'429 upstream: ' + note if ' 429' in note else note)
|
||
raise GovernError(str(e))
|
||
usage = data.get('usage') or {}
|
||
# 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
|
||
# 缺失 → derived 求值失败被兜底成 0 → 输入 token 永不计费（实测 qwen3.8-max
|
||
# uncache_tokens=0 记账失败根因）。sage/llmage 同步范式即 usages=json.dumps(usage)。
|
||
usages = dict(usage) if isinstance(usage, dict) else {}
|
||
if usages:
|
||
usages['model'] = ctx.get('model_id') or ''
|
||
await govern_settle(ctx, True, '' if usage else 'est', usages)
|
||
return data
|
||
|
||
|
||
async def list_models(org_id, catelogid='', limit=200):
|
||
"""按能力分类列模型（对齐 llmage v1/models 的分类方式）。
|
||
|
||
catelogid 空 = 全部能力。返回 [{'id','name','model_id','capability',
|
||
'vendor','status'}]。
|
||
机构未配策略（无可用模型）返回空列表，由调用方决定是否提示。
|
||
"""
|
||
from .gateway import _active_policy_org, _get_db
|
||
db, dbname = _get_db()
|
||
rows = []
|
||
async with db.sqlorContext(dbname) as sor:
|
||
if not await _active_policy_org(sor, org_id or ''):
|
||
return rows
|
||
sql = ("SELECT m.id, m.name, m.vendor_model_id, m.capability, m.status, "
|
||
"v.name AS vendor_name "
|
||
"FROM llm_model m LEFT JOIN llm_vendor v ON v.id=m.vendor_id "
|
||
"WHERE m.status='active'")
|
||
params = {}
|
||
cap = (catelogid or '').strip().lower()
|
||
if cap:
|
||
sql += " AND m.capability=${c}$"
|
||
params['c'] = cap
|
||
sql += " ORDER BY m.capability, m.name LIMIT %d" % int(limit)
|
||
recs = await sor.sqlExe(sql, params)
|
||
await sor.sqlExe("COMMIT", {})
|
||
for r in (recs or []):
|
||
rows.append({
|
||
'id': getattr(r, 'id', ''),
|
||
'name': getattr(r, 'name', ''),
|
||
'model_id': getattr(r, 'vendor_model_id', '') or getattr(r, 'name', ''),
|
||
'capability': getattr(r, 'capability', '') or 't2t',
|
||
'vendor': getattr(r, 'vendor_name', '') or '',
|
||
'status': getattr(r, 'status', ''),
|
||
})
|
||
return rows
|
||
|
||
|
||
def load_inference():
|
||
"""注册到 ServerEnv（供 dspy 端点调用）。"""
|
||
from ahserver.serverenv import ServerEnv
|
||
env = ServerEnv()
|
||
env.llm_chat_inference = chat_inference
|
||
env.llm_list_models = list_models
|
||
env.llm_pick_default_model_name = _pick_default_model_name
|
||
logger.info("[pipeline_llm] inference loaded")
|