"""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=''): """门禁链 ①-⑥ + 加载模型/适配模板。 purpose='utility'：辅助任务（分类/选择/摘要），模型链为 辅助→主→备 （需机构策略配置 utility_model_id；未配则等同普通链）。 返回 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 '') 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, 0, 0, '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 _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) # 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): 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() 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])) if attempt < _MAX_ATTEMPTS - 1: last = err await asyncio.sleep(2 * (attempt + 1)) continue raise err return data except (asyncio.TimeoutError, aiohttp.ClientError) as e: last = e # TimeoutError 的 str() 是空串，用 repr 保证消息真实可行动 detail = repr(e) or type(e).__name__ # 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, 0, 0, '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, 0, 0, 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, 0, 0, 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) 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): """按查询步骤模板请求一次任务状态。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() 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 = _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(submit_method, 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 {} 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, 0, 0, '' 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=''): """统一推理：门禁链 → 上游调用 → 结算。返回上游响应 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) 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) 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, 0, 0, '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, 0, 0, '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 根治，对齐异步路径）： # 定价引擎的 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, rt, ct, '' 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")