From eb6ca2f4bbb5cbecd1b9af5a6a8c932a33ac6248 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Tue, 8 Sep 2026 13:33:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(inference):=20=E5=90=8C=E6=AD=A5=E7=94=9F?= =?UTF-8?q?=E6=88=90=E6=89=A7=E8=A1=8C=E8=B7=AF=E5=BE=84=E2=80=94=E2=80=94?= =?UTF-8?q?=E4=BA=A4=E4=BA=92=E5=BD=A2=E6=80=81=E7=AC=AC=E4=B8=89=E7=A7=8D?= =?UTF-8?q?(2026-09-08=E7=94=A8=E6=88=B7=E5=AE=9A=E5=A4=BA:stream/async?= =?UTF-8?q?=E4=B9=8B=E5=A4=96,=E5=AE=A2=E6=88=B7=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=E5=90=8E=E6=A8=A1=E5=9E=8B=E5=8D=B3=E6=97=B6=E7=BB=99=E5=87=BA?= =?UTF-8?q?=E4=BA=A7=E7=89=A9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _sync_generation_inference:一次POST即时拿产物,复用异步路径的response模板渲染 +downloadfile2url落地+usages记账;模板失效按DashScope实测形态搜索产物 (output.choices[0].message.content[0].{image|video|audio}) - _post_upstream抽公共(require_choices参数:chat语义/生成语义);上游业务错误如实抛 - chat_inference同步分流三分支:CHAT_CAPS→chat;生成类(t2i/i2i/t2v/i2v/r2v/tts/asr) →同步生成;其余(embedding/rerank)如实报错禁静默 - 协议字典加dashscope_sync种子(同步生成,不带X-DashScope-Async头) - 根治qwen-image-plus:文档MultiModalConversation.call(stream=False)被误配async +异步头→密钥不支持异步403;sync路径原先只认chat,t2i无执行路径 --- init/data.json | 1 + pipeline_llm/inference.py | 135 +++++++++++++++++++++++++++++++++++--- 2 files changed, 126 insertions(+), 10 deletions(-) diff --git a/init/data.json b/init/data.json index 1139fe0..d67f741 100644 --- a/init/data.json +++ b/init/data.json @@ -2,6 +2,7 @@ "appcodes": [ {"parentid": "llm_protocol", "parentname": "供应商协议", "items": [ {"k": "openai_compat", "v": "OpenAI兼容"}, + {"k": "dashscope_sync", "v": "DashScope同步生成(即时返回)"}, {"k": "dashscope_async", "v": "DashScope异步任务"}, {"k": "custom", "v": "自定义"} ]}, diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index b9e3b56..f436351 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -381,9 +381,11 @@ async def _build_upstream_body(ctx, payload): return body -async def _call_upstream_chat(ctx, payload): - """POST 上游（path/headers/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 @@ -424,8 +426,9 @@ async def _call_upstream_chat(ctx, payload): continue raise err data = await resp.json(content_type=None) - if not isinstance(data, dict) or 'choices' not in data: - err = ValueError('上游响应缺 choices: %s' % ( + 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 @@ -448,6 +451,111 @@ async def _call_upstream_chat(ctx, payload): 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 @@ -721,16 +829,23 @@ async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): 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/i2t/m2t）—— - # embedding/rerank/图视频生成等另有专属入口（list_models 按能力分栏选择）。 + # 同步模型按能力三分支（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 mismatch: %s not in chat caps' % cap) - raise GovernError('模型「%s」能力为 %s，聊天入口仅支持 %s——请按能力分类选择模型' % ( - model_row.get('name', ''), cap, '/'.join(CHAT_CAPS))) + 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: