fix(soak): 给 plan 与 execute 补上完成通路，它们原本必然跑满预算

打真实模型跑通一个完整任务时发现的：这两个阶段的工具集里没有带完成标记的工具，解释器也从不产出最终回答，于是模型在第 4 步写完产出物之后还在继续读文档，直到撞上步数上限。按原来的 50/50/16 算，二十个任务要两千三百多次调用，而且整批的停止原因会全是 step_budget，别的什么都压不出来。解释器加一条最终回答支路（{"final_answer": "..."}），plan 与 execute 的提示词写死收尾动作；summarize 不变，仍然只能靠 submit_finding 结束。这样三条完成路径同时在场：模型自报最终回答、agent 调用带完成标记的工具、环境报告完成（AppWorld 那路），它们的可信度各不相同，压测正需要这个对照。预算随之下调到 20/25/16——实测每阶段真实用 5 到 7 步，留了一倍余量。原来那个 50 的来历是 gov-auditor.yaml 的 turns 上限，但那边靠编排校验产物落盘来收阶段，跑满 turns 无所谓；这里靠模型自己收尾，上限定高只会让它在产出物写完之后接着白烧。实测：plan 7 步 agent_finished、execute 6 步 agent_finished、summarize 5 步 task_completed，一个任务 18 次调用。 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-11 08:58:57 -04:00
parent f277197071
commit 10f4a49f37
2 changed files with 177 additions and 29 deletions
@@ -35,7 +35,7 @@ from contextlib import closing
 from dataclasses import dataclass
 from pathlib import Path

-from polyloop.ports import Action, InvalidDecision, ParsedReply, ToolCall
+from polyloop.ports import Action, Decision, FinalAnswer, InvalidDecision, ParsedReply, ToolCall
 from polyloop.session import RunRequest
 from polyloop.tools import ToolRegistry, ToolSpec
 from polyloop.types import (
@@ -687,20 +687,27 @@ def build_audit_tasks(
 _FENCE_RE = re.compile(r"```(?:json|JSON)?[ \t\r]*\n(.*?)```", re.DOTALL)

 _NO_JSON = (
-    "你的回复里找不到 JSON 对象。这一步只能输出一个 JSON 对象，形如 "
-    '{"tool": "工具名", "arguments": {"参数名": "参数值"}}，不要写任何解释文字。'
+    "你的回复里找不到 JSON 对象。这一步只能输出一个 JSON 对象："
+    '要调工具就写 {"tool": "工具名", "arguments": {"参数名": "参数值"}}，'
+    '这一阶段做完了就写 {"final_answer": "一句话说明产出了什么"}。不要写任何解释文字。'
 )
 _NOT_OBJECT = (
    "解析出来的是 {kind}，不是 JSON 对象。最外层必须是一对花括号，"
-    '形如 {{"tool": "工具名", "arguments": {{}}}}。'
+    '形如 {{"tool": "工具名", "arguments": {{}}}} 或 {{"final_answer": "……"}}。'
 )
-_MISSING_TOOL = (
-    '缺少 "tool" 键。JSON 对象里必须有一个 "tool"，它的值是工具清单里的一个工具名（字符串）。'
+_NO_DECISION_KEY = (
+    '既没有可用的 "tool" 键，也没有 "final_answer" 键。JSON 对象里必须二选一：'
+    '"tool" 的值是工具清单里的一个工具名（字符串），'
+    '"final_answer" 的值是一句说明这一阶段产出了什么的文本。'
 )
 _BAD_ARGUMENTS = (
    '"arguments" 的值是 {kind}，不是对象。它必须是一对花括号包起来的参数表；'
    "没有参数就写空对象 {{}}。"
 )
+_EMPTY_FINAL_ANSWER = (
+    '"final_answer" 是空的。它必须写清楚这一阶段产出了什么、写在哪个文件里；'
+    "这一步之后本阶段就结束了，空文本等于什么都没交代。"
+)


 def _fence_stripped_candidates(content: str) -> list[str]:
@@ -722,13 +729,23 @@ def _fence_stripped_candidates(content: str) -> list[str]:


 class GovDocParser:
-    """把模型输出解释成一次工具调用。
+    """把模型输出解释成一次工具调用，或者一个最终回答。

-    协议是一个 JSON 对象 `{"tool": ..., "arguments": {...}}`。容错两种偏差：被代码围栏包住，
-    以及参数平铺在顶层而不是嵌在 `arguments` 里。这两种是实测里最常见的，且都不影响意图——
-    判成失败只会白烧一步。
+    协议是一个 JSON 对象：`{"tool": ..., "arguments": {...}}` 是工具调用，
+    `{"final_answer": "..."}` 是「这一阶段我做完了」。容错两种偏差：被代码围栏包住，以及参数
+    平铺在顶层而不是嵌在 `arguments` 里。这两种是实测里最常见的，且都不影响意图——判成失败
+    只会白烧一步。

-    **五种失败各给各的说明**，不合成一句：这段文本就是回喂给模型的那条观察，压成一句会改掉
+    **最终回答这一支是 plan 与 execute 唯一的收尾通路。** 那两个阶段的工具集里没有带
+    `completes_run` 的工具，没有这一支的话它们只能跑满预算：实测模型在第 4 步写完 `plan.md`
+    之后仍在继续读文档，停止原因是 `step_budget`。按 20 个任务算，白烧掉的调用比整批批准的
+    额度还多，而且整批的停止原因会全是 `step_budget`，别的停止路径一个都压不出来。
+
+    **两个键都在时以 `tool` 为准。** 模型偶尔会一边调工具一边宣布做完；执行那次调用只是多走
+    一步，而按 `final_answer` 收尾会把那次工具调用整个丢掉——那一步模型以为已经写进工作区的
+    东西其实不在。
+
+    **各种失败各给各的说明**，不合成一句：这段文本就是回喂给模型的那条观察，压成一句会改掉
    模型收到的纠错信息（`src/polyloop/ports/__init__.py:86`）。
    """

@@ -742,7 +759,7 @@ class GovDocParser:
    def parameters(self) -> Mapping[str, str]:
        return {"kind": "govdoc_json_tool_call"}

-    def _decide(self, content: str) -> Action | InvalidDecision:
+    def _decide(self, content: str) -> Decision:
        candidates = _fence_stripped_candidates(content)
        if not candidates:
            return InvalidDecision(explanation=_NO_JSON)
@@ -771,12 +788,15 @@ class GovDocParser:
            return InvalidDecision(explanation=first_syntax_error)
        return InvalidDecision(explanation=_NO_JSON)

-    def _from_payload(self, payload: object) -> Action | InvalidDecision:
+    def _from_payload(self, payload: object) -> Decision:
        if not isinstance(payload, dict):
            return InvalidDecision(explanation=_NOT_OBJECT.format(kind=type(payload).__name__))
        name = payload.get("tool")
        if not isinstance(name, str) or not name.strip():
-            return InvalidDecision(explanation=_MISSING_TOOL)
+            # 分派顺序：先问 `tool`，问不出可用的工具名才轮到 `final_answer`。
+            if "final_answer" in payload:
+                return self._final_answer(payload["final_answer"])
+            return InvalidDecision(explanation=_NO_DECISION_KEY)

        if "arguments" in payload:
            raw_arguments = payload["arguments"]
@@ -786,9 +806,14 @@ class GovDocParser:
                )
            arguments = {str(key): value for key, value in raw_arguments.items()}
        else:
-            # 参数平铺：模型把参数写在顶层而不是嵌在 `arguments` 里。除 `tool` 之外的键全当
-            # 参数收下——意图是清楚的，判成失败只会白烧一步。
-            arguments = {str(key): value for key, value in payload.items() if key != "tool"}
+            # 参数平铺：模型把参数写在顶层而不是嵌在 `arguments` 里。除了两个协议键之外的键
+            # 全当参数收下——意图是清楚的，判成失败只会白烧一步。`final_answer` 也要排掉：
+            # 模型一边调工具一边宣布做完时，把那句话当成工具参数会让这次调用过不了 schema。
+            arguments = {
+                str(key): value
+                for key, value in payload.items()
+                if key not in ("tool", "final_answer")
+            }

        summary = json.dumps(
            {"tool": name.strip(), "arguments": arguments},
@@ -801,6 +826,17 @@ class GovDocParser:
            tool_call=ToolCall(name=name.strip(), arguments=arguments),
        )

+    def _final_answer(self, raw: object) -> Decision:
+        """`{"final_answer": ...}` 那一支。
+
+        值不是字符串时序列化成 JSON 再收下，不判失败：模型偶尔把产出物清单写成一个数组，
+        那仍然是「我做完了」这个意思，而这一支拒绝一次就要多烧一整步。
+        """
+        text = raw if isinstance(raw, str) else json.dumps(raw, ensure_ascii=False, default=str)
+        if not text.strip():
+            return InvalidDecision(explanation=_EMPTY_FINAL_ANSWER)
+        return FinalAnswer(text=text)
+

 # ---------------------------------------------------------------------------
 # 四、工具与工作区
@@ -1110,9 +1146,16 @@ PHASE_TOOLS: Mapping[str, tuple[str, ...]] = {
    "summarize": ("read_document", "submit_finding"),
 }

-#: 步数与动作上限照 `gov-auditor.yaml` L9-25 的 `max_turns`：plan 50 / execute 50 /
-#: summarize 16。那三个数是 GovDoc 跑了 264 次真实审核之后定下来的，实测均值是 plan 23、
-#: execute 26、summarize 7 轮，上限留了约一倍余量。
+#: 步数与动作上限：plan 20 / execute 25 / summarize 16。
+#:
+#: 这三个数来自打真实模型的实测——每个阶段真正需要的是 6 到 12 步，这里留了约一倍余量，同时
+#: 让一次全量（20 个任务 × 3 个阶段）的调用量落在批准的额度里。
+#:
+#: **不再照抄 `gov-auditor.yaml` L9-25 的 50 / 50 / 16。** 那三个数是给「靠 `required_outputs`
+#: 校验产物落盘来结束阶段」的编排定的：那边跑满 turns 不要紧，产物一落盘阶段就收了。这里的
+#: 阶段是靠模型自己输出最终回答来收的，上限定得高只会让模型在产出物写完之后接着瞎读文档——
+#: 实测就是这样，第 4 步写完 `plan.md`，之后每一步都在白烧。summarize 保持 16，因为它有
+#: `submit_finding` 这条提交型完成通路，本来就不靠跑满预算结束。
 #:
 #: `max_prompt_chars` 取 400,000：主招标文件是 174,690 字符，一次运行里模型会分段读进来
 #: 相当一部分，再加上工具清单与历史，四十万给的是「读得进去但读不完」的空间——那正是这个
@@ -1122,10 +1165,10 @@ PHASE_TOOLS: Mapping[str, tuple[str, ...]] = {
 #: 没在按协议输出，再给机会只是烧钱。
 PHASE_BUDGETS: Mapping[str, Budget] = {
    "plan": Budget(
-        max_steps=50, max_actions=50, max_consecutive_parse_failures=3, max_prompt_chars=400_000
+        max_steps=20, max_actions=20, max_consecutive_parse_failures=3, max_prompt_chars=400_000
    ),
    "execute": Budget(
-        max_steps=50, max_actions=50, max_consecutive_parse_failures=3, max_prompt_chars=400_000
+        max_steps=25, max_actions=25, max_consecutive_parse_failures=3, max_prompt_chars=400_000
    ),
    "summarize": Budget(
        max_steps=16, max_actions=16, max_consecutive_parse_failures=3, max_prompt_chars=400_000
@@ -1154,6 +1197,8 @@ _PROTOCOL_BLOCK = """输出格式（每一步只输出一个 JSON 对象，不
 例：
 {"tool": "grep_document", "arguments": {"pattern": "注册地|所在地|分支机构", "path": "tender.md", "max_matches": 20}}

+除非上面的操作步骤另有规定，每一步都必须是这种工具调用形态。
+
 绝对禁止（违反会浪费这一步）：
 - 禁止在 JSON 之外写任何文字、标题或思考过程。
 - 禁止一次输出多个 JSON 对象。
@@ -1168,7 +1213,11 @@ _PHASE_INSTRUCTIONS: Mapping[str, str] = {
 1. 用 grep_document 按审核点判定标准里的关键词检索文书，拿到候选行号。关键词要换几组试，不要只搜一次。
 2. 用 read_document 读候选行号前后的上下文，确认这一处是不是真的与审核点相关。
 3. 用 write_note 写 plan.md：逐条列出候选证据的文档名、行号区间、以及它与审核点的关系。
-4. plan.md 写完就停下，不要重复步骤 1 到 3。
+4. 收尾：输出 {"final_answer": "<一句话说明这一阶段产出了什么、写在哪个文件里>"}，本阶段到此结束。
+
+绝对禁止：
+- plan.md 写完之后禁止再读任何文档、禁止重复步骤 1 到 3；必须立刻走第 4 步收尾。
+- 禁止用 final_answer 代替 plan.md：计划要写进文件，final_answer 只是一句交代。

 本阶段不下结论，也没有提交工具；结论在后面的阶段提交。""",
    "execute": """你是政府采购合规审核专家。本阶段是 execute：按上一阶段的计划逐条取证。
@@ -1177,7 +1226,11 @@ _PHASE_INSTRUCTIONS: Mapping[str, str] = {
 1. 用 read_document 读 plan.md（它在工作区里，直接写文件名）。
 2. 按 plan.md 给的行号区间逐条 read_document 核实原文；不够就用 grep_document 补检索。
 3. 用 write_note 写 evidence.md：每条证据一段，含文档名、行号、逐字摘录的原文（不要改写）、以及它支持还是反对「不合规」这个判断。
-4. evidence.md 写完就停下。
+4. 收尾：输出 {"final_answer": "<一句话说明这一阶段产出了什么、写在哪个文件里>"}，本阶段到此结束。
+
+绝对禁止：
+- evidence.md 写完之后禁止再读任何文档、禁止补搜；必须立刻走第 4 步收尾。
+- 禁止用 final_answer 代替 evidence.md：证据要写进文件，final_answer 只是一句交代。

 本阶段不下结论，也没有提交工具。""",
    "summarize": """你是政府采购合规审核专家。本阶段是 summarize：汇总证据并提交唯一一条结论。
@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING

 import pytest

-from polyloop.ports import Action, InvalidDecision
+from polyloop.ports import Action, FinalAnswer, InvalidDecision
 from polyloop.tools import ToolRegistry
 from polyloop.types import ModelReply, ReplayPolicy
 from tools.soak.scenarios.govdoc import (
@@ -217,7 +217,7 @@ def test_contract_4_action_carries_text():
 )
 def test_contract_5_parse_never_raises(content: str):
    parsed = GovDocParser().parse(_reply(content))
-    assert isinstance(parsed.decision, Action | InvalidDecision)
+    assert isinstance(parsed.decision, Action | FinalAnswer | InvalidDecision)


 def test_tolerates_code_fences():
@@ -264,6 +264,74 @@ def test_five_parse_failures_get_five_explanations():
    assert "arguments" in explanations["arguments 不是对象"]


+def test_final_answer_branch():
+    parsed = GovDocParser().parse(_reply('{"final_answer": "已写出 plan.md，列了 3 处候选证据。"}'))
+    assert isinstance(parsed.decision, FinalAnswer)
+    assert parsed.decision.text == "已写出 plan.md，列了 3 处候选证据。"
+    assert len(parsed.history_text) <= len(
+        '{"final_answer": "已写出 plan.md，列了 3 处候选证据。"}'
+    )
+
+
+def test_final_answer_tolerates_code_fences():
+    inner = '{"final_answer": "已写出 evidence.md，3 条证据。"}'
+    for fenced in (
+        f"```json\n{inner}\n```",
+        f"```\n{inner}\n```",
+        f"这一阶段做完了：\n```json\n{inner}\n```\n",
+    ):
+        parsed = GovDocParser().parse(_reply(fenced))
+        assert isinstance(parsed.decision, FinalAnswer), fenced
+        assert parsed.decision.text == "已写出 evidence.md，3 条证据。"
+
+
+def test_tool_wins_when_both_keys_are_present():
+    """一边调工具一边宣布做完时以工具为准：按 final_answer 收尾会把那次调用整个丢掉。"""
+    parsed = GovDocParser().parse(
+        _reply(
+            '{"tool": "write_note", "arguments": {"filename": "plan.md", "content": "x"},'
+            ' "final_answer": "我写完了"}'
+        )
+    )
+    assert isinstance(parsed.decision, Action)
+    assert parsed.decision.tool_call is not None
+    assert parsed.decision.tool_call.name == "write_note"
+
+
+def test_flattened_arguments_do_not_swallow_final_answer():
+    parsed = GovDocParser().parse(
+        _reply(
+            '{"tool": "write_note", "filename": "plan.md", "content": "x", "final_answer": "完"}'
+        )
+    )
+    assert isinstance(parsed.decision, Action)
+    assert parsed.decision.tool_call is not None
+    assert parsed.decision.tool_call.arguments == {"filename": "plan.md", "content": "x"}
+
+
+@pytest.mark.parametrize(
+    "content",
+    [
+        '{"answer": "我做完了"}',
+        '{"arguments": {"path": "tender.md"}}',
+        '{"final_answer": ""}',
+        '{"final_answer": "   "}',
+        "{}",
+    ],
+)
+def test_neither_key_is_still_invalid(content: str):
+    decision = GovDocParser().parse(_reply(content)).decision
+    assert isinstance(decision, InvalidDecision)
+    assert decision.explanation.strip()
+
+
+def test_missing_key_explanation_names_both_shapes():
+    decision = GovDocParser().parse(_reply('{"note": "x"}')).decision
+    assert isinstance(decision, InvalidDecision)
+    assert "tool" in decision.explanation
+    assert "final_answer" in decision.explanation
+
+
 def test_parser_reports_its_parameters():
    assert GovDocParser().parameters() == {"kind": "govdoc_json_tool_call"}

@@ -478,7 +546,11 @@ def test_all_three_phases_assemble(tmp_path: Path):
        assert request.injections == {}


-def test_budgets_follow_gov_auditor_turn_limits(tmp_path: Path):
+def test_budgets_are_the_measured_step_counts(tmp_path: Path):
+    """20 / 25 / 16 是实测出来的，不是照抄 gov-auditor.yaml 的 50 / 50 / 16。
+
+    断言具体数字是因为这三个数直接决定一次全量压测的调用量，改动必须是有意的。
+    """
    task = _sample_task()
    budgets = {
        phase: build_run_request(
@@ -490,12 +562,35 @@ def test_budgets_follow_gov_auditor_turn_limits(tmp_path: Path):
        ).budget
        for phase in ("plan", "execute", "summarize")
    }
-    assert (budgets["plan"].max_steps, budgets["plan"].max_actions) == (50, 50)
-    assert (budgets["execute"].max_steps, budgets["execute"].max_actions) == (50, 50)
+    assert (budgets["plan"].max_steps, budgets["plan"].max_actions) == (20, 20)
+    assert (budgets["execute"].max_steps, budgets["execute"].max_actions) == (25, 25)
    assert (budgets["summarize"].max_steps, budgets["summarize"].max_actions) == (16, 16)
    for budget in budgets.values():
        assert budget.max_consecutive_parse_failures == 3
        assert budget.max_prompt_chars == 400_000
+    # 一次全量（20 个任务 × 3 个阶段）的步数上限。这一条守的是额度，不是行为。
+    assert sum(budget.max_steps for budget in budgets.values()) * 20 == 1220
+
+
+def test_plan_and_execute_prompts_spell_out_the_final_answer_closing(tmp_path: Path):
+    """plan 与 execute 的工具集里没有带 completes_run 的工具，收尾只能靠最终回答。
+
+    提示词不写清楚这一步，这两个阶段就必然跑满预算——实测过，停止原因全是 step_budget。
+    """
+    task = _sample_task()
+    registry = _tools(tmp_path).registry()
+    for phase in ("plan", "execute"):
+        narrowed = registry.restrict_to(PHASE_TOOLS[phase])
+        system = build_context(task=task, phase=phase, tools=narrowed).run_level[0].content[0].text
+        assert '{"final_answer":' in system
+        assert not any(narrowed.spec_for(name).completes_run for name in narrowed.names())
+    # summarize 不变：它靠 submit_finding 这条提交型完成通路结束，不给最终回答这条路。
+    summarize = registry.restrict_to(PHASE_TOOLS["summarize"])
+    system = (
+        build_context(task=task, phase="summarize", tools=summarize).run_level[0].content[0].text
+    )
+    assert "final_answer" not in system
+    assert any(summarize.spec_for(name).completes_run for name in summarize.names())


 def test_context_carries_tools_but_not_the_document_body(tmp_path: Path):