diff --git a/pipeline_llm/gateway.py b/pipeline_llm/gateway.py index bd64898..fb6cc9f 100644 --- a/pipeline_llm/gateway.py +++ b/pipeline_llm/gateway.py @@ -317,17 +317,6 @@ def _filter_by_pref(candidates_with_ep, pref: str): return matched + rest -async def _model_protocol(sor, model: dict) -> str: - """模型适配模板的协议（2026-09-05 端点协议感知选择用）。查不到返回空串。""" - pid = (model.get('profile_id') or '').strip() - if not pid: - return '' - recs = await sor.sqlExe( - "SELECT protocol FROM llm_api_profile WHERE id=${i}$ LIMIT 1", {"i": pid}) - await sor.sqlExe("COMMIT", {}) - return (getattr(recs[0], 'protocol', '') or '').strip() if recs else '' - - async def _pick_candidate(sor, model: dict, pref: str, session_id: str = ''): """构建候选并选一个：协议过滤 → 区域偏好过滤 → 钱包余额门禁 → 冷却避让 → 会话粘性（同会话固定账号省上游缓存钱）→ 跨会话轮询。 @@ -354,49 +343,30 @@ async def _pick_candidate(sor, model: dict, pref: str, session_id: str = ''): accounts = await _account_candidates(sor, model) if not accounts: return False, '模型「%s」的供应商下无启用账号（或账号余额状态异常）' % model.get('name', '') - # 协议感知（2026-09-05 404 教训）：模型模板协议与端点标签匹配才可用。 - # 端点无标签（历史数据）视为通用——同供应商混布 compatible-mode/api/v1 - # 两类端点时，无标签过滤会把原生异步模型拼到 openai_compat 前缀下 404。 - mproto = await _model_protocol(sor, model) - if mproto: - tagged = [(i, e) for i, e in enumerate(endpoints) if (e.get('protocol') or '').strip()] - if tagged: - matched = {i for i, e in tagged if (e.get('protocol') or '').strip() == mproto} - untagged = {i for i, e in enumerate(endpoints) - if not (e.get('protocol') or '').strip()} - # 有协议匹配端点 → 只选匹配的（同账号多端点余额同分，按序会选到 - # 第一个即 compatible-mode → 404，必须排除不匹配项而非并集）； - # 无匹配但有未标注（历史）端点 → 兜底用未标注；全无 → 报错 - if matched: - usable = matched - elif untagged: - usable = untagged - else: - return False, ('供应商端点均为其他协议（模型模板协议 %s，端点协议 %s）——' - '原生异步(openai_compat 之外)与兼容模式端点 base_url 不同，' - '混用必 404。请在供应商端点目录添加协议 %s 的端点' - % (mproto, sorted({(e.get('protocol') or '') for _, e in tagged}), - mproto)) - else: - usable = set(range(len(endpoints))) - else: - usable = set(range(len(endpoints))) - # 展开 (账号, 端点) 对：账号选用的端点下标 + # 端点 = 纯连接点（主机根+区域+超时），所有端点对所有模型可互换（2026-09-12 + # 用户定夺：模型和端点不能强绑定，多端点=冗余/区域/多key，每个端点都支持 + # 全部模型）。请求形态（协议/版本段/接口路径）完全由模型适配模板 profile + # 的完整相对 path 承载：t2t → /compatible-mode/v1/chat/completions， + # i2v → /api/v1/services/aigc/...（主机根 + 完整相对 path = 任意端点可拼）。 + # 旧协议过滤已删（它正是间歇 404 根因：模型无 profile → 过滤跳过 → t2t 被 + # 轮询到 api/v1 原生端点拼出 /chat/completions 404）。 + # 展开 (账号, 端点) 对：账号选用的端点下标。 + # endpoint_ids 空 = 选用全部端点（2026-09-12：端点可互换后，空不应再是 + # 「跳过该账号」——历史数据里单端点供应商的账号常留空，语义即「用唯一端点」）。 pairs = [] for acc in accounts: idxs = _parse_json(acc.get('endpoint_ids'), []) if not idxs: - continue + idxs = list(range(len(endpoints))) for i in idxs: try: i = int(i) except (TypeError, ValueError): continue - if i in usable: + if 0 <= i < len(endpoints): pairs.append((acc, endpoints[i], i)) if not pairs: - return False, ('所有账号都未选用可用端点（模型治理→供应商账号→选用端点）。' - '模型模板协议 %s——请把该协议端点加入账号的选用端点' % (mproto or '未知')) + return False, '供应商端点目录与账号选用端点无交集——请检查模型治理→供应商端点目录与账号选用端点配置' pairs = _filter_by_pref(pairs, pref) if not pairs: return False, ('端点偏好为「仅 %s」，但没有可用的该区域端点。按合规要求不跨端点降级——' diff --git a/pipeline_llm/inference.py b/pipeline_llm/inference.py index 87ed077..4e84d55 100644 --- a/pipeline_llm/inference.py +++ b/pipeline_llm/inference.py @@ -417,10 +417,10 @@ async def _post_upstream(ctx, payload, require_choices=True): 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) + # 2026-09-12 用户定夺：模型 profile 禁止把端点写进 path（path 必须是完整相对 + # 路径，主机根由所选端点提供）——原「path 为全 URL 则无视端点」的覆盖分支已删， + # 端点对所有模型可互换。 + 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' @@ -690,10 +690,7 @@ 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) + 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() @@ -740,10 +737,7 @@ async def _async_inference(ctx, payload, req_timeout): # ── 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) + 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() diff --git a/pipeline_llm/init.py b/pipeline_llm/init.py index 94d55e7..9678e05 100644 --- a/pipeline_llm/init.py +++ b/pipeline_llm/init.py @@ -54,12 +54,18 @@ def normalize_endpoints(text): """端点目录输入归一化：支持「一行一个 URL」或 JSON 数组两种写法。 行格式：URL [region=domestic|international] [timeout=秒]，例如： - https://dashscope.aliyuncs.com/compatible-mode/v1 - https://api.example.com/v1 region=international timeout=30 + https://dashscope.aliyuncs.com + https://api.example.com region=international timeout=30 缺省 region=domestic、timeout=60。存储格式恒为 JSON 数组 [{base_url, region, timeout}]（治理链按此读取）。 非法输入抛 ValueError（中文提示，直接展示给用户）。 + + 2026-09-12 端点可互换约定：base_url 一律归一为主机根（scheme+域名）—— + 输入带路径段（旧数据/旧习惯的 /api/v1、/compatible-mode/v1 等）时确定性 + 剥掉；接口路径归 profile.path 承载（完整相对路径）。同 (主机根, region) + 去重合并，timeout 取最大值（单端点须覆盖该主机全部接口形态的超时需求）。 """ + from urllib.parse import urlparse s = (text or '').strip() if not s: return '' @@ -90,10 +96,32 @@ def normalize_endpoints(text): else: raise ValueError('第 %d 行未知选项 %s（支持 region= / timeout=）' % (ln_no, k)) eps.append(ep) + merged = [] for i, ep in enumerate(eps): if not isinstance(ep, dict) or not (ep.get('base_url') or '').strip(): raise ValueError('端点 #%d 缺少 base_url' % (i + 1)) - return json.dumps(eps, ensure_ascii=False) + pu = urlparse((ep.get('base_url') or '').rstrip('/')) + if not pu.netloc: + raise ValueError('端点 #%d base_url 非法（无主机名）：%s' % (i + 1, ep.get('base_url'))) + root = pu.scheme + '://' + pu.netloc + region = ep.get('region') or 'domestic' + try: + to = int(ep.get('timeout') or 60) + except (TypeError, ValueError): + to = 60 + hit = None + for m in merged: + if m['base_url'] == root and m['region'] == region: + hit = m + break + if hit is None: + m = {'base_url': root, 'region': region, 'timeout': to} + if ep.get('protocol'): + m['protocol'] = ep['protocol'] + merged.append(m) + else: + hit['timeout'] = max(hit['timeout'], to) + return json.dumps(merged, ensure_ascii=False) def _norm_endpoint_ids(raw):