"""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 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 _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): """门禁链 ①-⑥ + 预授权 + 加载模型/适配模板。 返回 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 '') 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/headers/request_template，全 Jinja2，借鉴 uapi） profile = {} pid = model_row.get('profile_id') or '' if pid: recs = await sor.sqlExe( "SELECT path, 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 '', '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 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 async def _build_upstream_body(ctx, payload): """请求体模板渲染（uapi data 模板同款）。 合并顺序：模型 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 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) # 可诊断性：只记长度不记值（真 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.post( 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 'choices' not in data: err = ValueError('上游响应缺 choices: %s' % ( 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 if attempt < _MAX_ATTEMPTS - 1: logger.warning("inference: 瞬时错误重试 %d/%d: %s", attempt + 1, _MAX_ATTEMPTS, e) await asyncio.sleep(2 * (attempt + 1)) continue raise ValueError('上游调用失败（重试 %d 次仍失败）: %s' % (_MAX_ATTEMPTS, e)) raise last if last else ValueError('上游调用失败') # ────────────────────────── 对外主入口 ────────────────────────── async def chat_inference(org_id, user_id, payload, model_name='', task_ref=''): """t2t 推理：门禁链 → 上游调用 → 结算。返回上游响应 dict（OpenAI 兼容）。 payload: 客户端请求体（messages 必需；tools/temperature/max_tokens 透传）。 model_name 空 → 机构策略缺省模型（主→备链第一个）。 失败抛 GovernError（消息真实可行动）。 """ 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') 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, 't2t', est, task_ref) 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'))) await govern_settle(ctx, True, rt, ct, '' if usage else 'est') return data async def list_models(org_id, catelogid='', limit=200): """按能力分类列模型（对齐 llmage v1/models 的分类方式）。 catelogid 空 = 全部能力。返回 [{'id','name','model_id','capability', 'vendor','status','price_input','price_output'}]。 机构未配策略（无可用模型）返回空列表，由调用方决定是否提示。 """ 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, " "m.price_input, m.price_output, 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', ''), 'price_input': float(_fnum(getattr(r, 'price_input', 0))), 'price_output': float(_fnum(getattr(r, 'price_output', 0))), }) 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")