From 688f83fdc5fa540b1690518bfe9ff533dd799c1f Mon Sep 17 00:00:00 2001 From: lili Date: Sat, 4 Jul 2026 00:37:16 -0700 Subject: [PATCH] =?UTF-8?q?feat(tier2):=20=E6=94=B6=E6=95=9B=E7=8E=AF?= =?UTF-8?q?=E5=8F=B0=E8=B4=A6=20infra=20=E5=BD=92=E5=9B=A0=E7=A9=BF?= =?UTF-8?q?=E7=BA=BF(observe-only=20=E7=AC=AC=E4=B8=80=E8=BF=AD=E4=BB=A3)?= =?UTF-8?q?=E2=80=94=E2=80=94=E6=B2=BB=20F-2=20R1=20=E8=AE=BE=E8=AE=A1?= =?UTF-8?q?=E5=9B=A2=E9=98=9F=E8=B6=85=E6=97=B6=E9=99=8D=E7=BA=A7=E8=A2=AB?= =?UTF-8?q?=E8=AF=AF=E5=88=A4"=E5=88=86=E6=95=A3"=E7=9A=84=E7=94=9F?= =?UTF-8?q?=E4=BA=A7=E7=9B=B2=E5=8C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把「设计团队超时降级 / writer 撞轮数墙 / step_cap 熔断 / ¥ 软预算软停 / 墙钟超时」这类 编排层事件从 studio 运行态穿到批跑台账,让退路树能一眼归因。此前两败局其实 100% 集中在 「设计团队超时降级路」,但台账 fail_system 空 → decide_n5 按 fail_system 看"分散"出 R3_conditional,主持人只能人工翻日志归因。 做法(不塞 fail_system,游戏系统桶语义不污染;decide_n5 的 R1/R2 分流逻辑一字不动): - RunRecord 新增 infra_flags: list[str](field(default_factory=list);受控值集 INFRA_FLAGS 五项常量 design_team_degraded/writer_iter_wall/step_cap_tripped/soft_budget_tripped/ wall_timeout,注释写明可扩)+ 纯映射 infra_flags_from_runtime(信号→值集,去重稳定序); - studio 运行态采集:设计团队降级点(捕获异常,.kind=='timeout' 归 wall_timeout)、外层 resume 轮数墙、熔断返回 kind、¥ 软预算软停标记,经 _infra_flags_from_runtime 归一进 result; design_team.DesignTeamError 加 kind 区分墙钟超时; - batch_run.result_to_record 从 result 抽 infra_flags 落台账(防御式复制); - decide_n5 只把各轮 infra_flags 分布打进 reasons(如「r1 infra 分布={design_team_degraded:2, wall_timeout:2}」),不分支、不改任何出口。 向后兼容:旧 jsonl 行无 infra_flags → from_jsonl_line 读空列表不炸;r{N} 轮分组正则不动; middleware.py 不动;退路树出口枚举与判定不动;decide_n5 签名不破坏。 验收:新增单测 5(纯映射/studio fake 运行态采集/batch 落账/旧行兼容/decide_n5 分布行+出口不变 含 --no-strong-baseline 语义原样);全套 102 passed(97 基线 + 5);run_record 与 fallback_tree 的 __main__ 自检、decide_n5_from_jsonl 新旧混合行端到端均通过。 Co-Authored-By: Claude Fable 5 --- tier2/gen-worker/batch_run.py | 3 + tier2/gen-worker/tests/test_infra_flags.py | 245 ++++++++++++++++++ .../worker/agent_loop/design_team.py | 10 +- tier2/gen-worker/worker/agent_loop/studio.py | 67 ++++- tier2/gen-worker/worker/fallback_tree.py | 24 ++ tier2/gen-worker/worker/run_record.py | 73 ++++++ 6 files changed, 412 insertions(+), 10 deletions(-) create mode 100644 tier2/gen-worker/tests/test_infra_flags.py diff --git a/tier2/gen-worker/batch_run.py b/tier2/gen-worker/batch_run.py index 62bdd34f..521cb2a8 100644 --- a/tier2/gen-worker/batch_run.py +++ b/tier2/gen-worker/batch_run.py @@ -256,6 +256,9 @@ def result_to_record(result: dict, *, run_id: str, brief_variant: str, decision=decision, # type: ignore[arg-type] finished=bool(result.get("finished")), timestamp=timestamp, + # 编排层 infra 归因(observe-only):从 studio result 抽 infra_flags(缺则空列表,向后兼容旧桩)。 + # 防御式 list(...) 复制,避免与 result 内引用共享;取值 ∈ run_record.INFRA_FLAGS 受控值集。 + infra_flags=list(result.get("infra_flags") or []), ) diff --git a/tier2/gen-worker/tests/test_infra_flags.py b/tier2/gen-worker/tests/test_infra_flags.py new file mode 100644 index 00000000..df857b82 --- /dev/null +++ b/tier2/gen-worker/tests/test_infra_flags.py @@ -0,0 +1,245 @@ +"""test_infra_flags.py —— 收敛环台账 infra 归因穿线单测(工单 k,observe-only 第一迭代)。 + +跑:cd tier2/gen-worker && cheap-worker/.venv/bin/python -m pytest tests/test_infra_flags.py -q + +验收门(≥4): + ① studio 态 flag 采集(构造 fake 运行态)—— test_studio_collects_* / test_infra_flags_from_runtime_*; + ② batch_run 落账含 infra_flags —— test_result_to_record_carries_infra_flags; + ③ 旧行(无字段)解析兼容 —— test_from_jsonl_line_backward_compat; + ④ decide_n5 reasons 含分布行且出口不变(含 --no-strong-baseline 语义原样)—— + test_decide_n5_reasons_infra_dist_and_exit_unchanged。 + +穿线红线复核:不塞 fail_system(游戏系统桶语义不污染)、退路树出口枚举与判定不动、台账向后兼容。 +""" +import json + +from worker.run_record import ( + RunRecord, + infra_flags_from_runtime, + INFRA_FLAGS, + INFRA_FLAG_DESIGN_TEAM_DEGRADED, + INFRA_FLAG_WRITER_ITER_WALL, + INFRA_FLAG_STEP_CAP_TRIPPED, + INFRA_FLAG_SOFT_BUDGET_TRIPPED, + INFRA_FLAG_WALL_TIMEOUT, +) +from worker import fallback_tree as ft + + +# ────────────────────────────────────────────────────────────────────────── +# 构造小工具 +# ────────────────────────────────────────────────────────────────────────── +def _mk(ok: bool, *, fail_system=None, infra=None) -> RunRecord: + """造一条 RunRecord:ok=True → 过门款;ok=False → 失败款(可带 fail_system / infra_flags)。""" + if ok: + return RunRecord(run_id="cv-r1-M3-面包-0", model="MiniMax-M3", stage="play", + brief_variant="面包", pass_gate=True, repairs=0, + decision="accept", finished=True, infra_flags=list(infra or [])) + return RunRecord(run_id="cv-r1-M3-面包-0", model="MiniMax-M3", stage="build", + brief_variant="面包", pass_gate=False, repairs=1, + decision="fix", finished=False, fail_system=fail_system, + infra_flags=list(infra or [])) + + +def _round(n_ok: int, n_fail: int, *, fail_system=None, infra=None) -> list: + """造一轮:n_ok 过门 + n_fail 失败(失败款统一带 fail_system / infra_flags)。""" + return ([_mk(True) for _ in range(n_ok)] + + [_mk(False, fail_system=fail_system, infra=infra) for _ in range(n_fail)]) + + +class _FakeTimeoutErr(Exception): + """fake 设计团队超时异常(带 .kind='timeout',镜像 design_team.DesignTeamError 的 kind 约定)。""" + kind = "timeout" + + +class _FakeDegradeErr(Exception): + """fake 设计团队非超时降级异常(无 kind,如预算耗尽 / leader 空产出 / 团队异常)。""" + + +class _FakeBreaker: + """fake CircuitBreakerMiddleware:只暴露 infra 采集读的 budget_soft_tripped 标记。""" + def __init__(self, budget_soft_tripped: bool = False) -> None: + self.budget_soft_tripped = budget_soft_tripped + + +# ────────────────────────────────────────────────────────────────────────── +# ① 纯映射:infra_flags_from_runtime(编排层信号 → 受控值集) +# ────────────────────────────────────────────────────────────────────────── +def test_infra_flags_from_runtime_pure_mapping(): + # 无信号 → 空列表。 + assert infra_flags_from_runtime() == [] + + # 设计团队超时降级 → design_team_degraded + wall_timeout(稳定序=值集登记序)。 + assert infra_flags_from_runtime(design_degraded=True, design_timeout=True) == [ + INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WALL_TIMEOUT] + + # 设计团队非超时降级 → 只 design_team_degraded。 + assert infra_flags_from_runtime(design_degraded=True) == [INFRA_FLAG_DESIGN_TEAM_DEGRADED] + + # 单写撞轮数墙 → writer_iter_wall。 + assert infra_flags_from_runtime(writer_iter_wall=True) == [INFRA_FLAG_WRITER_ITER_WALL] + + # 熔断 kind 映射:step_cap → step_cap_tripped;timeout → wall_timeout。 + assert infra_flags_from_runtime(breaker_kind="step_cap") == [INFRA_FLAG_STEP_CAP_TRIPPED] + assert infra_flags_from_runtime(breaker_kind="timeout") == [INFRA_FLAG_WALL_TIMEOUT] + # budget/stuck 不入首批值集(由 circuit_break 字段另记)→ 空。 + assert infra_flags_from_runtime(breaker_kind="budget") == [] + assert infra_flags_from_runtime(breaker_kind="stuck") == [] + + # ¥ 软预算软停 → soft_budget_tripped。 + assert infra_flags_from_runtime(budget_soft_tripped=True) == [INFRA_FLAG_SOFT_BUDGET_TRIPPED] + + # 去重 + 稳定序:设计团队超时(wall_timeout)叠加熔断 timeout(wall_timeout)→ wall_timeout 只一份; + # 多信号同现按 INFRA_FLAGS 登记序输出(不随触发先后抖动)。 + got = infra_flags_from_runtime( + design_degraded=True, design_timeout=True, writer_iter_wall=True, + breaker_kind="timeout", budget_soft_tripped=True) + assert got == [INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WRITER_ITER_WALL, + INFRA_FLAG_SOFT_BUDGET_TRIPPED, INFRA_FLAG_WALL_TIMEOUT] + assert got.count(INFRA_FLAG_WALL_TIMEOUT) == 1 # 去重坐实 + # 输出只含受控值集里的项。 + assert all(f in INFRA_FLAGS for f in got) + + +# ────────────────────────────────────────────────────────────────────────── +# ① studio 态采集(构造 fake 运行态,委托纯映射) +# ────────────────────────────────────────────────────────────────────────── +def test_studio_collects_infra_flags_from_fake_runtime(): + # 延迟 import(studio 拉 agentscope;与既有 middleware 测试同环境,baseline 已证可 import)。 + from worker.agent_loop import studio + + # 设计团队墙钟超时降级 → design_team_degraded + wall_timeout(studio 从异常 .kind 抽 design_timeout)。 + flags = studio._infra_flags_from_runtime( + design_error=_FakeTimeoutErr("超时"), breaker=_FakeBreaker(), + breaker_tripped=None, hit_writer_iter_wall=False) + assert flags == [INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WALL_TIMEOUT] + + # 非超时降级 → 只 design_team_degraded。 + flags = studio._infra_flags_from_runtime( + design_error=_FakeDegradeErr("预算耗尽"), breaker=_FakeBreaker(), + breaker_tripped=None, hit_writer_iter_wall=False) + assert flags == [INFRA_FLAG_DESIGN_TEAM_DEGRADED] + + # 撞轮数墙 + step_cap 熔断 + 软停:三信号齐 → 稳定序输出。 + flags = studio._infra_flags_from_runtime( + design_error=None, breaker=_FakeBreaker(budget_soft_tripped=True), + breaker_tripped={"kind": "step_cap", "reason": "步数超顶"}, hit_writer_iter_wall=True) + assert flags == [INFRA_FLAG_WRITER_ITER_WALL, INFRA_FLAG_STEP_CAP_TRIPPED, + INFRA_FLAG_SOFT_BUDGET_TRIPPED] + + # 全无编排层事件(正常收敛的 run)→ 空列表。 + flags = studio._infra_flags_from_runtime( + design_error=None, breaker=_FakeBreaker(), + breaker_tripped=None, hit_writer_iter_wall=False) + assert flags == [] + + +# ────────────────────────────────────────────────────────────────────────── +# ② batch_run 落账含 infra_flags(result → RunRecord 接线) +# ────────────────────────────────────────────────────────────────────────── +def test_result_to_record_carries_infra_flags(): + import batch_run + + # 造一个形状对齐 studio result 的假结果:带 infra_flags(设计团队超时降级路)。 + result = { + "model": "MiniMax-M3", "finished": False, + "last_verdict": {"decision": "fix", "layerResults": {"L1": {"passed": False}}}, + "circuit_break": None, "breaker_counters": {"tool_calls": 3, "model_calls": 5}, + "tokens": {"prompt": 100000, "completion": 30000, "cached": 5000}, + "cost": {"cost_rmb": 1.2, "tokens_by_model": {}}, + "wall_s": 90.0, "file_tree": [], "source_project": {}, + "infra_flags": [INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WALL_TIMEOUT], + } + rec = batch_run.result_to_record(result, run_id="feie-r1-M3-面包-0", + brief_variant="面包", timestamp=1750000000.0) + assert rec.infra_flags == [INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WALL_TIMEOUT] + # 防御式复制:改 result 里的列表不应回污染 rec(不共享引用)。 + result["infra_flags"].append("xxx") + assert rec.infra_flags == [INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WALL_TIMEOUT] + # 落盘 JSONL 往返一致(infra_flags 原样保留)。 + back = RunRecord.from_jsonl_line(rec.to_jsonl_line()) + assert back.infra_flags == [INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WALL_TIMEOUT] + + # result 缺 infra_flags 键(旧桩)→ 落空列表,不炸。 + result.pop("infra_flags") + rec2 = batch_run.result_to_record(result, run_id="feie-r1-M3-面包-1", + brief_variant="面包", timestamp=1750000000.0) + assert rec2.infra_flags == [] + + +# ────────────────────────────────────────────────────────────────────────── +# ③ 旧行(无 infra_flags 字段)解析兼容 —— 台账向后兼容红线 +# ────────────────────────────────────────────────────────────────────────── +def test_from_jsonl_line_backward_compat(): + # 旧台账行:没有 infra_flags 列(模拟本工单前落的行)。 + old_line = json.dumps({ + "run_id": "feie-001#1", "model": "MiniMax-M3", "stage": "play", + "brief_variant": "面包店", "pass": True, "repairs": 4, + "cost_yuan": 2.13, "wall_seconds": 187.4, "decision": "accept", "finished": True, + }, ensure_ascii=False) + rec = RunRecord.from_jsonl_line(old_line) # 不炸 + assert rec.infra_flags == [] # 缺字段读为空列表 + assert rec.pass_gate is True and rec.model == "MiniMax-M3" # 其余字段照常 + + # 新行:带 infra_flags,往返一致。 + new_rec = _mk(False, infra=[INFRA_FLAG_STEP_CAP_TRIPPED]) + line = new_rec.to_jsonl_line() + assert '"infra_flags"' in line and "step_cap_tripped" in line + assert RunRecord.from_jsonl_line(line).infra_flags == [INFRA_FLAG_STEP_CAP_TRIPPED] + + # 未知多余列(台账演进新增)仍被过滤,不炸。 + weird = json.loads(old_line) + weird["some_future_col"] = 123 + weird["infra_flags"] = ["design_team_degraded"] + rec3 = RunRecord.from_jsonl_line(json.dumps(weird, ensure_ascii=False)) + assert rec3.infra_flags == ["design_team_degraded"] + + +# ────────────────────────────────────────────────────────────────────────── +# ④ decide_n5 reasons 含分布行 且 出口不变(含 --no-strong-baseline 语义原样) +# ────────────────────────────────────────────────────────────────────────── +def test_decide_n5_reasons_infra_dist_and_exit_unchanged(): + # —— F-2 复现场景:两轮不收敛、失败款 fail_system 全空(退路树按 fail_system 看「分散」)、 + # 但 infra_flags 全集中「设计团队超时降级路」(design_team_degraded + wall_timeout)。 + # 这正是被误判分散出 R3_conditional 的生产盲区:穿线后 reasons 应打印出集中的 infra 分布。—— + infra = [INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WALL_TIMEOUT] + rounds_flagged = [_round(3, 2, fail_system=None, infra=infra), + _round(3, 2, fail_system=None, infra=infra)] + # 同构但不带 infra_flags 的对照(证「加 infra_flags 不改出口」)。 + rounds_plain = [_round(3, 2, fail_system=None), + _round(3, 2, fail_system=None)] + + # 无强档对照(--no-strong-baseline 语义):失败分散无集中 → R3_conditional(出口不变)。 + out_flagged = ft.decide_n5(rounds_flagged, has_strong_baseline=False) + out_plain = ft.decide_n5(rounds_plain, has_strong_baseline=False) + assert out_flagged["exit"] == ft.EXIT_R3_CONDITIONAL # 误判出口原样(本工单不改判定) + assert out_flagged["exit"] == out_plain["exit"] # 加 infra_flags 不改出口 + # reasons 含各轮 infra 分布行,且点名集中的编排层原因(治盲区的核心穿线证据)。 + joined = "\n".join(out_flagged["reasons"]) + assert "infra 分布" in joined + assert "design_team_degraded:2" in joined and "wall_timeout:2" in joined + assert "r1 infra 分布" in joined and "r2 infra 分布" in joined + # 不带 infra_flags 的对照:无分布行(无 flag 不加噪)。 + assert "infra 分布" not in "\n".join(out_plain["reasons"]) + + # 有强档对照:同场景 → R3_天花板(前瞻支);加 infra_flags 亦不改出口。 + assert ft.decide_n5(rounds_flagged, has_strong_baseline=True)["exit"] == ft.EXIT_R3_CEILING + assert (ft.decide_n5(rounds_flagged, has_strong_baseline=True)["exit"] + == ft.decide_n5(rounds_plain, has_strong_baseline=True)["exit"]) + + # —— 收敛(go)出口也不受 infra_flags 影响:软预算软停发生在最终干净轮上,仍判 go,且打印分布。—— + go_flagged = [_round(5, 0), [_mk(True, infra=[INFRA_FLAG_SOFT_BUDGET_TRIPPED]) for _ in range(5)]] + go_plain = [_round(5, 0), _round(5, 0)] + o_go = ft.decide_n5(go_flagged, has_strong_baseline=False) + assert o_go["exit"] == ft.EXIT_GO + assert o_go["exit"] == ft.decide_n5(go_plain, has_strong_baseline=False)["exit"] + assert "soft_budget_tripped:5" in "\n".join(o_go["reasons"]) + + # —— 集中表现层仍走 R1(fail_system 分流不被 infra_flags 干扰):证 fail_system 与 infra_flags 各管一摊。—— + r1_flagged = [_round(1, 4, fail_system="presentation", infra=infra), + _round(1, 4, fail_system="presentation", infra=infra)] + r1_plain = [_round(1, 4, fail_system="presentation"), + _round(1, 4, fail_system="presentation")] + assert ft.decide_n5(r1_flagged, has_strong_baseline=False)["exit"] == ft.EXIT_R1_TEMPLATE + assert (ft.decide_n5(r1_flagged, has_strong_baseline=False)["exit"] + == ft.decide_n5(r1_plain, has_strong_baseline=False)["exit"]) diff --git a/tier2/gen-worker/worker/agent_loop/design_team.py b/tier2/gen-worker/worker/agent_loop/design_team.py index 41a8a934..b9dbef15 100644 --- a/tier2/gen-worker/worker/agent_loop/design_team.py +++ b/tier2/gen-worker/worker/agent_loop/design_team.py @@ -80,8 +80,15 @@ class DesignTeamError(Exception): """工作室设计团队失败信号(模型超时/预算耗尽/leader 未产出有效设计稿)。 studio 的 _design_stage 接住它 → degrade 回落原单 agent 单轮设计,绝不让设计阶段中断主链。 + + kind:失败大类(供 studio 采集编排层 infra_flags 时区分);'timeout' = 整团队墙钟硬超时(降级归 + wall_timeout),其余(预算耗尽 / leader 空产出 / 团队异常)= None(只归 design_team_degraded)。 """ + def __init__(self, message: str, *, kind: str | None = None) -> None: + self.kind = kind + super().__init__(message) + # ── 四面专家定义:(工具名, 工具中文描述, system prompt)────────────────────────── # 工具名/描述会被 AgentScope 从 FunctionTool 抽成 leader 可见的工具 schema(描述即 leader 选工具的依据)。 @@ -257,7 +264,8 @@ async def run_design_team( try: resp = await asyncio.wait_for(leader.reply(_user_msg(kick)), timeout=timeout_s) except asyncio.TimeoutError as e: - raise DesignTeamError(f"工作室设计团队超时(> {timeout_s}s):{e}") from e + # kind='timeout':整团队墙钟硬超时——studio 采集时据此归 infra_flags 的 wall_timeout(+ design_team_degraded)。 + raise DesignTeamError(f"工作室设计团队超时(> {timeout_s}s):{e}", kind="timeout") from e except Exception as e: # noqa: BLE001 —— 任何团队异常都转成 DesignTeamError(让 studio 走 degrade) raise DesignTeamError(f"工作室设计团队异常:{type(e).__name__}: {e}") from e diff --git a/tier2/gen-worker/worker/agent_loop/studio.py b/tier2/gen-worker/worker/agent_loop/studio.py index 6168c340..371275e8 100644 --- a/tier2/gen-worker/worker/agent_loop/studio.py +++ b/tier2/gen-worker/worker/agent_loop/studio.py @@ -33,7 +33,7 @@ from pathlib import Path # 包内/直跑兼容导入(直跑时把 tier2/ 加进 sys.path,使 `gen_worker.*` 可解析)。 try: - from .. import config, roles, run, genconfig + from .. import config, roles, run, genconfig, run_record from ..toolkit import Tier2Session, build_toolkit from ..middleware import CircuitBreakerMiddleware, Tier2CircuitBreak, Tier2TraceMiddleware from . import design_team # 阶段 1 工作室星形多 agent 设计团队 @@ -42,7 +42,7 @@ except ImportError: # pragma: no cover —— 直接 python studio.py 兜底 # 本文件在 tier2/gen-worker/worker/agent_loop/studio.py;把 gen-worker/ 加进 sys.path, # 使顶层包 `worker` 可解析(包名 worker 合法;目录名 gen-worker 含连字符不可直接 import)。 sys.path.insert(0, str(Path(__file__).resolve().parents[2])) - from worker import config, roles, run, genconfig # type: ignore + from worker import config, roles, run, genconfig, run_record # type: ignore from worker.toolkit import Tier2Session, build_toolkit # type: ignore from worker.middleware import ( # type: ignore CircuitBreakerMiddleware, Tier2CircuitBreak, Tier2TraceMiddleware) @@ -109,12 +109,15 @@ async def _design_single_agent(brief: str, model) -> str: return text_of(resp) -async def _design_stage(brief: str, model_factory, *, model_sink: list, use_team: bool = True) -> str: +async def _design_stage(brief: str, model_factory, *, model_sink: list, + use_team: bool = True) -> tuple[str, Exception | None]: """阶段 1:产富游戏设计稿。优先工作室星形多 agent 团队(过门头号杠杆),失败 degrade 回单 agent。 - 产物形状 = str(连贯设计稿),与原单 agent 完全一致 → 阶段 2 单写消费接口(roles.writer_system 的 - design_text)零改、爆炸半径最小化。design_team 造的所有 model(leader + 专家)都进 model_sink, - 供 run_studio 汇总 token 成本(多 agent 烧的 token 必须被成本台账抓到,P0 纪律)。 + 产物形状 = (连贯设计稿 str, 降级异常 or None)。设计稿与原单 agent 完全一致 → 阶段 2 单写消费接口 + (roles.writer_system 的 design_text)零改、爆炸半径最小化。第二个返回值是**编排层 infra 归因**用: + 团队降级时把捕获的异常带回(None=未降级),run_studio 据它(及 .kind=='timeout')采集 infra_flags 的 + design_team_degraded / wall_timeout(治 F-2 R1 生产盲区)。design_team 造的所有 model(leader + 专家) + 都进 model_sink,供 run_studio 汇总 token 成本(多 agent 烧的 token 必须被成本台账抓到,P0 纪律)。 Args: brief: 题面。 @@ -123,20 +126,46 @@ async def _design_stage(brief: str, model_factory, *, model_sink: list, use_team model_sink: 收集本阶段造的所有 model(team 的 + 兜底单 agent 的),供成本汇总。 use_team: True → 先试工作室多 agent 团队(默认);False → 直接走单 agent(留旁路/调试用)。 """ + design_error: Exception | None = None # 团队降级时捕获的异常(供 infra 归因;None=未降级) if use_team: try: # 工作室星形多 agent 设计团队(worker-as-tool;预算约束 + 超限 degrade 在 design_team 内)。 - return await design_team.run_design_team(brief, model_factory, model_sink=model_sink) + return await design_team.run_design_team(brief, model_factory, model_sink=model_sink), None except design_team.DesignTeamError as e: # 团队失败(超时/预算耗尽/leader 无有效产出)→ degrade 回单 agent,绝不中断主链。 + # 捕获异常带回(其 .kind=='timeout' 标记设计团队墙钟超时),供 infra_flags 采集归因。 + design_error = e print(f"[tier2-studio] 阶段 1 工作室多 agent 团队失败,degrade 回单 agent 设计:{e}", flush=True) except Exception as e: # noqa: BLE001 —— 任何意外也 degrade(设计阶段绝不能成为新失败点) + design_error = e print(f"[tier2-studio] 阶段 1 工作室多 agent 团队异常,degrade 回单 agent 设计:" f"{type(e).__name__}: {e}", flush=True) # degrade / 旁路:单 design agent 单轮。它造的 model 也进 sink 供成本汇总。 m = model_factory() model_sink.append(m) - return await _design_single_agent(brief, m) + return await _design_single_agent(brief, m), design_error + + +def _infra_flags_from_runtime(*, design_error: Exception | None, breaker, + breaker_tripped: dict | None, + hit_writer_iter_wall: bool) -> list[str]: + """据 studio 运行态收集编排层 infra_flags(observe-only;委托 run_record 纯映射,绝不碰 fail_system)。 + + 把四类编排层运行态信号抽成原语后交 run_record.infra_flags_from_runtime 归一(去重、稳定序): + - design_error:设计阶段降级时捕获的异常(None=未降级;.kind=='timeout' → 设计团队墙钟超时); + - hit_writer_iter_wall:单写外层 resume 轮数墙是否撞到(预算耗尽仍未收敛); + - breaker_tripped:熔断返回 dict {'kind':...}(None=未熔断;kind ∈ step_cap/budget/stuck/timeout); + - breaker.budget_soft_tripped:¥ 软预算是否软停触发(soft 档越软停线,不断链)。 + + 单独抽成函数便于用「构造 fake 运行态」单测(不必真跑 agentscope ReAct 全链)。 + """ + return run_record.infra_flags_from_runtime( + design_degraded=design_error is not None, + design_timeout=(getattr(design_error, "kind", None) == "timeout"), + writer_iter_wall=bool(hit_writer_iter_wall), + breaker_kind=(breaker_tripped or {}).get("kind") if breaker_tripped else None, + budget_soft_tripped=bool(getattr(breaker, "budget_soft_tripped", False)), + ) def _bypass_state() -> AgentState: @@ -359,11 +388,13 @@ async def run_studio( # 客户端,兜底单 agent 也从它取一个。design_models 收集本阶段造的所有客户端,供下方成本汇总(多 agent # 烧的 token 必须进台账,P0 纪律)。design_text 形状 = str,与原单 agent 一致 → 阶段 2 消费接口零改。 design_text = "" + design_error: Exception | None = None # 设计团队降级时捕获的异常(供 infra_flags 归因;None=未降级) design_models: list = [] # 阶段 1 造的所有模型客户端(team leader + 专家 + 兜底单 agent),供成本汇总 if do_design: design_model_factory = functools.partial( config.build_model, mname, max_tokens=max_tokens, thinking_budget=thinking_budget) - design_text = await _design_stage(brief, design_model_factory, model_sink=design_models) + design_text, design_error = await _design_stage( + brief, design_model_factory, model_sink=design_models) # ── 阶段 2:单写 ReAct(放开 max_iters + 九工具 + 四熔断 + trace 接线 + BYPASS 权限)── session = Tier2Session(game_id, play_spec=play_spec) @@ -407,6 +438,7 @@ async def run_studio( cumulative_iter = 0 breaker_tripped = None + hit_writer_iter_wall = False # 单写外层 resume 轮数墙:预算耗尽仍未收敛(infra_flags 采集用) final_text = "" # ── 外层有界自纠循环(spike feie-001 根因 1)── # AgentScope 2.0.2 原生 ReAct 在「模型产出无 tool_call 的纯文本回合」即退出(_agent.py:612)—— @@ -451,6 +483,8 @@ async def run_studio( kick_text = "验收门已全绿。现在直接调 finish 交付源工程,不要再改。" continue if attempt >= max_resumes: + # 撞外层 resume 轮数墙:预算耗尽仍未收敛 → 置 infra 标记(writer_iter_wall),供退路树归因。 + hit_writer_iter_wall = True print(f"[tier2-studio] game={game_id} resume 预算耗尽({max_resumes} 次)仍未收敛,停。", flush=True) break @@ -604,6 +638,17 @@ async def run_studio( # ── trace 收口(H1/H2):读 tracer 摘要(traceId / 步数 / 丢弃数),供编排器对账反查 ── trace_summary = tracer.summary() + # ── 编排层 infra 归因采集(observe-only;治 F-2 R1 生产盲区)────────────────────────── + # 把本次 run 的四类编排层运行态信号(设计团队降级 / 单写撞轮数墙 / 熔断返回 kind / ¥ 软预算软停) + # 归一成 infra_flags 受控值集,写进 result 供 batch_run 落账、退路树 decide_n5 打印分布归因。 + # **绝不塞 fail_system**(游戏系统桶),与退路树 R1/R2 分流键语义正交;绝不参与 decision。 + infra_flags = _infra_flags_from_runtime( + design_error=design_error, breaker=breaker, + breaker_tripped=breaker_tripped, hit_writer_iter_wall=hit_writer_iter_wall) + if infra_flags: + print(f"[tier2-studio] game={game_id} 编排层 infra_flags={infra_flags}" + "(observe-only,退路树归因用,不改 decision)", flush=True) + result = { "game_id": game_id, "model": mname, @@ -627,6 +672,10 @@ async def run_studio( # 四熔断触发记录(对接 verdict.breakerKind:step_cap/budget/stuck/timeout)。 "circuit_break": breaker_tripped, "breaker_counters": {"tool_calls": breaker.tool_calls, "model_calls": breaker.model_calls}, + # 编排层 infra 归因(observe-only;取值 ∈ run_record.INFRA_FLAGS 受控值集,可空)。 + # RunRecord 接线(batch_run.result_to_record)从此抽 infra_flags 落台账,供退路树 decide_n5 + # 打印各轮分布归因(治 F-2 R1「设计团队超时降级路」被误判分散的生产盲区);绝不参与 decision。 + "infra_flags": infra_flags, # 工作记忆 checkpoint 摘要(U2,A2:跨 resume 累计轮次 + 是否从 checkpoint 续跑)。 # 累计轮次由本编排层维护(2.0.2 state.cur_iter 每 reply 归零);checkpoint 详情落 workdir/.tier2-work-memory.json。 "work_memory": {"cumulativeIter": cumulative_iter, "resumedFromCheckpoint": resume_ckpt is not None}, diff --git a/tier2/gen-worker/worker/fallback_tree.py b/tier2/gen-worker/worker/fallback_tree.py index c819ffdc..8d86e844 100644 --- a/tier2/gen-worker/worker/fallback_tree.py +++ b/tier2/gen-worker/worker/fallback_tree.py @@ -317,6 +317,19 @@ def _fail_system_dist(rounds: list[list[Any]]) -> dict[str, float]: return dist +def _infra_flag_dist(round_records: list[Any]) -> dict[str, int]: + """一轮内各 infra_flag 出现次数(编排层归因;鸭子类型读 rec.infra_flags,decide_n5 保持零依赖)。 + + observe-only:只喂 decide_n5 的 reasons 打印(治 F-2 R1 生产盲区),**绝不参与分流/改出口**。 + 统计的是全款(不限失败款):软预算软停等可能发生在最终过门的款上,也是有价值的编排层归因信号。 + """ + dist: dict[str, int] = {} + for r in round_records: + for f in (getattr(r, "infra_flags", None) or []): + dist[f] = dist.get(f, 0) + 1 + return dist + + def decide_n5(rounds: list[list[Any]], *, has_strong_baseline: bool) -> dict[str, Any]: """n=5 收敛环判定。rounds = 各轮的 RunRecord 列表(有序,Round1 在前)。 @@ -337,6 +350,17 @@ def decide_n5(rounds: list[list[Any]], *, has_strong_baseline: bool) -> dict[str last = fails[-1] reasons.append(f"各轮失败数={fails};收敛判据 KTD1 = 末轮≤1 且 ≥2 轮(压 n=5 首轮裸运气)。") + # ── 编排层 infra 归因(observe-only;治 F-2 R1 生产盲区)──────────────────────────────── + # 把各轮的 infra_flags 分布打进 reasons(如「r1 infra 分布={design_team_degraded:2}」),让主持人/ + # 机器一眼看出「两败局是否集中在设计团队超时降级路」这类编排层原因——过去 fail_system 空时退路树 + # 误判「分散」出 R3_conditional,归因只能人工翻日志。**只打印、不分支、不改任何出口**(纯观测穿线)。 + # 放在收敛/未收敛分流之前 → 无论走哪个出口,reasons 都带上分布行(有 flag 的轮才打,无则不加噪)。 + for _idx, _rd in enumerate(rounds, start=1): + _idist = _infra_flag_dist(_rd) + if _idist: + _shown = ", ".join(f"{k}:{v}" for k, v in _idist.items()) + reasons.append(f"r{_idx} infra 分布={{{_shown}}}(编排层归因,observe-only,不参与分流)。") + # ── 收敛分支:末轮 ≤1 ── if last <= 1: if len(rounds) < 2: diff --git a/tier2/gen-worker/worker/run_record.py b/tier2/gen-worker/worker/run_record.py index 3209ae75..9ad6a54f 100644 --- a/tier2/gen-worker/worker/run_record.py +++ b/tier2/gen-worker/worker/run_record.py @@ -63,6 +63,28 @@ FailSystem = Literal["resource", "merge", "order", "presentation"] # decision:终判三值,取自 verdict.decision(tier2-verdict 契约)。accept/fix/kill。 Decision = Literal["accept", "fix", "kill"] +# infra_flags:编排层失控/降级归因标记(G4 采集字段表之外的加性观测列;**与 fail_system 严格分离**)。 +# 为什么单独一列、绝不塞进 fail_system:fail_system 是「游戏系统桶」(resource/merge/order/presentation), +# 语义是"哪个游戏系统的骨架缺口",退路树 decide_n5 的 R1/R2 分流直接读它;而「设计团队超时降级 / +# writer 撞轮数墙 / step_cap 熔断 / 软预算软停 / 墙钟超时」是**编排层**事件(与具体游戏系统无关),塞进 +# fail_system 会污染系统桶语义、误导 R1/R2 分流。故另立 infra_flags 一列专收编排层归因,治 F-2 R1 +# 「两败局全在设计团队超时降级路,但 fail_system 空 → 退路树误判分散出 R3_conditional」的生产盲区。 +# 本列 observe-only:只进 decide_n5 的 reasons 打印供主持人/机器归因,退路树绝不据它分支、不改出口。 +INFRA_FLAG_DESIGN_TEAM_DEGRADED = "design_team_degraded" # 设计团队(超时/预算/leader 空产出)→ degrade 回单 agent +INFRA_FLAG_WRITER_ITER_WALL = "writer_iter_wall" # 单写外层 resume 轮数墙:预算耗尽仍未收敛 +INFRA_FLAG_STEP_CAP_TRIPPED = "step_cap_tripped" # step_cap 熔断:工具调用步数超硬顶 +INFRA_FLAG_SOFT_BUDGET_TRIPPED = "soft_budget_tripped" # ¥ 软预算软停触发(soft 档越软停线,不断链) +INFRA_FLAG_WALL_TIMEOUT = "wall_timeout" # 墙钟超时(设计团队墙钟超时 / 单写 reply 墙钟或单步静默熔断) +# 受控值集(首批五项;**可扩**——新增编排层失控信号只需在此登记一个常量,studio 采集点与本集同源)。 +INFRA_FLAGS: tuple[str, ...] = ( + INFRA_FLAG_DESIGN_TEAM_DEGRADED, INFRA_FLAG_WRITER_ITER_WALL, + INFRA_FLAG_STEP_CAP_TRIPPED, INFRA_FLAG_SOFT_BUDGET_TRIPPED, INFRA_FLAG_WALL_TIMEOUT, +) +InfraFlag = Literal[ + "design_team_degraded", "writer_iter_wall", "step_cap_tripped", + "soft_budget_tripped", "wall_timeout", +] + # ────────────────────────────────────────────────────────────────────────── # RunRecord —— G4 采集字段表的数据类(每款 run 一条,落 JSONL 台账) @@ -195,6 +217,16 @@ class RunRecord: """agent 对 driver 的编辑次数(仅第二段)。来源:第二段编排观测; 用途:防 agent 用改 driver 逃避卡死探测。""" + # ── 编排层 infra 归因组(加性观测;**与上面 fail_system 游戏系统桶严格分离**,见 INFRA_FLAGS 注释)── + infra_flags: list[str] = field(default_factory=list) + """编排层失控/降级归因标记列表(取值 ∈ INFRA_FLAGS 受控值集,可空、可扩)。来源:studio 运行态在 + 设计团队降级点 / 单写外层 resume 轮数墙 / 熔断返回点 / ¥ 软预算软停点采集(见 agent_loop/studio.py + 的 _infra_flags_from_runtime 收集口),经 batch_run 落账带入;用途:治 F-2 R1 生产盲区——退路树 + decide_n5 把各轮 infra_flags 分布打进 reasons 供主持人/机器归因,但**绝不据它分支或改出口** + (observe-only)。默认空列表(field(default_factory=list) 防可变默认在实例间共享);旧台账行无此列时 + from_jsonl_line 读为空列表(向后兼容,r{N} 轮分组不受影响)。**绝不与 fail_system 混用**:fail_system + 是游戏系统桶(退路树 R1/R2 分流键),本列是编排层事件,语义正交、各管一摊。""" + def to_jsonl_line(self) -> str: """序列化成一行 JSONL(末尾不带换行;append_record 落盘时补 '\\n')。 @@ -225,6 +257,47 @@ class RunRecord: return cls(**filtered) +# ────────────────────────────────────────────────────────────────────────── +# infra_flags_from_runtime —— 编排层运行态信号 → 受控 infra_flags 值集(纯映射,零副作用) +# ────────────────────────────────────────────────────────────────────────── +def infra_flags_from_runtime( + *, + design_degraded: bool = False, + design_timeout: bool = False, + writer_iter_wall: bool = False, + breaker_kind: Optional[str] = None, + budget_soft_tripped: bool = False, +) -> list[str]: + """把 studio 运行态的编排层信号归一成 infra_flags(去重、稳定序;纯函数,供 studio 采集点委托)。 + + 信号 → 标记映射(与 INFRA_FLAGS 受控值集同源;新增信号在此加一条分支 + 上面登记一个常量即可): + - design_degraded → design_team_degraded(设计团队降级回单 agent); + - design_timeout → wall_timeout(设计团队墙钟超时;通常与 design_degraded 同现,但独立成标记); + - writer_iter_wall → writer_iter_wall(单写外层 resume 轮数墙); + - breaker_kind=='step_cap' → step_cap_tripped;=='timeout' → wall_timeout + (熔断 kind 见 middleware.Tier2CircuitBreak;budget/stuck 暂不入首批值集,由 circuit_break 字段另记); + - budget_soft_tripped → soft_budget_tripped(¥ 软预算软停,不断链)。 + + 稳定序:按 INFRA_FLAGS 登记序去重排列,便于台账/断言比对(不随触发先后抖动);无信号 → 空列表。 + 纯映射不碰 fail_system(游戏系统桶),严守编排层归因与游戏系统语义的正交分界。 + """ + hit: set[str] = set() + if design_degraded: + hit.add(INFRA_FLAG_DESIGN_TEAM_DEGRADED) + if design_timeout: + hit.add(INFRA_FLAG_WALL_TIMEOUT) # 设计团队墙钟超时也归 wall_timeout + if writer_iter_wall: + hit.add(INFRA_FLAG_WRITER_ITER_WALL) + if breaker_kind == "step_cap": + hit.add(INFRA_FLAG_STEP_CAP_TRIPPED) + elif breaker_kind == "timeout": + hit.add(INFRA_FLAG_WALL_TIMEOUT) # 单写 reply 墙钟/单步静默熔断 + if budget_soft_tripped: + hit.add(INFRA_FLAG_SOFT_BUDGET_TRIPPED) + # 按受控值集登记序输出(稳定、去重),便于台账比对与断言。 + return [f for f in INFRA_FLAGS if f in hit] + + # ────────────────────────────────────────────────────────────────────────── # append_record —— 唯一的显式落盘出口(JSONL 台账追加;本模块唯一副作用) # ──────────────────────────────────────────────────────────────────────────