From 26d083bdd2f059faccdd8923eb5a43eea62b85c4 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Fri, 4 Sep 2026 18:21:10 +0800 Subject: [PATCH] =?UTF-8?q?refactor(inference):=20=E7=BB=84=E5=8C=85?= =?UTF-8?q?=E5=80=9F=E9=89=B4uapi=E6=A8=A1=E6=9D=BF=E6=B8=B2=E6=9F=93?= =?UTF-8?q?=E2=80=94=E2=80=94path/headers/body=E5=85=A8Jinja2(tmpl=5Fengin?= =?UTF-8?q?e.renders+json.loads),api=5Fkey=E8=B5=B0=E6=A8=A1=E6=9D=BF?= =?UTF-8?q?=E5=91=BD=E5=90=8D=E7=A9=BA=E9=97=B4=E6=B3=A8=E5=85=A5;?= =?UTF-8?q?=E5=BA=9F=E5=BC=83=E5=9C=9F=E5=88=B6=E5=AD=97=E7=AC=A6=E4=B8=B2?= =?UTF-8?q?=E6=9B=BF=E6=8D=A2(401=E6=A0=B9=E5=9B=A0=E6=96=B9=E5=90=91);?= =?UTF-8?q?=E5=86=85=E7=BD=AEopenai-chat=E7=BC=BA=E7=9C=81=E6=A8=A1?= =?UTF-8?q?=E6=9D=BF,=E6=96=B0=E5=BD=A2=E6=80=81=E5=8F=AA=E5=8A=A0profile?= =?UTF-8?q?=E4=B8=8D=E6=94=B9=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_llm/inference.py | 145 ++++++++++++++++++++++++++++++-------- 1 file changed, 114 insertions(+), 31 deletions(-) diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index 0dc3507..0fea1e8 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -19,9 +19,20 @@ import logging logger = logging.getLogger("pipeline_llm.inference") -# OpenAI 兼容缺省适配（模型未挂 profile / profile 缺字段时的兜底） +# OpenAI 兼容缺省适配（模型未挂 profile / profile 缺字段时的兜底）—— +# 组包方式借鉴 uapi 模块（llmage 即调 uapi 组包）：path/headers/body 全部 +# Jinja2 模板，经 tmpl_engine 渲染后 json.loads；api_key 作模板变量注入。 _DEFAULT_CHAT_PATH = "/chat/completions" -_DEFAULT_CHAT_HEADERS = '{"Authorization": "***}"}' +_BUILTIN_HEADERS_TMPL = '{"Authorization": "***", "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 @@ -98,7 +109,7 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability, raise GovernError( '模型「%s」能力为 %s，与请求能力 %s 不匹配。请按能力分类选择模型' % ( model_row.get('name', ''), cap, capability)) - # 适配模板（决定上游 path/headers） + # 适配模板（组包：path/headers/request_template，全 Jinja2，借鉴 uapi） profile = {} pid = model_row.get('profile_id') or '' if pid: @@ -111,6 +122,8 @@ async def _resolve_call(sor, org_id, user_id, model_name, capability, profile = { 'path': getattr(r, 'path', '') or '', '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'), {}) @@ -156,52 +169,122 @@ async def _pick_default_model_name(org_id, capability='t2t'): return '' -# ────────────────────────── 上游调用 ────────────────────────── +# ────────────────────────── 上游调用（组包借鉴 uapi） ────────────────────────── +# uapi 模块（llmage 即调它组包）的做法：path/headers/body 全是 Jinja2 模板， +# tmpl_engine.renders() 渲染后 json.loads；api_key 等作模板变量注入命名空间。 +# 本层同款实现：模板来自 llm_api_profile（模型挂的适配模板），缺省用内置 +# openai-chat 模板——新增供应商形态只加 profile 记录，不改代码。 -def _render_headers(tmpl, api_key): - """适配模板 headers：JSON 对象，值里 {api_key} 占位符替换为真实 key。""" - h = _parse_json(tmpl, None) - if not isinstance(h, dict) or not h: - h = _parse_json(_DEFAULT_CHAT_HEADERS, {"Authorization": "***}"}) - out = {} - for k, v in h.items(): - v = str(v) - if '{api_key}' in v: - v = v.replace('{api_key}', api_key) - out[str(k)] = v - # 兜底认证头（模板没写 Authorization 时补标准 Bearer） +_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 None or not callable(getattr(te, 'renders', None)): + try: + from jinja2 import Environment + te = Environment(enable_async=True) + except Exception: + raise GovernError('模板引擎不可用：jinja2 未安装或未初始化，无法组包上游请求') + _tmpl_engine_cache = te + return await te.renders(tmplstr, ns) + + +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' % api_key + out['Authorization'] = 'Bearer %s' % (ns.get('api_key') or '') out.setdefault('Content-Type', 'application/json') return out -async def _call_upstream_chat(ctx, payload): - """POST 上游 chat/completions（按适配模板 path/headers），带瞬时错误重试。 +async def _build_upstream_body(ctx, payload): + """请求体模板渲染（uapi data 模板同款）。 - 返回上游响应 dict。失败抛 GovernError/ValueError（消息真实可行动）。 + 合并顺序：模型 default_params 打底 → 调用方显式参数覆盖 → 模板渲染。 + 模板取 profile.request_template，缺省内置 openai-chat 模板。 + 模板命名空间：model / messages / extra(白名单透传参数) / params(全部) / + api_key / org_id。 + """ + 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, + 'params': upstream, + '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 _call_upstream_chat(ctx, payload): + """POST 上游（path/headers/body 全模板组包），带瞬时错误重试。 + + 返回上游响应 dict。失败抛 ValueError（消息真实可行动）。 """ import aiohttp import asyncio - url = (ctx.get('api_base') or '').rstrip('/') + ( - (ctx.get('profile') or {}).get('path') or _DEFAULT_CHAT_PATH) - headers = _render_headers((ctx.get('profile') or {}).get('headers', ''), - ctx.get('api_key') or '') + 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 = (ctx.get('api_base') or '').rstrip('/') + path + headers = await _render_headers(profile.get('headers', ''), ns) + body = await _build_upstream_body(ctx, payload) timeout = int(_fnum(ctx.get('timeout')) or _TOTAL_TIMEOUT) - - 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) + # 可诊断性：只记长度不记值（真 key 不落日志） + logger.info("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.post( - url, headers=headers, json=upstream, + url, headers=headers, json=body, timeout=aiohttp.ClientTimeout(total=timeout, connect=_CONNECT_TIMEOUT), ) as resp: if resp.status != 200: