diff --git a/pipeline_platform/platform_ability.py b/pipeline_platform/platform_ability.py index 59fd7e0..0414955 100644 --- a/pipeline_platform/platform_ability.py +++ b/pipeline_platform/platform_ability.py @@ -44,6 +44,106 @@ OWNER_ORG = "owner" # 其他协议的适配模板会存入 llm_api_profile 备查，但运行时暂不渲染。 RUNTIME_PROTOCOLS = ("openai_compat", "dashscope_async") +# ────────────────────── 会话级规格缓存（extract 锚定，2026-09-05）────────────────────── +# 根因：apply_llm_config 要求 LLM 把 extract 返回的大 JSON 规格逐字复制进参数， +# 模型「转述」大 JSON 必然编造结构（实测三轮全编造 vendor{}/api_profile{} 等契约外字段）。 +# 对策：extract 成功后把规格锚定到「会话级缓存」，apply/apply_model_pricing 用 +# use_last_extract=true 直接取用 + 少量覆盖项（vendor_name 等），LLM 永不搬运大 JSON。 +# +# 会话级隔离（用户硬要求）：缓存键 = pipeline_id:user_id:session_id，三段缺一不可， +# 防跨会话/跨用户/跨产线串数据。存 Redis db4（治理命名空间，gateway 已在用）， +# 跨 worker 进程不丢；TTL 兜底过期。Redis 不可用时降级进程内字典（仍按同键隔离）。 +_SPEC_CACHE_TTL = 7200 # 规格缓存 2 小时（一次配置会话足够） +_SPEC_CACHE_PREFIX = "llm_spec" +_spec_cache_local = {} # Redis 不可用时的进程内降级（同样按隔离键） + + +def _spec_cache_key(ctx): + """会话级隔离键：pipeline_id:user_id:session_id。任一段缺失返回 ''（拒绝缓存）。""" + pl = (ctx.get("pipeline_id") or "").strip() + uid = (ctx.get("user_id") or "").strip() + sid = (ctx.get("session_id") or "").strip() + if not (pl and uid and sid): + return '' + return "%s:%s:%s:%s" % (_SPEC_CACHE_PREFIX, pl, uid, sid) + + +def _spec_redis(): + """复用治理命名空间 Redis db4（与 gateway 限流同库，会话态用 db3 不冲突）。""" + try: + import redis as _r + try: + from appPublic.jsonConfig import getConfig + url = (getConfig().website or {}).get('session_redis', {}).get('url', '') + except Exception: + url = '' + if not url: + url = 'redis://127.0.0.1:6379/3' + base = url.rsplit('/', 1)[0] + return _r.Redis.from_url(base + '/4', socket_timeout=2) + except Exception: + return None + + +def _spec_save(ctx, spec): + """锚定规格到会话缓存。返回 True/False。无隔离键则不缓存（apply 会要求显式传 spec）。""" + key = _spec_cache_key(ctx) + if not key: + return False + try: + payload = json.dumps(spec, ensure_ascii=False) + except Exception: + return False + r = _spec_redis() + if r is not None: + try: + r.set(key, payload, ex=_SPEC_CACHE_TTL) + return True + except Exception: + pass + _spec_cache_local[key] = (payload, time.time() + _SPEC_CACHE_TTL) + return True + + +def _spec_load(ctx): + """取本会话锚定的规格。无则返回 None。严格会话级隔离（键含 session_id）。""" + key = _spec_cache_key(ctx) + if not key: + return None + r = _spec_redis() + raw = None + if r is not None: + try: + v = r.get(key) + raw = v.decode('utf-8') if isinstance(v, (bytes, bytearray)) else v + except Exception: + raw = None + if raw is None: + ent = _spec_cache_local.get(key) + if ent and ent[1] > time.time(): + raw = ent[0] + elif ent: + _spec_cache_local.pop(key, None) + if not raw: + return None + try: + return json.loads(raw) + except Exception: + return None + + +def _spec_overlay(base, overrides): + """覆盖项合并：仅允许少量标量覆盖（vendor_name 等），不做大 JSON 搬运。 + + overrides 只认白名单键，防止 LLM 借覆盖项塞回编造结构。 + """ + allowed = {'vendor_name', 'base_url', 'protocol', 'doc_url', 'doc_notes'} + out = dict(base or {}) + for k in allowed: + if k in (overrides or {}) and str(overrides.get(k) or '').strip(): + out[k] = str(overrides[k]).strip() + return out + # ────────────────────── 权限门禁（代码层） ────────────────────── @@ -297,7 +397,18 @@ async def _h_extract_llm_api_spec(sor, params, ctx): spec = json.loads(m.group(0)) except Exception: return "LLM 输出不是合法 JSON：%s" % txt[:300] - return json.dumps(spec, ensure_ascii=False) + # 会话级锚定（2026-09-05）：规格存会话缓存（键含 session_id，跨会话隔离）， + # 后续 apply_llm_config / apply_model_pricing 用 use_last_extract=true 取用—— + # LLM 不再搬运/转述大 JSON（转述必编造结构，实测三轮全编造）。 + anchored = _spec_save(ctx, spec) + out = dict(spec) + if anchored: + out["__anchored__"] = ("规格已锚定到本会话缓存。下一步调用 apply_llm_config 与 " + "apply_model_pricing 时传 {\"use_last_extract\": true}，" + "需改动只传覆盖项（overrides，仅 vendor_name/base_url/" + "protocol/doc_url/doc_notes 白名单键）。" + "禁止把本规格复制进 spec 参数——复制即编造。") + return json.dumps(out, ensure_ascii=False) # ────────────────────── 工具 4：写入配置（幂等） ────────────────────── @@ -310,21 +421,52 @@ def _f(v, default=0.0): async def _h_apply_llm_config(sor, params, ctx): - """按提取的规格写库：供应商（复用/新建）+ 适配模板 + 模型（含定价）。 + """按提取的规格写库：供应商（复用/新建）+ 适配模板 + 模型。 幂等：供应商按名称复用；模型按 (vendor, vendor_model_id) 复用—— - 已存在则更新定价，不重复创建。 + 已存在则更新，不重复创建。 + + 规格来源（2026-09-05 防编造改造）： + use_last_extract=true（推荐）→ 取本会话 extract 锚定的规格， + overrides 白名单覆盖（vendor_name 等少量标量）； + 否则用 params.spec（LLM 手传，历史路径）——结构不符时报错列出期望键。 """ err = await _require_owner(sor, ctx) if err: return err - try: - spec = json.loads(params.get("spec") or "{}") - except Exception: - return "spec 不是合法 JSON" + use_last = str(params.get("use_last_extract") or "").lower() in ("true", "1", "yes") + overrides = params.get("overrides") or {} + if isinstance(overrides, str): + try: + overrides = json.loads(overrides) + except Exception: + overrides = {} + spec = None + if use_last: + spec = _spec_load(ctx) + if spec is None: + return ("本会话没有锚定的提取规格——先调 extract_llm_api_spec（成功后规格自动" + "锚定本会话），再传 {\"use_last_extract\": true}。" + "注意会话级隔离：其他会话提取的规格本会话不可见，需在本会话重新提取。") + spec = _spec_overlay(spec, overrides) + else: + try: + spec = json.loads(params.get("spec") or "{}") + except Exception: + return "spec 不是合法 JSON" + # 结构校验（修B：报错给出路，不让模型盲改重试） + if not isinstance(spec, dict) or "vendor_name" not in spec or "models" not in spec: + got = sorted(spec.keys())[:12] if isinstance(spec, dict) else "(非对象)" + return ("spec 结构不符——不要自己编写/改写规格结构。期望顶层键：" + "vendor_name / base_url / endpoints / protocol / chat_path / models / " + "pricing / doc_url / doc_notes（extract_llm_api_spec 的原样输出）。" + "你传入的顶层键：%s。正确做法：调 extract_llm_api_spec 后传 " + "{\"use_last_extract\": true}（规格已锚定本会话，禁止复制转述），" + "需改动加 overrides（仅 vendor_name/base_url/protocol/doc_url/doc_notes）。" + % got) vendor_name = (spec.get("vendor_name") or "").strip() if not vendor_name: - return "spec 缺少 vendor_name" + return "spec 缺少 vendor_name（可用 overrides.vendor_name 指定）" endpoints = spec.get("endpoints") or [] if not endpoints and spec.get("base_url"): endpoints = [{"base_url": spec["base_url"], "region": "domestic", "timeout": 60}] @@ -657,14 +799,28 @@ async def _h_apply_model_pricing(sor, params, ctx): 幂等键：模型已有 ppid 且方案描述含同一出处 URL → 更新时序（拉链）； 否则新建方案。价格只来自文档原文（提取层强制 doc_quote）。 + + 规格来源（2026-09-05 防编造改造）：use_last_extract=true → 取本会话 + extract 锚定规格（含 pricing/doc_url），禁止 LLM 复制转述大 JSON。 """ err = await _require_owner(sor, ctx) if err: return err - try: - spec = json.loads(params.get("spec") or "{}") - except Exception: - return "spec 不是合法 JSON" + use_last = str(params.get("use_last_extract") or "").lower() in ("true", "1", "yes") + if use_last: + spec = _spec_load(ctx) + if spec is None: + return ("本会话没有锚定的提取规格——先调 extract_llm_api_spec（成功后规格自动" + "锚定本会话，含 pricing 与 doc_url），再传 {\"use_last_extract\": true}。" + "会话级隔离：其他会话的规格本会话不可见。") + else: + try: + spec = json.loads(params.get("spec") or "{}") + except Exception: + return "spec 不是合法 JSON" + if not isinstance(spec, dict) or "pricing" not in spec: + return ("spec 结构不符——不要自己编写规格。正确做法：extract_llm_api_spec 后传 " + "{\"use_last_extract\": true}（pricing 与 doc_url 已锚定本会话）。") pricing = spec.get("pricing") or {} items = pricing.get("items") or [] if not items: @@ -996,15 +1152,22 @@ PLATFORM_TOOLS = [ ), ToolDefinition( name="apply_llm_config", - description="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型。幂等——已有模型只更新。定价不在此工具（用 apply_model_pricing）。仅owner组织角色可用。", - parameters={"spec": "extract_llm_api_spec 返回的 JSON 规格"}, + description="按提取的规格写入模型治理配置：供应商/端点/适配模板/模型。幂等。推荐 use_last_extract=true 取本会话锚定规格（禁止把规格复制进 spec——复制即编造）；需改动加 overrides（仅 vendor_name/base_url/protocol/doc_url/doc_notes）。仅owner组织角色可用。", + parameters={ + "use_last_extract": "true=用本会话 extract_llm_api_spec 锚定的规格（推荐，会话级隔离）", + "overrides": "覆盖项 JSON（白名单：vendor_name/base_url/protocol/doc_url/doc_notes，如用户要求改供应商名）", + "spec": "（仅无锚定时的兜底）extract 返回的 JSON 规格原文", + }, category="platform", requires_confirmation=True, ), ToolDefinition( name="apply_model_pricing", - description="定价自动导入：按提取规格的 pricing 建定价方案（pricing_program+时序YAML）并挂模型 ppid。幂等（已有 ppid 拉链更新时序）。一个定价方案只服务一个模型；定价相同的模型共享同一 ppid。价格必须来自文档原文（每条带 doc_quote 否则拒绝）。仅owner组织角色可用。", - parameters={"spec": "extract_llm_api_spec 返回的 JSON 规格（含 pricing 与 doc_url）"}, + description="定价自动导入：建定价方案（pricing_program+时序YAML）并挂模型 ppid。幂等（内容无变化跳过；同日原地更新；历史行才拉链——时序表始终只一条有效行）。一个定价方案只服务一个模型；定价相同的模型共享同一 ppid。推荐 use_last_extract=true 取本会话锚定规格（含 pricing/doc_url），禁止复制转述。仅owner组织角色可用。", + parameters={ + "use_last_extract": "true=用本会话锚定规格的 pricing+doc_url（推荐）", + "spec": "（仅无锚定时的兜底）extract 返回的 JSON 规格原文", + }, category="platform", requires_confirmation=True, ), @@ -1043,12 +1206,14 @@ PLATFORM_PROMPT = """ ## 模型自动配置全链工作流（用户给文档 URL 要求配置模型时，按序独立完成） 1. **先 `load_skill` 加载 `model-auto-config`**（含媒体转换铁律/能力分类先行/定价新模式），严格执行 2. `fetch_model_doc` 抓取官方文档（返回文本末尾带 [出处URL] 行） -3. `extract_llm_api_spec` 通读文档提取规格：端点/协议/能力/sync_mode/async_steps， - 以及 **pricing（每条价格带 doc_quote 文档原文）和 doc_url（照抄出处行）** +3. `extract_llm_api_spec` 通读文档提取规格——**成功后规格自动锚定到本会话缓存** + （会话级隔离），返回里有 __anchored__ 提示 4. 把提取摘要给用户确认（端点/能力/定价原文/出处），确认后： -5. `apply_llm_config` 落库供应商/适配模板/模型（模型描述自动带 [出处:URL]） -6. `apply_model_pricing` 建定价方案并挂 ppid（定价描述带出处URL+文档原文； - 一个方案只服务一个模型，无 model 维度无 filters，维度平铺；幂等拉链更新） +5. `apply_llm_config` 传 **{"use_last_extract": true}**（用户要求改名等只加 + overrides 白名单覆盖项，如 {"overrides": {"vendor_name": "阿里云百炼"}}）。 + **绝对禁止把规格 JSON 复制/转述进 spec 参数——转述必编造结构（历史教训： + 三轮全编造 vendor{}/api_profile{} 等不存在的字段）。** +6. `apply_model_pricing` 同样传 **{"use_last_extract": true}**（定价+出处已锚定） 7. `test_model_call` 真实调用测试（前置：账号已配 api_key，没配则如实告知用户去补， 不要伪造结果）——生成类模型传 params（image_file/resolution/duration 等） 8. `check_model_accounting` 检查记账（传上一步的 task_ref）： @@ -1057,6 +1222,12 @@ PLATFORM_PROMPT = """ - failed → 按 reason/hint 处置（未映射产品→告知用户在产品管理导入产线模型） 9. 汇报：配置/定价/测试/记账四段结果 + 出处 URL + 遗留事项（如模板需人工核对项） +## 工具报错自救规则 +- 报「spec 结构不符/缺少 xxx」→ **不要改写结构重试**，改用 use_last_extract=true； + 本会话没锚定就重新走 fetch→extract（锚定是会话级的，跨会话不可见） +- 报「能力分类未登记」→ 如实转告用户需先加能力分类（这是人工评审门，不要绕过） +- 报「media_audit/runtime_note/需人工核对」警告 → 原样转告用户，不假装完成 + ## 硬规则 - **定价只允许来自文档原文**（doc_quote 强制），禁止编造/换算/推导 - **出处必留**：模型注册表描述字段与定价方案描述字段都必须保存文档 URL