diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcher.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcher.java index ff07a0ec..a0a315b4 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcher.java +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/main/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcher.java @@ -538,6 +538,15 @@ public class SaaGraphDispatcher implements GenerationDispatcher { // player：K_PLAYER_PANEL（player 软门评判，诊断用）；缺/非法则省略。 putPlayerIfPresent(trace, s); + // ── U3/B5 救场阶梯扩展键（additive，与 worker _extract_trace 同口径；缺则省略键，存量 trace_json 字节兼容）── + // 严格逐键 putIfPresent：modelTier 仅「真升档态」（stage2 或 escalationEvents 非空）才落，默认 stage1 省略 + // （P0-1：保「无扩展键时键集与扩前一致」）；K_ESCALATION_EVENTS/K_GIVEUP_DUMP_PATH 未发生救场时缺 → 省略。 + // 任一键解析失败不连坐（外层 try 已兜底）。 + putEscalationKeysIfPresent(trace, s); + // cacheHit：SAA 路 callAndRecord 现只读 prompt/completion token（Spring AI Usage 接口无 cache 字段）， + // 无命中数据源 → 不伪造、整段省略（与 cost 省略同因：没有就不挂，避免 D11/审核台误读）。 + // SAA 缓存命中捕获 = follow-up（须改 callAndRecord 读 native usage + 加 token state 键，超本单元范围）。 + return trace.isEmpty() ? null : trace; } catch (Exception e) { // 命门：抽取异常一律吞 + warn，返 null（→ reqVO.trace=null → persistTraceQuietly 守卫旁路 → trace_json 留 NULL）， @@ -692,6 +701,46 @@ public class SaaGraphDispatcher implements GenerationDispatcher { } } + /** + * U3/B5 救场阶梯扩展键 additive 抽取（modelTier / escalationEvents / giveupDumpPath）。 + * + *

口径对齐 worker {@code _extract_trace}（camelCase 键名一致；worker 同段亦逐键 present-then-put）： + *

+ * + *

字节兼容：全 additive——未升档/未救场时三键全省略（modelTier 默认 stage1 不再落）， + * 存量及正常路 trace_json 键集与扩前一致；{@code ReadinessScorer} 不读这三键（仅诊断/成本台账展示），两路同读不破。 + * 单键解析失败不连坐（escalationEvents 脏 JSON → 该键省略，modelTier/giveupDumpPath 照常）。 + */ + private static void putEscalationKeysIfPresent(Map trace, OverAllState s) { + // escalationEvents：JSON 数组串 → 非空数组则原样塞（List）；缺/空/非数组/脏 JSON 则省略键（单键失败不连坐）。 + // 先解析它，因为「是否真升档」要据它（或 stage2）判定，决定 modelTier 是否落。 + JsonNode events = parseTrace(s.value(SaaStudioNodes.K_ESCALATION_EVENTS, String.class).orElse("")); + boolean hasEvents = events != null && events.isArray() && events.size() > 0; + if (hasEvents) { + trace.put("escalationEvents", TRACE_MAPPER.convertValue(events, java.util.List.class)); + } + // modelTier：终态模型档（stage1|stage2）。仅「真升档态」才塞——stage2 或 escalationEvents 非空； + // 默认 stage1（未升档，render 初值噪声）则省略（P0-1：保「无扩展键时键集与扩前一致」字节兼容）。 + String modelTier = s.value(SaaStudioNodes.K_MODEL_TIER, String.class).filter(v -> !v.isBlank()).orElse(null); + boolean escalated = "stage2".equals(modelTier) || hasEvents; + if (modelTier != null && escalated) { + trace.put("modelTier", modelTier); + } + // giveupDumpPath：放弃 dump 落盘路径。非空才塞；未放弃（缺）则省略。 + String giveupPath = s.value(SaaStudioNodes.K_GIVEUP_DUMP_PATH, String.class).filter(v -> !v.isBlank()).orElse(null); + if (giveupPath != null) { + trace.put("giveupDumpPath", giveupPath); + } + } + /** 兜底失败回调（图执行线程异常时调；handleCallback 再异常也只留痕不抛，避免拖垮后台线程）。 */ private void safeFailedCallback(Map job, String traceId, String cause) { try { diff --git a/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcherTraceTest.java b/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcherTraceTest.java index 47068d26..518f9f00 100644 --- a/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcherTraceTest.java +++ b/game-cloud/game-module-aigc/game-module-aigc-server/src/test/java/com/wanxiang/huijing/game/module/aigc/saa/SaaGraphDispatcherTraceTest.java @@ -337,4 +337,127 @@ class SaaGraphDispatcherTraceTest { Integer score = new ReadinessScorer().score(trace); assertNotNull(score, "cost 省略不应让算分报错（efficiency 中性 0.5）"); } + + // ============================== 用例8：U3/B5 救场阶梯扩展键 additive + 字节兼容 ============================== + + /** + * U3/B5 字节兼容：succeededStub 未设 救场阶梯键（K_MODEL_TIER/K_ESCALATION_EVENTS/K_GIVEUP_DUMP_PATH， + * 均无初值）→ 抽出 trace 不含 modelTier/escalationEvents/giveupDumpPath/cacheHit 四键。 + * 守门：未升档/未救场的存量回放，trace_json 键集与扩前一致（additive 不破存量）。 + */ + @Test + void extract_noEscalationKeys_omitsExtensionKeys() throws Exception { + Map trace = invokeExtract(succeededStub(), 8000L); + assertFalse(trace.containsKey("modelTier"), "未设 K_MODEL_TIER → modelTier 省略（字节兼容）"); + assertFalse(trace.containsKey("escalationEvents"), "未升档 → escalationEvents 省略（字节兼容）"); + assertFalse(trace.containsKey("giveupDumpPath"), "未放弃 → giveupDumpPath 省略（字节兼容）"); + // SAA 路无缓存命中数据源（callAndRecord 只读 prompt/completion token）→ cacheHit 整段省略，不伪造。 + assertFalse(trace.containsKey("cacheHit"), "SAA 路无命中数据源 → cacheHit 省略（不伪造）"); + } + + /** + * P0-1 字节兼容（核心守门）：render 节点默认落 {@code K_MODEL_TIER="stage1"}（初值，未升档）且无救场事件 + * → 抽出 trace 不含 modelTier 键。 + * + *

守门：正常未升档路（render 已落 stage1 初值）的 trace_json 键集必须与扩前一致——modelTier=stage1 是 + * 初值噪声，落它会平白给每条正常 trace 多一个 modelTier 键，破「无扩展键时键集与扩前一致」。本用例坐实 + * 「默认 stage1 → modelTier 省略」（仅 stage2 或 escalationEvents 非空才落），与 worker {@code _extract_trace} 同口径。 + */ + @Test + void extract_renderDefaultStage1_omitsModelTier() throws Exception { + // 模拟 render 节点真实初态：K_MODEL_TIER="stage1"（render out.put 落的初值），无救场事件/无放弃。 + Map init = new HashMap<>(); + init.put(SaaStudioNodes.K_STATUS, "succeeded"); + init.put(SaaStudioNodes.K_REPAIR_COUNT, 0); + init.put(SaaStudioNodes.K_GAME_ID, "saa-render-default"); + init.put(SaaStudioNodes.K_MODEL_TIER, "stage1"); // render 默认初值（未升档） + OverAllState s = newStubState(init); + + Map trace = invokeExtract(s, 5000L); + assertNotNull(trace); + assertFalse(trace.containsKey("modelTier"), + "render 默认 stage1（未升档）→ modelTier 省略（字节兼容：键集与扩前一致）"); + assertFalse(trace.containsKey("escalationEvents"), "无救场事件 → escalationEvents 省略"); + assertFalse(trace.containsKey("giveupDumpPath"), "未放弃 → giveupDumpPath 省略"); + } + + /** + * P0-1 边界：stage1 但 escalationEvents 非空（极端态：事件已 append 但 modelTier 尚未翻 stage2） + * → 仍判为「真升档态」，modelTier=stage1 也落（与 escalationEvents 同读，诚实反映轨迹）。 + */ + @Test + @SuppressWarnings("unchecked") + void extract_stage1WithEvents_keepsModelTier() throws Exception { + Map init = new HashMap<>(); + init.put(SaaStudioNodes.K_STATUS, "failed"); + init.put(SaaStudioNodes.K_REPAIR_COUNT, 5); + init.put(SaaStudioNodes.K_MODEL_TIER, "stage1"); // 尚未翻档，但已有救场事件 + init.put(SaaStudioNodes.K_ESCALATION_EVENTS, + "[{\"tierBefore\":\"stage1\",\"tierAfter\":\"stage2\",\"atFailCount\":5,\"ts\":\"2026-06-17T00:00:00Z\"}]"); + OverAllState s = newStubState(init); + + Map trace = invokeExtract(s, 6000L); + assertEquals("stage1", trace.get("modelTier"), "escalationEvents 非空 → 判真升档态，modelTier 照落"); + List> events = (List>) trace.get("escalationEvents"); + assertNotNull(events, "escalationEvents 非空应抽出"); + assertEquals(1, events.size()); + } + + /** + * U3/B5 additive 抽取：喂救场升档后的桩（modelTier=stage2 + escalationEvents 数组串 + giveupDumpPath）→ + * 抽出 trace 含 modelTier=stage2、escalationEvents（List 含一条升档事件）、giveupDumpPath。 + * 守门：救场升档/放弃轨迹随 trace 落库，与 worker _extract_trace 同口径（camelCase）。 + */ + @Test + @SuppressWarnings("unchecked") + void extract_withEscalationKeys_extractsAdditive() throws Exception { + Map init = new HashMap<>(); + init.put(SaaStudioNodes.K_STATUS, "failed"); + init.put(SaaStudioNodes.K_REPAIR_COUNT, 6); + init.put(SaaStudioNodes.K_GAME_ID, "saa-escalated"); + // 救场升档真实形态：escalate 节点已把 modelTier=stage2 并 append 一条事件（JSON 数组串）。 + init.put(SaaStudioNodes.K_MODEL_TIER, "stage2"); + init.put(SaaStudioNodes.K_ESCALATION_EVENTS, + "[{\"tierBefore\":\"stage1\",\"tierAfter\":\"stage2\",\"atFailCount\":5,\"ts\":\"2026-06-17T00:00:00Z\"}]"); + // giveup 节点产的 dump 落盘路径（U5 真落盘；此处验抽取）。 + init.put(SaaStudioNodes.K_GIVEUP_DUMP_PATH, "games/_wg1-gen/saa-escalated/evidence/giveup-dump.json"); + OverAllState s = newStubState(init); + + Map trace = invokeExtract(s, 30000L); + assertNotNull(trace); + assertEquals("stage2", trace.get("modelTier"), "modelTier 应抽出升档后档位"); + List> events = (List>) trace.get("escalationEvents"); + assertNotNull(events, "escalationEvents 应抽出为 List"); + assertEquals(1, events.size(), "应含一条升档事件"); + assertEquals("stage1", events.get(0).get("tierBefore")); + assertEquals("stage2", events.get(0).get("tierAfter")); + assertEquals("games/_wg1-gen/saa-escalated/evidence/giveup-dump.json", trace.get("giveupDumpPath")); + } + + /** + * U3/B5 边界：escalationEvents 为脏 JSON / 空数组 → 该键省略，但 modelTier/giveupDumpPath 仍正常抽出（单键失败不连坐）。 + */ + @Test + void extract_escalationEventsDirtyOrEmpty_omitsOnlyThatKey() throws Exception { + // 空数组 "[]" → escalationEvents 省略（size==0 不塞）；脏 JSON 同样省略。 + Map init = new HashMap<>(); + init.put(SaaStudioNodes.K_STATUS, "failed"); + init.put(SaaStudioNodes.K_REPAIR_COUNT, 5); + init.put(SaaStudioNodes.K_MODEL_TIER, "stage2"); + init.put(SaaStudioNodes.K_ESCALATION_EVENTS, "[]"); // 空数组 → 省略 + init.put(SaaStudioNodes.K_GIVEUP_DUMP_PATH, "games/_wg1-gen/x/evidence/giveup-dump.json"); + OverAllState s = newStubState(init); + Map trace = invokeExtract(s, 1000L); + assertEquals("stage2", trace.get("modelTier"), "空 escalationEvents 不连坐 modelTier"); + assertFalse(trace.containsKey("escalationEvents"), "空数组 → escalationEvents 省略"); + assertTrue(trace.containsKey("giveupDumpPath"), "空 escalationEvents 不连坐 giveupDumpPath"); + + // 脏 JSON → escalationEvents 省略，整体不抛。 + init.put(SaaStudioNodes.K_ESCALATION_EVENTS, "{不是数组"); + OverAllState s2 = newStubState(init); + Map trace2 = invokeExtract(s2, 1000L); + assertNotNull(trace2, "脏 escalationEvents 不应让抽取崩"); + assertFalse(trace2.containsKey("escalationEvents"), "脏 JSON → escalationEvents 省略"); + assertEquals("stage2", trace2.get("modelTier"), "脏 escalationEvents 不连坐 modelTier"); + } } diff --git a/wg1/gen-worker/tests/test_cost.py b/wg1/gen-worker/tests/test_cost.py new file mode 100644 index 00000000..31147fcf --- /dev/null +++ b/wg1/gen-worker/tests/test_cost.py @@ -0,0 +1,312 @@ +# -*- coding: utf-8 -*- +"""tests/test_cost.py —— U4/B9 缓存命中落成本台账 + U3 trace 扩展键字节兼容自动化用例。 + +【为什么 AST 隔离加载而非直接 import（对齐 wg1_groupb_test.py 范式）】 + - cost.py 顶部 `import _client`（→ import openai）+ `import httpx`：6c6g 裸 python3 无这些依赖。 + - 本测试只验 `cache_hit_tokens` / `compute`（纯算术，零网络/零外部依赖）与 service._extract_trace（纯 dict 操作）， + 故用 AST 隔离把这三个函数体单独 exec 出来（绕开模块级 import 副作用），6c6g 与 mini-desktop 同款可跑。 + - compute 体内只用 dict.get / 算术；cache_hit_tokens 体内只用 getattr/isinstance；_extract_trace 体内纯 dict——均零外部依赖。 + +【覆盖（执行版 §5.8 / 计划 U4 test scenarios）】 + U4-① happy：命中 token 存在（两套字段之一）→ 折扣计入 totalRmb（低于不计缓存的账）。 + U4-② edge：两套字段都缺 → 按 miss 全价计（cached=0，与扩前口径一致）。 + U4-③ edge：两套字段都在 → 取 max。 + U4-④ cache_ratio 缺/None（MiniMax-M2.7）→ 视为 1.0（不折扣，全价，不报错）。 + U4-⑤ cached 截断到 [0, prompt_tokens]（脏 usage 命中超 prompt 不致负 quota）。 + U4-⑥ AgentScope ChatUsage.cache_input_tokens 路命中可被 cache_hit_tokens 捕获。 + U3-⑦ _extract_trace cacheHit 落值（promptCacheHitTokens + source）。 + U3-⑧ 字节兼容：result 无扩展键 → trace 省略 modelTier/escalationEvents/giveupDumpPath/cacheHit（与扩前键集一致）。 + U3-⑨ result 有扩展键 → trace 抽出（additive 生效）。 + U3-⑩⑪ P0-1：modelTier="stage1"（render 默认）无事件→省略；stage1+事件→照落（与 Java 同口径）。 + U3-⑫ P1-2：脏/空形态（空串 modelTier、非数组/空 list escalationEvents、空白 giveupDumpPath）只省略键、不落脏值（镜像 Java）。 + P1-1：cost.costFallback 标记随 trace 透传（tokens-only 降级，无 totalRmb）。 + P0-2：config.RecordingChatModel._safe_int 对脏值/None/负数绝不抛（usage 解析不中断生成）。 + P2-1：studio._classify_cache_source 据命中模型名归类，独立于 pricing 是否可达。 + +运行：python3 wg1/gen-worker/tests/test_cost.py（退出码 0=全过）；或 mini-desktop pytest -q（亦被收集为 test_*）。 +""" + +import ast +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] # wg1/gen-worker + +_passed = 0 +_failed = 0 + + +def _check(name, cond, detail=""): + global _passed, _failed + if cond: + _passed += 1 + print(f" [PASS] {name} {detail}") + else: + _failed += 1 + print(f" [FAIL] {name} {detail}") + + +def _exec_funcdef(src, node, ns): + """把一个 FunctionDef 节点单独 exec 进 ns（去装饰器——静态方法的 @staticmethod 脱离类无意义且会报错）。""" + seg = ast.get_source_segment(src, node) + # 去掉前导装饰器行（如 @staticmethod）：FunctionDef.lineno 指向 def 行，装饰器在其上方，get_source_segment 含装饰器。 + lines = seg.splitlines() + while lines and lines[0].lstrip().startswith("@"): + lines.pop(0) + # 顶格化（类内方法有缩进；exec 顶层不允许缩进）。 + import textwrap + exec(textwrap.dedent("\n".join(lines)), ns) + + +def _load_funcs_from(src_path, names): + """AST 隔离加载指定函数（绕开模块级 import 副作用）。names 内函数体须零外部依赖。 + + 支持两种：① 模块级顶层函数；② 类内方法（递归进 ClassDef 找；去装饰器+顶格化后单独 exec）。 + """ + src = src_path.read_text(encoding="utf-8") + tree = ast.parse(src) + ns = {} + found = {} + + def _scan(body): + for node in body: + if isinstance(node, ast.FunctionDef) and node.name in names and node.name not in found: + _exec_funcdef(src, node, ns) + found[node.name] = ns[node.name] + elif isinstance(node, ast.ClassDef): + _scan(node.body) # 递归进类体找方法 + + _scan(tree.body) + missing = [n for n in names if n not in found] + if missing: + raise RuntimeError(f"{src_path.name} 未找到函数：{missing}") + return found + + +# 模拟 usage 对象（openai 原始 usage 形态：属性式取值 + 嵌套 prompt_tokens_details）。 +class _FakeDetails: + def __init__(self, cached_tokens=None): + if cached_tokens is not None: + self.cached_tokens = cached_tokens + + +class _FakeUsage: + """可配 DeepSeek 顶层 prompt_cache_hit_tokens / MiniMax 嵌套 prompt_tokens_details.cached_tokens / + AgentScope cache_input_tokens 任意子集。""" + + def __init__(self, prompt_cache_hit_tokens=None, cached_tokens=None, cache_input_tokens=None): + if prompt_cache_hit_tokens is not None: + self.prompt_cache_hit_tokens = prompt_cache_hit_tokens + if cached_tokens is not None: + self.prompt_tokens_details = _FakeDetails(cached_tokens) + if cache_input_tokens is not None: + self.cache_input_tokens = cache_input_tokens + + +# 固定 pricing/口径（与生产 /api/pricing 真值同形：deepseek 低 cache_ratio、M2.7 cache_ratio=None）。 +PRICING = { + "deepseek-v4-flash": {"model_ratio": 0.06, "completion_ratio": 1.75, "cache_ratio": 0.025}, + "MiniMax-M3": {"model_ratio": 0.3, "completion_ratio": 4, "cache_ratio": 0.2}, + "MiniMax-M2.7": {"model_ratio": 0.15, "completion_ratio": 4, "cache_ratio": None}, +} +QPU = 500000 +USD = 7.3 + + +def test_cache_hit_tokens(cache_hit_tokens): + print("\n=== U4 cache_hit_tokens：两套字段兼容取 max ===") + # ② 两套字段都缺 → 0。 + _check("两套字段都缺→0", cache_hit_tokens(_FakeUsage()) == 0) + _check("None usage→0", cache_hit_tokens(None) == 0) + # 单字段：DeepSeek 顶层。 + _check("仅 DeepSeek prompt_cache_hit_tokens", + cache_hit_tokens(_FakeUsage(prompt_cache_hit_tokens=120)) == 120) + # 单字段：MiniMax 嵌套。 + _check("仅 MiniMax cached_tokens", + cache_hit_tokens(_FakeUsage(cached_tokens=88)) == 88) + # ⑥ AgentScope ChatUsage.cache_input_tokens。 + _check("仅 AgentScope cache_input_tokens", + cache_hit_tokens(_FakeUsage(cache_input_tokens=77)) == 77) + # ③ 两套都在 → 取 max。 + _check("两套都在→取 max", + cache_hit_tokens(_FakeUsage(prompt_cache_hit_tokens=50, cached_tokens=200)) == 200) + _check("三路都在→取 max", + cache_hit_tokens(_FakeUsage(prompt_cache_hit_tokens=50, cached_tokens=200, cache_input_tokens=300)) == 300) + # dict 形态（_client.chat 返回值或已序列化 usage）兼容。 + _check("dict 形态 prompt_cache_hit_tokens", + cache_hit_tokens({"prompt_cache_hit_tokens": 33}) == 33) + _check("dict 嵌套 details", + cache_hit_tokens({"prompt_tokens_details": {"cached_tokens": 44}}) == 44) + + +def test_compute_cache_discount(compute): + print("\n=== U4 compute：cache_ratio 折扣口径 ===") + P, C = 10000, 2000 # prompt / completion + # 全价（cached=0，旧口径）。 + full = compute("deepseek-v4-flash", P, C, PRICING, QPU, USD) + # ① 命中 8000 token → 折扣（deepseek cache_ratio=0.025，命中段近乎免单）→ rmb < 全价。 + hit = compute("deepseek-v4-flash", P, C, PRICING, QPU, USD, cached_tokens=8000) + _check("命中折扣 < 全价（deepseek 0.025）", hit["rmb"] < full["rmb"], + f"hit={hit['rmb']} full={full['rmb']}") + # 手算校验：quota = ((P-8000)+8000*0.025 + C*1.75)*0.06。 + expect_quota = ((P - 8000) + 8000 * 0.025 + C * 1.75) * 0.06 + _check("命中折扣 quota 手算吻合", abs(hit["quota"] - round(expect_quota, 2)) < 0.01, + f"got={hit['quota']} expect={round(expect_quota, 2)}") + _check("compute 落 cached_tokens 字段", hit["cached_tokens"] == 8000) + _check("compute 落 cache_ratio 字段", hit["cache_ratio"] == 0.025) + # ② cached=0 退化为旧口径：quota == (P + C*completion_ratio)*model_ratio（与扩前字节兼容）。 + legacy_quota = (P + C * 1.75) * 0.06 + _check("cached=0 退化旧口径（字节兼容）", abs(full["quota"] - round(legacy_quota, 2)) < 0.01, + f"got={full['quota']} expect={round(legacy_quota, 2)}") + # ④ cache_ratio=None（M2.7）→ 视为 1.0，命中不折扣（quota 与全价相等）。 + m27_full = compute("MiniMax-M2.7", P, C, PRICING, QPU, USD) + m27_hit = compute("MiniMax-M2.7", P, C, PRICING, QPU, USD, cached_tokens=5000) + _check("cache_ratio=None→不折扣（命中=全价）", abs(m27_hit["quota"] - m27_full["quota"]) < 0.01, + f"hit={m27_hit['quota']} full={m27_full['quota']}") + _check("cache_ratio=None 回退 1.0", m27_hit["cache_ratio"] == 1.0) + # ⑤ cached 截断到 [0, prompt_tokens]：命中超 prompt → 截到 prompt，quota 不为负。 + over = compute("deepseek-v4-flash", 100, 50, PRICING, QPU, USD, cached_tokens=99999) + _check("命中超 prompt→截断到 prompt", over["cached_tokens"] == 100) + _check("截断后 quota ≥ 0", over["quota"] >= 0, f"quota={over['quota']}") + # 负命中 → 截到 0。 + neg = compute("deepseek-v4-flash", 100, 50, PRICING, QPU, USD, cached_tokens=-5) + _check("负命中→截到 0", neg["cached_tokens"] == 0) + # MiniMax-M3 命中折扣（cache_ratio=0.2）。 + m3_full = compute("MiniMax-M3", P, C, PRICING, QPU, USD) + m3_hit = compute("MiniMax-M3", P, C, PRICING, QPU, USD, cached_tokens=6000) + _check("M3 命中折扣 < 全价（0.2）", m3_hit["rmb"] < m3_full["rmb"], + f"hit={m3_hit['rmb']} full={m3_full['rmb']}") + + +def test_extract_trace_compat(extract): + print("\n=== U3 _extract_trace：cacheHit 落值 + 扩展键字节兼容 ===") + # 最小 run_studio 形态 result（必填子集 + cache_hit）。 + base = { + "game_id": "saa-x", "stage": "stage1", "pass": True, "repairs": 0, + "wall_s": 12.3, "models": {"code": "deepseek-v4-flash"}, "attempts": [{"attempt": 0, "role": "code"}], + "tokens": {"prompt": 10000, "completion": 2000}, + "cache_hit": {"promptCacheHitTokens": 8000, "source": "deepseek"}, + } + t = extract(base) + # ⑦ cacheHit 落值。 + _check("cacheHit 落值 promptCacheHitTokens", + t.get("cacheHit", {}).get("promptCacheHitTokens") == 8000, f"cacheHit={t.get('cacheHit')}") + _check("cacheHit 落 source", t.get("cacheHit", {}).get("source") == "deepseek") + + # ⑧ 字节兼容：无扩展键 result → trace 不含 modelTier/escalationEvents/giveupDumpPath，且无 cacheHit。 + no_ext = dict(base) + no_ext.pop("cache_hit") + t2 = extract(no_ext) + for k in ("modelTier", "escalationEvents", "giveupDumpPath", "cacheHit"): + _check(f"无扩展键→省略 {k}", k not in t2, f"keys={sorted(t2.keys())}") + + # ⑨ 有扩展键 result → 抽出（additive）。 + with_ext = dict(base) + with_ext["modelTier"] = "stage2" + with_ext["escalationEvents"] = [{"tierBefore": "stage1", "tierAfter": "stage2", "atFailCount": 5, "ts": "2026-06-17T00:00:00Z"}] + with_ext["giveupDumpPath"] = "games/_wg1-gen/saa-x/evidence/giveup-dump.json" + t3 = extract(with_ext) + _check("有扩展键→modelTier=stage2", t3.get("modelTier") == "stage2") + _check("有扩展键→escalationEvents 抽出", isinstance(t3.get("escalationEvents"), list) and len(t3["escalationEvents"]) == 1) + _check("有扩展键→giveupDumpPath 抽出", t3.get("giveupDumpPath", "").endswith("giveup-dump.json")) + + # ⑩ P0-1 字节兼容：modelTier="stage1"（render 默认初值）且无 escalationEvents → modelTier 省略（与 Java 同口径）。 + s1 = dict(base) + s1["modelTier"] = "stage1" + t5 = extract(s1) + _check("P0-1 默认 stage1 无事件→modelTier 省略", "modelTier" not in t5, f"keys={sorted(t5.keys())}") + + # ⑪ P0-1 边界：stage1 但 escalationEvents 非空 → 判真升档态，modelTier=stage1 照落（与 Java 同口径）。 + s1e = dict(base) + s1e["modelTier"] = "stage1" + s1e["escalationEvents"] = [{"tierBefore": "stage1", "tierAfter": "stage2", "atFailCount": 5, "ts": "x"}] + t6 = extract(s1e) + _check("P0-1 stage1+事件→modelTier 照落", t6.get("modelTier") == "stage1") + _check("P0-1 stage1+事件→escalationEvents 抽出", isinstance(t6.get("escalationEvents"), list)) + + # ⑫ P1-2 形态校验（镜像 Java 严格口径）：脏/空形态只省略键、不落脏值。 + dirty = dict(base) + dirty["modelTier"] = "" # 空串 → 省略（须非空串） + dirty["escalationEvents"] = "not-a-list" # 非 list → 省略（须非空 list） + dirty["giveupDumpPath"] = " " # 纯空白 → 省略（须非空串） + t7 = extract(dirty) + _check("P1-2 空串 modelTier→省略", "modelTier" not in t7) + _check("P1-2 非数组 escalationEvents→省略", "escalationEvents" not in t7) + _check("P1-2 空白 giveupDumpPath→省略", "giveupDumpPath" not in t7) + + # P1-2：空 list escalationEvents → 省略（与 Java size>0 同口径）；stage2 仍照落（modelTier 不连坐）。 + empty_ev = dict(base) + empty_ev["modelTier"] = "stage2" + empty_ev["escalationEvents"] = [] + t8 = extract(empty_ev) + _check("P1-2 空 list escalationEvents→省略", "escalationEvents" not in t8) + _check("P1-2 空 list 不连坐 modelTier(stage2)", t8.get("modelTier") == "stage2") + + # P1-1：costFallback 标记随 trace.cost 透传（tokens-only 降级口径）。 + fb = dict(base) + fb["cost"] = {"error": "pricing unreachable", "costFallback": True, + "by_model": {"deepseek-v4-flash": {"in": 100, "out": 50, "cached": 0}}} + t9 = extract(fb) + _check("P1-1 costFallback 透传", t9.get("cost", {}).get("costFallback") is True, f"cost={t9.get('cost')}") + _check("P1-1 降级 cost 无 totalRmb", "totalRmb" not in t9.get("cost", {})) + + # snake_case 兜底（万一上游写 snake 键）。 + snake = dict(base) + snake.pop("cache_hit") + snake["cache_hit"] = {"prompt_cache_hit_tokens": 6000, "source": "minimax"} + t4 = extract(snake) + _check("cacheHit snake 内键兜底", t4.get("cacheHit", {}).get("promptCacheHitTokens") == 6000) + + # 空/None 兜底（与原有口径一致）。 + _check("空 result→None", extract(None) is None and extract({}) is None) + + +def test_safe_int_no_throw(safe_int): + """P0-2：RecordingChatModel._safe_int 对脏值/None/负数绝不抛——usage 解析不得中断生成主链。""" + print("\n=== P0-2 _safe_int：脏 usage 不抛（异常按 0） ===") + _check("合法整数→原值", safe_int(123) == 123) + _check("合法字符串数→int", safe_int("456") == 456) + _check("None→0", safe_int(None) == 0) + _check("负数→0（不致负 quota）", safe_int(-5) == 0) + _check("脏字符串→0（不抛）", safe_int("abc") == 0) + _check("空串→0（不抛）", safe_int("") == 0) + _check("dict 脏值→0（不抛 TypeError）", safe_int({"x": 1}) == 0) + _check("list 脏值→0（不抛）", safe_int([1, 2]) == 0) + _check("float→截 int", safe_int(3.9) == 3) + + +def test_classify_cache_source(classify): + """P2-1：_classify_cache_source 据命中模型名归类，独立于 pricing 是否可达。""" + print("\n=== P2-1 _classify_cache_source：source 独立于 pricing ===") + # 无命中（cached=0）→ none。 + _check("无命中→none", classify({"deepseek-v4-flash": [100, 50, 0]}) == "none") + # deepseek 命中 → deepseek。 + _check("deepseek 命中→deepseek", classify({"deepseek-v4-flash": [100, 50, 30]}) == "deepseek") + # minimax 命中 → minimax（含 minimax / m3 / m2 前缀）。 + _check("MiniMax-M3 命中→minimax", classify({"MiniMax-M3": [100, 50, 30]}) == "minimax") + _check("M2.7 前缀命中→minimax", classify({"m2.7-xyz": [100, 50, 30]}) == "minimax") + # 旧 2 元素容错（无 cached）→ none，不抛。 + _check("旧 2 元素容错→none", classify({"deepseek-v4-flash": [100, 50]}) == "none") + # 空 per_model → none。 + _check("空 per_model→none", classify({}) == "none" and classify(None) == "none") + + +def main(): + print("U4/B9 缓存命中落成本台账 + U3 trace 扩展键 自动化用例") + funcs = _load_funcs_from(ROOT / "worker" / "cost.py", ["cache_hit_tokens", "compute"]) + extract = _load_funcs_from(ROOT / "worker" / "service.py", ["_extract_trace"])["_extract_trace"] + # P0-2：从 config.py 类内静态方法 AST 提取 _safe_int（零外部依赖，仅 builtins）。 + safe_int = _load_funcs_from(ROOT / "worker" / "agent_loop" / "config.py", ["_safe_int"])["_safe_int"] + # P2-1：从 studio.py 顶层提取 _classify_cache_source（零外部依赖，纯 dict/str 操作）。 + classify = _load_funcs_from(ROOT / "worker" / "agent_loop" / "studio.py", ["_classify_cache_source"])["_classify_cache_source"] + test_cache_hit_tokens(funcs["cache_hit_tokens"]) + test_compute_cache_discount(funcs["compute"]) + test_extract_trace_compat(extract) + test_safe_int_no_throw(safe_int) + test_classify_cache_source(classify) + print(f"\n=== 汇总：通过 {_passed} / 失败 {_failed} ===") + sys.exit(0 if _failed == 0 else 1) + + +if __name__ == "__main__": + main() diff --git a/wg1/gen-worker/worker/_client.py b/wg1/gen-worker/worker/_client.py index 8af58f4e..b67686ef 100644 --- a/wg1/gen-worker/worker/_client.py +++ b/wg1/gen-worker/worker/_client.py @@ -57,11 +57,35 @@ def get_client(): return openai.OpenAI(api_key=get_api_key(), base_url=BASE_URL, max_retries=2, timeout=180.0) +def _cache_hit_from_usage(usage): + """从 openai 原始 usage 抽命中缓存的 prompt token（兼容两套字段取 max，U4/B9）。 + + DeepSeek `usage.prompt_cache_hit_tokens` / MiniMax `usage.prompt_tokens_details.cached_tokens`，取 max； + 字段全缺/异常 → 0（按 miss 计，不报错）。逻辑与 cost.cache_hit_tokens 同口径，但本处只依赖 openai 原始对象、 + 不反向 import cost（避免 _client→cost 循环依赖；cost 已 import _client）。 + """ + if usage is None: + return None + try: + cands = [] + ds = getattr(usage, "prompt_cache_hit_tokens", None) + if ds is not None: + cands.append(int(ds)) + details = getattr(usage, "prompt_tokens_details", None) + cached = getattr(details, "cached_tokens", None) if details is not None else None + if cached is not None: + cands.append(int(cached)) + return max(cands) if cands else 0 + except Exception: + return 0 + + def chat(model, system, user, max_tokens=16000, temperature=0.0, tries=3, retry_delay=2.0): """单次裸调用 chat.completions。确定性产出 temperature=0；瞬时网关错误(502 burst)重试。 - 返回 dict：{content, prompt_tokens, completion_tokens, total_tokens, model, wall_s, id, finish_reason}。 - usage 字段用 openai 2.41.1 口径：prompt_tokens / completion_tokens（供成本折算与 token 取证）。 + 返回 dict：{content, prompt_tokens, completion_tokens, prompt_cache_hit_tokens, total_tokens, model, wall_s, id, finish_reason}。 + usage 字段用 openai 2.41.1 口径：prompt_tokens / completion_tokens（供成本折算与 token 取证）； + prompt_cache_hit_tokens（U4/B9 新增，additive 键）= 命中缓存的 prompt token（两套字段取 max；缺则 0），供 cost.py 折缓存价。 """ client = get_client() last_err = None @@ -86,6 +110,8 @@ def chat(model, system, user, max_tokens=16000, temperature=0.0, tries=3, retry_ "finish_reason": getattr(choice, "finish_reason", None), "prompt_tokens": getattr(usage, "prompt_tokens", None), "completion_tokens": getattr(usage, "completion_tokens", None), + # U4/B9：命中缓存的 prompt token（additive 键；旧 dict 消费方不受影响，新口径供 cost 折缓存价）。 + "prompt_cache_hit_tokens": _cache_hit_from_usage(usage), "total_tokens": getattr(usage, "total_tokens", None), "model": model, "wall_s": round(wall, 3), diff --git a/wg1/gen-worker/worker/agent_loop/calibrate.py b/wg1/gen-worker/worker/agent_loop/calibrate.py index 2090b335..a111a761 100644 --- a/wg1/gen-worker/worker/agent_loop/calibrate.py +++ b/wg1/gen-worker/worker/agent_loop/calibrate.py @@ -86,16 +86,17 @@ def run_calibration(labels_path, stage): brief = _load_brief(s["brief_ref"]) verdict = _load_verdict(gid) ut = _judge_user_text(brief, verdict) - # 视觉位（吃截图） + # 视觉位（吃截图）。U4/B9 后 _judge_* 返回 (in,out,cached) 三元组——校准只关心 in/out，cached 占位忽略 + # （P2-2：校准成本有意取「全价上界」，不折缓存——校准是离线评估，账走保守上界即可，不必对齐生产逐款精账）。 try: - vj, (vi, vo) = studio._judge_vision(vis_model, PERSONA_VISION, ut, gid) + vj, (vi, vo, _vca) = studio._judge_vision(vis_model, PERSONA_VISION, ut, gid) per_model[vis_model][0] += vi per_model[vis_model][1] += vo except Exception as e: vj = {"error": str(e)[:180]} # 文本位（仅运行数据） try: - tj, (ti, to) = studio._judge_text(txt_model, PERSONA_TEXT, ut) + tj, (ti, to, _tca) = studio._judge_text(txt_model, PERSONA_TEXT, ut) per_model[txt_model][0] += ti per_model[txt_model][1] += to except Exception as e: @@ -146,15 +147,17 @@ def run_calibration(labels_path, stage): } # ── 成本（测而不闸；复用 cost.py 的 new-api quota 口径）── + # P2-2：此处 compute 不传 cached_tokens（默认 0）→ 成本是「全价上界」（不折缓存命中），与生产 run_studio 的 + # 精账（折 cache_ratio）有意不同——校准为离线评估，账走保守上界即可。 try: pricing = cost.fetch_pricing() qpu, usd = cost.fetch_status() rws, total = {}, 0.0 for name, (i, o) in per_model.items(): - c = cost.compute(name, i, o, pricing, qpu, usd) + c = cost.compute(name, i, o, pricing, qpu, usd) # 不传 cached → 全价上界（P2-2） rws[name] = {"in": i, "out": o, "rmb": c["rmb"]} total += c["rmb"] - cost_info = {"by_model": rws, "total_rmb": round(total, 5)} + cost_info = {"by_model": rws, "total_rmb": round(total, 5), "costNote": "fullPriceUpperBound"} except Exception as e: cost_info = {"error": str(e)[:180], "by_model": {k: {"in": v[0], "out": v[1]} for k, v in per_model.items()}} diff --git a/wg1/gen-worker/worker/agent_loop/config.py b/wg1/gen-worker/worker/agent_loop/config.py index cd668a09..9946fbdc 100644 --- a/wg1/gen-worker/worker/agent_loop/config.py +++ b/wg1/gen-worker/worker/agent_loop/config.py @@ -24,21 +24,49 @@ _CFG_PATH = Path(__file__).resolve().parents[1] / "models.yaml" class RecordingChatModel(OpenAIChatModel): - """记录每次调用 usage 的模型子类（per-agent token 取证，供成本/Stage 对比）。""" + """记录每次调用 usage 的模型子类（per-agent token 取证，供成本/Stage 对比）。 + + U4/B9：records 元素由 (in, out) 扩为 (in, out, cached)——第 3 位 = 命中缓存的 prompt token。 + AgentScope 2.0.1 的 ChatUsage 已从 `prompt_tokens_details.cached_tokens` 提取 `cache_input_tokens`（MiniMax 路命中由此兜）； + DeepSeek 的 `prompt_cache_hit_tokens` 在 AgentScope 转 ChatUsage 时被丢弃（AgentScope 只读 cached_tokens）， + 故本处兜底再尝试 getattr(u, "prompt_cache_hit_tokens")，两者取 max（与 cost.cache_hit_tokens 同口径）。 + 向后兼容：旧消费方（usage_sum / studio._agent_reply）只读 r[0]/r[1]，3-tuple 不破。""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - self.records = [] # [(input_tokens, output_tokens)] + self.records = [] # [(input_tokens, output_tokens, cached_tokens)] + + @staticmethod + def _safe_int(v): + """脏值/None/负数 → 0；合法数 → int（绝不抛，P0-2：usage 解析不得中断生成主链）。""" + try: + n = int(v) + return n if n >= 0 else 0 + except (TypeError, ValueError): + return 0 async def __call__(self, messages, **kwargs): resp = await super().__call__(messages, **kwargs) u = resp.get("usage") if hasattr(resp, "get") else getattr(resp, "usage", None) if u is not None: - self.records.append((int(getattr(u, "input_tokens", 0) or 0), - int(getattr(u, "output_tokens", 0) or 0))) + # 命中 token：取 ChatUsage.cache_input_tokens（AgentScope 已从 cached_tokens 提取）与 + # DeepSeek 顶层 prompt_cache_hit_tokens（若 AgentScope 未透传则为 None）的 max；全缺/脏→0（miss 全价）。 + # P0-2：逐字段 _safe_int（脏值/None/负数→0），绝不抛——异常按 miss(0)，与 cost.cache_hit_tokens 同纪律。 + cands = [] + ci = getattr(u, "cache_input_tokens", None) + if ci is not None: + cands.append(self._safe_int(ci)) + ds = getattr(u, "prompt_cache_hit_tokens", None) + if ds is not None: + cands.append(self._safe_int(ds)) + cached = max(cands) if cands else 0 + self.records.append((self._safe_int(getattr(u, "input_tokens", 0)), + self._safe_int(getattr(u, "output_tokens", 0)), + cached)) return resp def usage_sum(self): + # 仅汇总 in/out（向后兼容）；命中 token 由 studio._agent_reply 经 records[2] 单独汇总。 return (sum(r[0] for r in self.records), sum(r[1] for r in self.records)) diff --git a/wg1/gen-worker/worker/agent_loop/studio.py b/wg1/gen-worker/worker/agent_loop/studio.py index 83bf048b..592f060b 100644 --- a/wg1/gen-worker/worker/agent_loop/studio.py +++ b/wg1/gen-worker/worker/agent_loop/studio.py @@ -125,14 +125,41 @@ def _parse_judge(raw): # ── AgentScope agent 单轮调用(生成域:design/code/fix)───────────────── +def _rec_cached(r): + """从一条 RecordingChatModel.records 取命中 token（3-tuple 第 3 位）；旧 2-tuple 容错返 0。""" + return r[2] if len(r) >= 3 else 0 + + +def _classify_cache_source(per_model): + """据 per_model（{model_name: [in,out,cached]}）里「有命中（cached>0）」的模型名归类缓存命中来源。 + + 返回 deepseek|minimax|none（P2-1：独立于 pricing 是否可达——只看命中模型名，不绑 cost.compute 成功路径， + 避免「promptCacheHitTokens>0 而 source=none」自相矛盾）。多模型同时命中时后者覆盖（落 trace.cacheHit.source， + 仅展示用，非精确账）。无任何命中 → none。 + """ + source = "none" + for name, toks in (per_model or {}).items(): + # toks = [in, out, cached]；旧 2 元素容错（无 cached 视为 0）。 + cached = toks[2] if len(toks) >= 3 else 0 + if cached and cached > 0: + lname = (name or "").lower() + if "deepseek" in lname: + source = "deepseek" + elif "minimax" in lname or lname.startswith("m3") or lname.startswith("m2"): + source = "minimax" + return source + + async def _agent_reply(name, system, model, user_text): - """新建一次性 Agent(空 toolkit 单轮)→ reply。返回 (文本, (in_tok,out_tok))。""" + """新建一次性 Agent(空 toolkit 单轮)→ reply。返回 (文本, (in_tok,out_tok,cached_tok))。 + + U4/B9：第 3 位 cached = 本轮命中缓存的 prompt token（汇总自 model.records[2]，RecordingChatModel 已捕获）。""" before = len(getattr(model, "records", []) or []) agent = Agent(name=name, system_prompt=system, model=model, react_config=ReActConfig(max_iters=1)) resp = await agent.reply(user_msg(user_text)) recs = (getattr(model, "records", []) or [])[before:] - return text_of(resp), (sum(r[0] for r in recs), sum(r[1] for r in recs)) + return text_of(resp), (sum(r[0] for r in recs), sum(r[1] for r in recs), sum(_rec_cached(r) for r in recs)) # ── player panel(裸客户端:M3 视觉位 + flash 文本位)────────────────── @@ -147,7 +174,9 @@ def _play_summary(verdict): def _judge_vision(model_name, persona, user_text, game_id): - """M3 多模态:文本 + 首帧/玩后截图 → 评判。返回 (判定dict, (in,out))。""" + """M3 多模态:文本 + 首帧/玩后截图 → 评判。返回 (判定dict, (in,out,cached))。 + + U4/B9：第 3 位 cached = 命中缓存的 prompt token（cost.cache_hit_tokens 兼容两套字段取 max）。""" client = _client.get_client() content = [{"type": "text", "text": user_text}] ev = GEN_DIR / game_id / "evidence" @@ -160,13 +189,16 @@ def _judge_vision(model_name, persona, user_text, game_id): {"role": "user", "content": content}] r = client.chat.completions.create(model=model_name, messages=msgs, max_tokens=900, temperature=0.3, stream=False) u = r.usage - return _parse_judge(r.choices[0].message.content or ""), (u.prompt_tokens, u.completion_tokens) + return _parse_judge(r.choices[0].message.content or ""), (u.prompt_tokens, u.completion_tokens, cost.cache_hit_tokens(u)) def _judge_text(model_name, persona, user_text): - """文本位:仅运行数据 → 评判(_client.chat;reasoning 档给足 max_tokens)。""" + """文本位:仅运行数据 → 评判(_client.chat;reasoning 档给足 max_tokens)。返回 (判定dict, (in,out,cached))。 + + U4/B9：cached 取 _client.chat 返回的 prompt_cache_hit_tokens（已两套字段取 max）。""" r = _client.chat(model_name, roles.player_system(persona), user_text, max_tokens=4000, temperature=0.3) - return _parse_judge(r["content"]), (r.get("prompt_tokens") or 0, r.get("completion_tokens") or 0) + return _parse_judge(r["content"]), ( + r.get("prompt_tokens") or 0, r.get("completion_tokens") or 0, r.get("prompt_cache_hit_tokens") or 0) def run_player_panel(game_id, brief, verdict, stage): @@ -177,15 +209,15 @@ def run_player_panel(game_id, brief, verdict, stage): vis_model = config.model_name_for("player_vision", stage) txt_model = config.model_name_for("player_text", stage) panel, problems = {}, [] - tbm = defaultdict(lambda: [0, 0]) # model_name -> [in, out] + tbm = defaultdict(lambda: [0, 0, 0]) # model_name -> [in, out, cached]（U4/B9：第 3 位命中 token） for tag, fn, mname in ( ("vision", lambda: _judge_vision(vis_model, "急性子休闲玩家,凭第一眼观感和反馈下判断", base, game_id), vis_model), ("text", lambda: _judge_text(txt_model, "细致的完整度审查者,只信运行数据、不脑补", base), txt_model), ): try: - verdict_obj, (i, o) = fn() - tbm[mname][0] += i; tbm[mname][1] += o + verdict_obj, (i, o, ca) = fn() + tbm[mname][0] += i; tbm[mname][1] += o; tbm[mname][2] += ca panel[tag] = {"model": mname, **verdict_obj} if str(verdict_obj.get("verdict")) == "fix": problems += [str(p) for p in (verdict_obj.get("problems") or [])] @@ -209,11 +241,13 @@ async def run_studio(game_id, brief, play_spec, stage="stage1", max_repairs=5, attempts = [] tin, tout = 0, 0 - per_model = defaultdict(lambda: [0, 0]) # model_name -> [in,out],per-model 成本取证 + tcached = 0 # U4/B9：全程命中缓存的 prompt token 累计（落 trace.cacheHit.promptCacheHitTokens） + per_model = defaultdict(lambda: [0, 0, 0]) # model_name -> [in,out,cached]，per-model 成本取证（含命中 token） # ① 设计 - design_text, (di, do) = await _agent_reply("design", roles.DESIGN_SYSTEM, design_model, brief) - tin += di; tout += do; per_model[design_name][0] += di; per_model[design_name][1] += do + design_text, (di, do, dca) = await _agent_reply("design", roles.DESIGN_SYSTEM, design_model, brief) + tin += di; tout += do; tcached += dca + per_model[design_name][0] += di; per_model[design_name][1] += do; per_model[design_name][2] += dca enriched = brief + "\n\n## 设计稿（设计 agent 产出）\n" + design_text # gate-H 推广：设计 agent 自产 gate-spec → 合入 play_spec（覆盖 brief 默认），使确定性门适用任意游戏（不再手写 brief）。 gatespec = _extract_gatespec(design_text) @@ -235,8 +269,9 @@ async def run_studio(game_id, brief, play_spec, stage="stage1", max_repairs=5, mname = code_name if role == "code" else fix_name system, user = prompt.build_messages(enriched, retry_feedback=feedback) rec = {"attempt": attempt, "role": role, "model": mname, "stage_fail": None} - text, (ci, co) = await _agent_reply(role, system, model, user) - tin += ci; tout += co; per_model[mname][0] += ci; per_model[mname][1] += co + text, (ci, co, cca) = await _agent_reply(role, system, model, user) + tin += ci; tout += co; tcached += cca + per_model[mname][0] += ci; per_model[mname][1] += co; per_model[mname][2] += cca rec["usage"] = {"in": ci, "out": co} src = validate.extract_code(text) @@ -268,8 +303,12 @@ async def run_studio(game_id, brief, play_spec, stage="stage1", max_repairs=5, # 七门过 → player panel(顾问;七门才是硬地板) passed = True player_result = run_player_panel(game_id, brief, verdict, stage) - for mn, (pi, po) in player_result["tokens_by_model"].items(): - tin += pi; tout += po; per_model[mn][0] += pi; per_model[mn][1] += po + for mn, toks in player_result["tokens_by_model"].items(): + # toks = [in, out, cached]（U4/B9 起含命中 token；旧 2 元素容错补 0）。 + pi, po = toks[0], toks[1] + pca = toks[2] if len(toks) >= 3 else 0 + tin += pi; tout += po; tcached += pca + per_model[mn][0] += pi; per_model[mn][1] += po; per_model[mn][2] += pca rec["player"] = player_result["panel"] attempts.append(rec) probs = player_result["problems"] @@ -281,19 +320,28 @@ async def run_studio(game_id, brief, play_spec, stage="stage1", max_repairs=5, break # per-model ¥(测而不闸;复用 cost.py 的 new-api quota 口径,精确按各模型拆账) + # U4/B9：传 per_model 第 3 位 cached → cost.compute 按 cache_ratio 折真实 quota（命中越多越省）; + # /api/pricing 不可达 → except 走「tokens-only 降级」（只 by_model token + costFallback=true 标记，无 total_rmb，P1-1） + # + 不阻断生成（best-effort）。 cost_info = None + # P2-1：命中来源归类「独立于 pricing 是否可达」——据 per_model 命中（cached>0）的模型名定 source， + # 不绑 cost.compute 成功路径。否则 pricing 失败 + tcached>0 时会出现 promptCacheHitTokens>0 而 source="none" 的自相矛盾。 + cache_source = _classify_cache_source(per_model) try: pricing = cost.fetch_pricing() qpu, usd = cost.fetch_status() rows, total = {}, 0.0 - for name, (i, o) in per_model.items(): - c = cost.compute(name, i, o, pricing, qpu, usd) - rows[name] = {"in": i, "out": o, "rmb": c["rmb"]} + for name, (i, o, ca) in per_model.items(): + c = cost.compute(name, i, o, pricing, qpu, usd, cached_tokens=ca) + rows[name] = {"in": i, "out": o, "cached": ca, "rmb": c["rmb"]} total += c["rmb"] cost_info = {"by_model": rows, "total_rmb": round(total, 5), "gate_rmb": cost.GATE_RMB} except Exception as e: - cost_info = {"error": str(e)[:200], - "by_model": {k: {"in": v[0], "out": v[1]} for k, v in per_model.items()}} + # P1-1 取价失败降级 = 诚实的「tokens-only 降级」：只剩 token（含 cached）+ error + costFallback=true 标记； + # 不折¥（无 pricing 无从算 total_rmb）、不阻断（§5.8 error 路，best-effort）。costFallback 标记让 + # _extract_trace / 审核台明确「本条 cost 是降级口径、无 total_rmb」，避免误读成「成本为 0」。 + cost_info = {"error": str(e)[:200], "costFallback": True, + "by_model": {k: {"in": v[0], "out": v[1], "cached": v[2]} for k, v in per_model.items()}} # ── D9 反同质化（W-G1 组B）：归一 brief/theme → 算签名 → 查 FS 历史登记表 → similarity 子段（随 trace 落库）── # 总开关 AIGC_DEDUP_ENABLED（env，默认关）：关 = 不查重不告警不写 similarity（§10 回滚，现行行为）。 @@ -320,6 +368,9 @@ async def run_studio(game_id, brief, play_spec, stage="stage1", max_repairs=5, "tokens": {"prompt": tin, "completion": tout}, "tokens_by_model": {k: v for k, v in per_model.items()}, "cost": cost_info, + # U4/B9：缓存命中台账（成本台账，仅观测不阻断）——命中 token 全程累计 + 来源归类； + # _extract_trace 据此落 trace.cacheHit{promptCacheHitTokens, source}。 + "cache_hit": {"promptCacheHitTokens": tcached, "source": cache_source}, "wall_s": round(time.perf_counter() - t0, 1), "models": {r: config.model_name_for(r, stage) for r in ("design", "code", "fix", "player_vision", "player_text")}, diff --git a/wg1/gen-worker/worker/cost.py b/wg1/gen-worker/worker/cost.py index 50fc1a8e..5f587c10 100644 --- a/wg1/gen-worker/worker/cost.py +++ b/wg1/gen-worker/worker/cost.py @@ -2,13 +2,17 @@ spike 无 usage→¥ 代码（worker-reusables 地图明示），故 L1 新建。 取证口径： - - 网关 /api/pricing 给每模型 model_ratio / completion_ratio；/api/status 给 quota_per_unit=500000、usd_exchange_rate=7.3。 - - new-api quota 公式：quota = (prompt_tokens + completion_tokens × completion_ratio) × model_ratio × group_ratio + - 网关 /api/pricing 给每模型 model_ratio / completion_ratio / cache_ratio；/api/status 给 quota_per_unit=500000、usd_exchange_rate=7.3。 + - new-api quota 公式（含缓存折扣，U4/B9）： + quota = ((prompt_tokens − cached) + cached × cache_ratio + completion_tokens × completion_ratio) × model_ratio × group_ratio + 其中 cached = 命中缓存的 prompt token 数（按 cache_ratio 折扣，非全价）；cache_ratio 缺/None → 视为 1.0（不折扣，回退全价）。 - USD = quota / quota_per_unit；¥ = USD × usd_exchange_rate。 【口径说明】 - group_ratio 取默认组 1.0（/api/status 未暴露分组倍率，default 组常为 1.0；若非 1 须按实际乘）。 - - 本折算【不计 prompt 缓存折扣】(cache_ratio)，因 worker 仅记 prompt_tokens/completion_tokens 不记 cache 命中数 - → 得【¥上界】(实际 logs.quota ≤ 此值)。精确 logs.quota 需网关 access token 读 /api/log/self（sk- key 无管理权）。 + - 【U4/B9 缓存折扣已计入】(2026-06-17)：worker 现于 usage 提取处捕获命中 token（兼容两套字段取 max，见 + `cache_hit_tokens()`），`compute()` 按 cache_ratio 折真实 quota → 折后 ¥（命中越多越省）。cached=0（未命中/字段缺） + 时退化为旧「¥上界」口径（与扩前字节兼容）。精确 logs.quota 仍需网关 access token 读 /api/log/self（sk- key 无管理权）， + 本折算据 usage + pricing 自算、与网关结算同公式（缓存折扣口径一致）。 - reasoning 模型(deepseek-v4-flash 等)的 reasoning_tokens 已计入 completion_tokens，本折算已含其成本。 """ @@ -51,17 +55,77 @@ def fetch_status(): return d.get("quota_per_unit", 500000), d.get("usd_exchange_rate", 7.3) -def compute(model, prompt_tokens, completion_tokens, pricing, qpu, usd_rate, group_ratio=1.0): - """单条 usage → quota/USD/¥（上界，不计缓存折扣）。""" +def cache_hit_tokens(usage): + """从一条 usage 抽「命中缓存的 prompt token 数」，兼容两套字段取 max（U4/B9，执行版 §5.8）。 + + - DeepSeek 口径：`usage.prompt_cache_hit_tokens`（顶层）。 + - MiniMax 口径：`usage.prompt_tokens_details.cached_tokens`（嵌套）。 + 两套同时存在则取 max（不同厂商命名不同，取大者保守计折扣，与网关结算口径对齐）。 + + :param usage: 一条调用的 usage——可为 + ① openai 原始 usage 对象（resp.usage，含上述属性 / prompt_tokens_details 子对象）； + ② dict（如 _client.chat 返回值或已序列化的 usage）； + ③ AgentScope ChatUsage（含 cache_input_tokens 字段，已由 AgentScope 从 cached_tokens 提取）。 + :return: 命中 token 数（int ≥ 0）；字段全缺/解析异常 → 0（按 miss 全价计，不报错——best-effort 不阻断）。 + """ + if usage is None: + return 0 + try: + def _get(obj, name): + # 兼容对象属性 / dict 键两种取值；缺则 None。 + if obj is None: + return None + if isinstance(obj, dict): + return obj.get(name) + return getattr(obj, name, None) + + candidates = [] + # ① DeepSeek 顶层 prompt_cache_hit_tokens。 + ds = _get(usage, "prompt_cache_hit_tokens") + if ds is not None: + candidates.append(int(ds)) + # ② MiniMax 嵌套 prompt_tokens_details.cached_tokens。 + details = _get(usage, "prompt_tokens_details") + cached = _get(details, "cached_tokens") + if cached is not None: + candidates.append(int(cached)) + # ③ AgentScope ChatUsage.cache_input_tokens（已从 cached_tokens 提取的归一字段；MiniMax 路命中由此兜）。 + ai = _get(usage, "cache_input_tokens") + if ai is not None: + candidates.append(int(ai)) + return max(candidates) if candidates else 0 + except Exception: + # best-effort：任何解析异常一律按 miss（0）计，绝不阻断成本折算/生成主链。 + return 0 + + +def compute(model, prompt_tokens, completion_tokens, pricing, qpu, usd_rate, group_ratio=1.0, cached_tokens=0): + """单条 usage → quota/USD/¥（U4/B9：按 cache_ratio 折缓存命中）。 + + 缓存折扣口径（与 new-api 网关结算同公式）：命中的 cached 个 prompt token 按 cache_ratio 折价，未命中部分全价： + quota = ((prompt_tokens − cached) + cached × cache_ratio + completion_tokens × completion_ratio) × model_ratio × group_ratio + - cache_ratio 缺/None（如 MiniMax-M2.7 未配）→ 视为 1.0（不折扣，回退全价，不报错）。 + - cached 截断到 [0, prompt_tokens]（防脏 usage 命中数超 prompt 数导致负 quota）。 + - cached=0（默认/未命中）→ 退化为旧口径 (prompt + completion×completion_ratio)×model_ratio，与扩前字节兼容。 + """ m = pricing.get(model, {}) model_ratio = m.get("model_ratio", 0) completion_ratio = m.get("completion_ratio", 1) - quota = (prompt_tokens + completion_tokens * completion_ratio) * model_ratio * group_ratio + # cache_ratio：缺/None → 1.0（不折扣）。命中 token 截断到 [0, prompt_tokens]（防脏数据负 quota）。 + cache_ratio = m.get("cache_ratio") + if cache_ratio is None: + cache_ratio = 1.0 + cached = max(0, min(int(cached_tokens or 0), int(prompt_tokens or 0))) + uncached = prompt_tokens - cached + # 缓存折扣量化的 prompt 段 = 未命中全价 + 命中按 cache_ratio 折价。 + prompt_quota = uncached + cached * cache_ratio + quota = (prompt_quota + completion_tokens * completion_ratio) * model_ratio * group_ratio usd = quota / qpu rmb = usd * usd_rate return { "model": model, "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, - "model_ratio": model_ratio, "completion_ratio": completion_ratio, + "cached_tokens": cached, # 实际计入折扣的命中 token（截断后） + "model_ratio": model_ratio, "completion_ratio": completion_ratio, "cache_ratio": cache_ratio, "quota": round(quota, 2), "usd": round(usd, 6), "rmb": round(rmb, 5), } diff --git a/wg1/gen-worker/worker/service.py b/wg1/gen-worker/worker/service.py index 2712a68c..6245f5e5 100644 --- a/wg1/gen-worker/worker/service.py +++ b/wg1/gen-worker/worker/service.py @@ -99,6 +99,11 @@ def _extract_trace(result): 字段全可选（worker 路径未定，按必填子集统计完整率，§5.3）： - 必填子集（run_studio 当前路径无条件产出 = 完整率分母）：pass/repairs/wallS/models/attempts/gameId/stage。 - 可选子集（路径相关/可能缺）：gatespec/sevenGateVerdict/cost/player/tokens/similarity。 + - U3/B5 扩展键（additive，缺/脏则省略键，与 SAA 路 extractTraceQuietly 同口径，字节兼容；P1-2 严格形态校验）： + modelTier（终态模型档 stage1|stage2，须非空串且仅「真升档态」=stage2 或 escalationEvents 非空才落，P0-1）/ + escalationEvents（升档事件数组，须非空 list 才落）/ giveupDumpPath（放弃 dump 路径，须非空串才落）/ + cacheHit（{promptCacheHitTokens, source}，U4/B9 缓存命中台账）。worker 现走 Python 循环（无 SAA 救场阶梯）， + 前三键暂不产出 → 省略（与 SAA 路同读 ReadinessScorer 不破）；cacheHit 由 run_studio 的 cache_hit 落值。 :param result: run_studio 返回的 result dict（嵌套结构，非扁平 verdict，见 studio.py:297-311） :return: 9d trace dict（camelCase 键）；result 为空时返回 None（不塞 trace） @@ -160,6 +165,9 @@ def _extract_trace(result): cost_out["gateRmb"] = cost.get("gate_rmb") if "error" in cost: cost_out["error"] = cost.get("error") + # P1-1：tokens-only 降级标记（pricing 不可达时为 True）——随 trace 落库，审核台据此知本条 cost 无 total_rmb、非 0 成本。 + if cost.get("costFallback"): + cost_out["costFallback"] = True trace["cost"] = cost_out player = result.get("player") @@ -175,6 +183,38 @@ def _extract_trace(result): # D9 相似度（dupHit/titleNorm/sig/dupWith）随 trace 落库（无独立列、无独立面板，§1.2 防越界） trace["similarity"] = similarity + # ===== U3/B5 救场阶梯扩展键（additive，与 SAA 路 extractTraceQuietly 同口径；缺则省略键，字节兼容）===== + # 严格形态校验，镜像 Java putEscalationKeysIfPresent（P1-2 口径对齐）：脏值/空值只省略键、绝不落库。 + # escalationEvents：升档事件数组 [{tierBefore,tierAfter,atFailCount,ts}]。须为「非空 list」才落； + # 非数组/空 list/缺 → 省略键（与 Java events.isArray() && size>0 同口径）。先算它，因 modelTier 是否落要据它。 + escalation = (result.get("escalationEvents") + if result.get("escalationEvents") is not None else result.get("escalation_events")) + has_events = isinstance(escalation, list) and len(escalation) > 0 + if has_events: + trace["escalationEvents"] = escalation + # modelTier：终态模型档（stage1|stage2）。须为「非空字符串」（镜像 Java filter(!isBlank)），且仅「真升档态」才落 + # ——modelTier=="stage2" 或 escalationEvents 非空（P0-1：默认 stage1 是 render 初值噪声，落它破字节兼容）。 + # worker 现走 Python 循环无救场阶梯 → 一般缺 → 省略。 + model_tier = result.get("modelTier") if result.get("modelTier") is not None else result.get("model_tier") + if isinstance(model_tier, str) and model_tier.strip(): + if model_tier == "stage2" or has_events: + trace["modelTier"] = model_tier + # giveupDumpPath：放弃前完整 dump 落盘路径（U5 产）。须为「非空字符串」（镜像 Java filter(!isBlank)）；现可能缺 → 省略。 + giveup = (result.get("giveupDumpPath") + if result.get("giveupDumpPath") is not None else result.get("giveup_dump_path")) + if isinstance(giveup, str) and giveup.strip(): + trace["giveupDumpPath"] = giveup + + # ===== U4/B9 缓存命中台账（{promptCacheHitTokens, source}）===== + # run_studio 产 result.cache_hit（已 camelCase 内键 promptCacheHitTokens + source）；缺则省略键（additive）。 + cache_hit = result.get("cacheHit") if result.get("cacheHit") is not None else result.get("cache_hit") + if isinstance(cache_hit, dict): + trace["cacheHit"] = { + "promptCacheHitTokens": cache_hit.get("promptCacheHitTokens", + cache_hit.get("prompt_cache_hit_tokens", 0)) or 0, + "source": cache_hit.get("source", "none"), + } + return trace