diff --git a/tools/soak/scenarios/govdoc.py b/tools/soak/scenarios/govdoc.py index fea5ed1..10ea0cc 100644 --- a/tools/soak/scenarios/govdoc.py +++ b/tools/soak/scenarios/govdoc.py @@ -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：汇总证据并提交唯一一条结论。 diff --git a/tools/soak/tests/test_govdoc_scenario.py b/tools/soak/tests/test_govdoc_scenario.py index 771daa5..7564d55 100644 --- a/tools/soak/tests/test_govdoc_scenario.py +++ b/tools/soak/tests/test_govdoc_scenario.py @@ -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):