iomgaa
4bf74bf5ac
四个纯函数各管一档，顺序本身由主循环走：预算准入（A）、提示词规模（B）、连续解析失败（D）、 完成判定（G）。加一个不可变的三计数结构。 三处刻意钉死的边界，都写了测试： - 两个预算上界同时耗尽报步数那一个。理由是兼容性不是原理（某下游按步数耗尽的占比告警）， 正因为推不出来才必须被测试钉死。 - 预算「达到即拦」用 >=，提示词规模「超过才拦」用 >。两处不是同一件事：一个数已经用掉 几个额度，一个量一个东西有多大。 - 环境故障排在完成判定前面；未执行不做完成判定；完成信号恒为假不是故障（初稿在这里 写错过，会让某个下游的每一次运行都在第一步终止）。 不 import polyloop.tools，五个逻辑层模块互不 import，所以「这次执行的工具被标了完成标记吗」 由调用方查好注册表传一个布尔进来。
230 lines
9.1 KiB
Python
230 lines
9.1 KiB
Python
"""停止判定每一档的行为。
|
|
|
|
这个模块的输入空间小到可以穷举，所以这里就穷举：三个计数各自在上限的下方、正好、上方，
|
|
四种动作状态叉乘两种完成信号与两种完成标记。**边界那一格是重点**——「恰好用满预算完成」
|
|
和「预算耗尽」的轨迹长度一模一样，判错了事后从数据里分不出来。
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from polyloop._stopping import (
|
|
RunCounters,
|
|
budget_admission,
|
|
completion_verdict,
|
|
parse_failure_admission,
|
|
prompt_size_admission,
|
|
)
|
|
from polyloop.types import ActionOutcome, ActionStatus, Budget, StopReason
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
BUDGET = Budget(
|
|
max_steps=10,
|
|
max_actions=4,
|
|
max_consecutive_parse_failures=3,
|
|
max_prompt_chars=1000,
|
|
)
|
|
|
|
|
|
def _outcome(
|
|
status: ActionStatus,
|
|
*,
|
|
env_reported_completion: bool = False,
|
|
) -> ActionOutcome:
|
|
return ActionOutcome(
|
|
status=status,
|
|
observation="",
|
|
observation_is_synthetic=False,
|
|
env_reported_completion=env_reported_completion,
|
|
observation_truncated_chars=0,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 计数
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_counters_start_at_zero() -> None:
|
|
assert RunCounters() == RunCounters(
|
|
steps_appended=0, actions_executed=0, consecutive_parse_failures=0
|
|
)
|
|
|
|
|
|
def test_advancing_a_counter_leaves_the_original_alone() -> None:
|
|
"""不可变，每次推进返回新实例。并发的多次运行各持一份。"""
|
|
start = RunCounters()
|
|
|
|
advanced = start.with_step_appended().with_action_executed()
|
|
|
|
assert start == RunCounters()
|
|
assert advanced == RunCounters(steps_appended=1, actions_executed=1)
|
|
|
|
|
|
def test_a_step_and_an_executed_action_are_counted_separately() -> None:
|
|
"""同一步里两个计数可能一个加一个不加。
|
|
|
|
步数数的是追加进轨迹的条数（解析失败、模型调用失败、环境故障的步都算），已执行动作数
|
|
只数动作执行接缝返回「已执行」的次数。合成一个的话，「模型反复调不存在的工具烧光预算」
|
|
和「真的做了五十步没做完」在轨迹上就分不开了。
|
|
"""
|
|
only_a_step = RunCounters().with_step_appended()
|
|
|
|
assert only_a_step.steps_appended == 1
|
|
assert only_a_step.actions_executed == 0
|
|
|
|
|
|
def test_a_parse_success_clears_the_consecutive_counter() -> None:
|
|
"""不清零的话，一次运行里零散的几次解析失败会累加到上限，然后被报成「连续失败」。"""
|
|
counters = RunCounters().with_parse_failure().with_parse_failure()
|
|
|
|
assert counters.with_parse_success().consecutive_parse_failures == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# A 档：预算准入
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("steps", [0, 1, 9])
|
|
def test_below_the_step_limit_the_run_continues(steps: int) -> None:
|
|
assert budget_admission(RunCounters(steps_appended=steps), BUDGET) is None
|
|
|
|
|
|
@pytest.mark.parametrize("steps", [10, 11])
|
|
def test_reaching_the_step_limit_stops_the_run(steps: int) -> None:
|
|
"""上限是可以取到的：已追加步数**达到**上限就不许再走一步。"""
|
|
assert budget_admission(RunCounters(steps_appended=steps), BUDGET) is StopReason.STEP_BUDGET
|
|
|
|
|
|
@pytest.mark.parametrize(("actions", "expected"), [(3, None), (4, StopReason.ACTION_BUDGET)])
|
|
def test_the_action_limit_has_its_own_stop_reason(actions: int, expected: object) -> None:
|
|
assert budget_admission(RunCounters(actions_executed=actions), BUDGET) is expected
|
|
|
|
|
|
def test_when_both_limits_are_hit_the_step_one_wins() -> None:
|
|
"""两个上界同时耗尽时报步数那一个。
|
|
|
|
理由是兼容性不是原理——某个下游的告警判据按步数耗尽的占比统计，报另一个会让那条判据在
|
|
这种情况下漏掉。正因为它不是从原理推出来的，才必须被测试钉死，而不是留给实现随手决定。
|
|
"""
|
|
both = RunCounters(steps_appended=10, actions_executed=4)
|
|
|
|
assert budget_admission(both, BUDGET) is StopReason.STEP_BUDGET
|
|
|
|
|
|
def test_the_budget_is_checked_before_the_step_that_would_exceed_it() -> None:
|
|
"""走完第 9 步（还差一步到上限）时不停，走完第 10 步才停。
|
|
|
|
这一条守的是「预算结算在下一次迭代的开头」：一次恰好用满预算完成的运行走的是完成判定，
|
|
记成目标达成；结算放在本次结尾的话，它会先撞上预算上限记成预算耗尽。两者的轨迹长度
|
|
一模一样，事后分不出来。
|
|
"""
|
|
assert budget_admission(RunCounters(steps_appended=9), BUDGET) is None
|
|
assert budget_admission(RunCounters(steps_appended=10), BUDGET) is StopReason.STEP_BUDGET
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# B 档：提示词规模
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("chars", "expected"),
|
|
[(0, None), (999, None), (1000, None), (1001, StopReason.CONTEXT_OVERFLOW)],
|
|
)
|
|
def test_the_prompt_size_limit_itself_is_allowed(chars: int, expected: object) -> None:
|
|
"""**超过**上限才拦，正好等于上限是允许的。
|
|
|
|
和 A 档那个「达到即拦」不是一回事：那里数的是已经用掉几个额度，这里量的是一个东西有
|
|
多大。
|
|
"""
|
|
assert prompt_size_admission(chars, BUDGET) is expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# D 档：连续解析失败
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("failures", "expected"),
|
|
[(0, None), (2, None), (3, StopReason.PARSE_FAILED_REPEATEDLY)],
|
|
)
|
|
def test_consecutive_parse_failures_stop_the_run_at_the_limit(
|
|
failures: int, expected: object
|
|
) -> None:
|
|
assert (
|
|
parse_failure_admission(RunCounters(consecutive_parse_failures=failures), BUDGET)
|
|
is expected
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# G 档：完成判定
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("tool_completes_run", [True, False])
|
|
@pytest.mark.parametrize("env_reported_completion", [True, False])
|
|
def test_an_env_error_stops_the_run_whatever_else_is_true(
|
|
env_reported_completion: bool, tool_completes_run: bool
|
|
) -> None:
|
|
"""环境故障排在完成判定前面，其余两个信号都盖不过它。
|
|
|
|
环境坏了就没有下一步可走，继续跑只会产出一串同样的故障，把预算烧光而轨迹上全是噪声。
|
|
"""
|
|
outcome = _outcome(ActionStatus.ENV_ERROR, env_reported_completion=env_reported_completion)
|
|
|
|
assert completion_verdict(outcome, tool_completes_run) is StopReason.ENV_ERROR
|
|
|
|
|
|
@pytest.mark.parametrize("tool_completes_run", [True, False])
|
|
@pytest.mark.parametrize("env_reported_completion", [True, False])
|
|
def test_a_rejected_action_never_completes_the_run(
|
|
env_reported_completion: bool, tool_completes_run: bool
|
|
) -> None:
|
|
"""状态是「未执行」时不做完成判定。
|
|
|
|
动作根本没进入真实执行，环境状态没变，完成条件不可能因为它成立。这一档回到预算准入
|
|
接着跑。
|
|
"""
|
|
outcome = _outcome(ActionStatus.NOT_EXECUTED, env_reported_completion=env_reported_completion)
|
|
|
|
assert completion_verdict(outcome, tool_completes_run) is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("env_reported_completion", "tool_completes_run", "expected"),
|
|
[
|
|
(False, False, None),
|
|
(True, False, StopReason.TASK_COMPLETED),
|
|
(False, True, StopReason.TASK_COMPLETED),
|
|
(True, True, StopReason.TASK_COMPLETED),
|
|
],
|
|
)
|
|
def test_either_completion_channel_ends_an_executed_action(
|
|
env_reported_completion: bool, tool_completes_run: bool, expected: object
|
|
) -> None:
|
|
"""两条完成通路任一成立都收尾，可信度的差别不体现在停止原因上。
|
|
|
|
一条是环境状态里真的留下了记录，一条是 agent 调了一个被标为完成标记的工具、环境状态
|
|
一点没变。它们共用一个停止原因，因为「这次运行为什么停」的答案是同一个；要区分是哪一种
|
|
看那一步的步记录。
|
|
"""
|
|
outcome = _outcome(ActionStatus.EXECUTED, env_reported_completion=env_reported_completion)
|
|
|
|
assert completion_verdict(outcome, tool_completes_run) is expected
|
|
|
|
|
|
def test_a_completion_signal_that_is_always_false_is_not_a_failure() -> None:
|
|
"""没有环境完成信号的环境就是这么返回的，走「接着跑」那一支。
|
|
|
|
初稿在这里写错过：那时完成信号被定成「布尔或空、空表示取不到」，这一档写着「取不到就是
|
|
环境故障」——而有一个下游每一步都返回空，于是它的每一次运行都会在第一步撞环境故障终止。
|
|
不是边缘情况，是全部。
|
|
"""
|
|
outcome = _outcome(ActionStatus.EXECUTED, env_reported_completion=False)
|
|
|
|
assert completion_verdict(outcome, tool_completes_run=False) is None
|