iomgaa
3492a2994a
四件事，都来自第一个下游的 issue。 **RunRequest 新增 fingerprints。** 上下文与注入内容刻意不进快照（它们是数据不是参数）， 而这条规则把生成它们的提示词模板也一起挡在外面了。具体的失败场景：换一份模板续跑不报错， 前几步用 A 模板、后几步用 B，那次运行的数据已经废了却没有任何提示。键形状 request.fingerprint.<name>，和 model_binding 那个坐标分开——一个是「这次运行属于哪一格」， 一个是「用的配方是哪一版」，混在一个字段里事后分不开。默认空映射时一个键都不写，所以已有 配置算出来的快照逐字节不变。 **注入的通道维度不再拍平。** 一通道一键，于是「声明了通道但一条都没选中」和「压根没有这个 通道」分得开：前者是值为空串的键，后者是键不存在。有一档实验要比较的正是这两种情形。 **ActionExecutor 的 docstring 写定抛异常时会怎样**：环境故障走 ENV_ERROR 返回值，实现方真 抛了库不接管、异常原样穿出。理由在 design 0016；简言之库替它编一个结算结果就是在编造，而 副作用状态在那一刻是未知的。 **三个映射字段在构造期冻成只读。** 对抗审查发现 frozen=True 不禁止改字段里那个 dict，于是 构造期那道「快照取值必须是字符串」的校验能被绕过去：构造完往 fingerprints 里塞一个整数， 它一路进日志，要到续跑读日志时才炸——而那时这次运行已经完整跑过一遍。复用 tools 里已有的 _frozen，没另写一套。
292 lines
10 KiB
Python
292 lines
10 KiB
Python
"""上下文装配的行为。
|
|
|
|
最要紧的一条是**空注入等同**：注入为空时，装配结果必须与「根本没有这个槽位」逐字节相同。
|
|
某个下游第一阶段不注入任何东西，库要是悄悄多写一个空标题或一个换行，那一阶段的基线就和后续
|
|
阶段不可比了——而这件事不会报错，只表现成「这批分数有点不一样」。
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from polyloop._assembly import (
|
|
AssemblyError,
|
|
assemble,
|
|
check_observation_template,
|
|
history_messages,
|
|
injected_entry_ids,
|
|
injection_messages,
|
|
prompt_chars,
|
|
)
|
|
from polyloop.types import (
|
|
ActionStatus,
|
|
Context,
|
|
Injection,
|
|
Message,
|
|
Role,
|
|
StepRecord,
|
|
TextBlock,
|
|
)
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
TEMPLATE = "Output:\n```\n{observation}\n```\n\n"
|
|
|
|
|
|
def _msg(role: Role, text: str) -> Message:
|
|
return Message(role=role, content=(TextBlock(text=text),))
|
|
|
|
|
|
def _context() -> Context:
|
|
return Context(
|
|
run_level=(_msg(Role.USER, "你是一个助手"), _msg(Role.ASSISTANT, "好的")),
|
|
goal_level=(_msg(Role.USER, "任务：数到三"),),
|
|
)
|
|
|
|
|
|
def _step(idx: int, *, raw_output: str = "1", observation: str = "ok") -> StepRecord:
|
|
return StepRecord(
|
|
step_idx=idx,
|
|
raw_output=raw_output,
|
|
content_chars=len(raw_output),
|
|
thinking_chars=0,
|
|
action="1",
|
|
parse_ok=True,
|
|
parse_error=None,
|
|
observation=observation,
|
|
observation_is_synthetic=False,
|
|
observation_truncated_chars=0,
|
|
prompt_chars=0,
|
|
call_id=None,
|
|
step_wall_ms=0,
|
|
action_status=ActionStatus.EXECUTED,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 段序
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_the_segments_come_in_order() -> None:
|
|
"""run 级片段 → 注入槽 → 目标级片段 → 逐步交替。
|
|
|
|
照变化频率从低到高排：供应商按前缀缓存计费，把逐次变化的东西排到前面会让缓存静默失效，
|
|
而多付的幅度随注入内容的规模变化——于是缓存伪影会精确地伪装成下游想测的效应。
|
|
"""
|
|
messages = assemble(
|
|
context=_context(),
|
|
injections={"skill": (Injection(entry_id="e1", content="记得先看目录"),)},
|
|
steps=[_step(0)],
|
|
observation_template=TEMPLATE,
|
|
)
|
|
|
|
assert [block.text for message in messages for block in message.content] == [
|
|
"你是一个助手",
|
|
"好的",
|
|
"记得先看目录",
|
|
"任务：数到三",
|
|
"1",
|
|
"Output:\n```\nok\n```\n\n",
|
|
]
|
|
|
|
|
|
def test_the_library_never_synthesises_a_system_message() -> None:
|
|
"""有一个下游全程没有 system 消息，库自己加一条它的提示词会凭空多出一段。
|
|
|
|
要 system 消息的项目把它放进自己的 run 级片段里。
|
|
"""
|
|
messages = assemble(context=_context(), injections={}, steps=[], observation_template=TEMPLATE)
|
|
|
|
assert all(message.role is not Role.SYSTEM for message in messages)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 空注入等同
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_an_empty_injection_slot_leaves_nothing_behind() -> None:
|
|
"""注入为空时，装配结果与「根本没有这个槽位」逐字节相同。
|
|
|
|
这是 `design/0003` 点名的一条验收标准。库要是悄悄多写一个空标题或一个换行，不注入的那个
|
|
阶段就和后续阶段不可比了。
|
|
"""
|
|
context = _context()
|
|
steps = [_step(0)]
|
|
|
|
with_slot = assemble(context=context, injections={}, steps=steps, observation_template=TEMPLATE)
|
|
without_slot = (*context.run_level, *context.goal_level, *history_messages(steps, TEMPLATE))
|
|
|
|
assert with_slot == without_slot
|
|
|
|
|
|
def test_a_channel_with_no_entries_leaves_nothing_behind() -> None:
|
|
"""一个空通道和「没有这个通道」也必须等同，否则按阶段建通道的项目会踩到。"""
|
|
assert injection_messages({"skill": (), "hint": ()}) == ()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 注入槽
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_each_injection_is_its_own_message_with_nothing_added() -> None:
|
|
"""每条一条消息，正文原样，前后不加任何标题、分隔符或换行。
|
|
|
|
拼成一条就要选一个分隔符，而分隔符是渲染格式、归项目侧。
|
|
"""
|
|
messages = injection_messages(
|
|
{"skill": (Injection(entry_id="a", content="甲"), Injection(entry_id="b", content="乙"))}
|
|
)
|
|
|
|
assert messages == (_msg(Role.USER, "甲"), _msg(Role.USER, "乙"))
|
|
|
|
|
|
def test_channels_are_ordered_by_name_not_by_mapping_order() -> None:
|
|
"""映射的迭代顺序取决于调用方怎么构造它，用集合推导建出来的话每进程都可能不同。
|
|
|
|
照那个顺序渲染，同一份配置在不同进程里会渲染出不同的提示词。
|
|
"""
|
|
entries = {
|
|
"zeta": (Injection(entry_id="z", content="Z"),),
|
|
"alpha": (Injection(entry_id="a", content="A"),),
|
|
}
|
|
|
|
assert [block.text for m in injection_messages(entries) for block in m.content] == ["A", "Z"]
|
|
|
|
|
|
def test_entry_ids_keep_the_channel_they_came_from() -> None:
|
|
"""参数快照里记的是条目标识，正文不进——正文可能很大，而「这次贴了哪几条」事后要查得到。
|
|
|
|
通道维度保留：拍平之后「声明了这个通道但一条都没选中」和「压根没有这个通道」是同一个
|
|
结果，而有下游要比较的正是这两种情形。
|
|
"""
|
|
entries = {
|
|
"zeta": (Injection(entry_id="z", content="Z"),),
|
|
"alpha": (Injection(entry_id="a", content="A"), Injection(entry_id="b", content="B")),
|
|
}
|
|
|
|
assert injected_entry_ids(entries) == {"alpha": ("a", "b"), "zeta": ("z",)}
|
|
|
|
|
|
def test_a_declared_but_empty_channel_is_not_the_same_as_a_missing_one() -> None:
|
|
"""一个是值为空元组的项，另一个是这个项不存在。两者拼进快照才分得开。"""
|
|
assert injected_entry_ids({"skill": ()}) == {"skill": ()}
|
|
assert injected_entry_ids({}) == {}
|
|
|
|
|
|
def test_entry_ids_come_back_in_the_same_order_as_the_messages() -> None:
|
|
"""两个函数的顺序必须一致，它们同住一个模块就是为了守住这一点。
|
|
|
|
对不上的话，快照记的贴入顺序和模型真正看到的顺序是两回事，而续跑守卫照样全绿。
|
|
"""
|
|
entries = {
|
|
"zeta": (Injection(entry_id="z", content="Z"),),
|
|
"alpha": (Injection(entry_id="a", content="A"), Injection(entry_id="b", content="B")),
|
|
}
|
|
by_content = {"A": "a", "B": "b", "Z": "z"}
|
|
|
|
flattened = [
|
|
entry_id for entry_ids in injected_entry_ids(entries).values() for entry_id in entry_ids
|
|
]
|
|
from_messages = [
|
|
by_content[block.text] for m in injection_messages(entries) for block in m.content
|
|
]
|
|
|
|
assert flattened == from_messages
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 历史轮次
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_each_step_becomes_two_messages() -> None:
|
|
messages = history_messages([_step(0), _step(1, raw_output="2", observation="ok2")], TEMPLATE)
|
|
|
|
assert [(m.role, m.content[0].text) for m in messages] == [
|
|
(Role.ASSISTANT, "1"),
|
|
(Role.USER, "Output:\n```\nok\n```\n\n"),
|
|
(Role.ASSISTANT, "2"),
|
|
(Role.USER, "Output:\n```\nok2\n```\n\n"),
|
|
]
|
|
|
|
|
|
def test_the_assistant_turn_is_the_stored_raw_output_not_a_re_rendered_one() -> None:
|
|
"""`raw_output` 是解释器交回来的那段文本，不一定等于模型原文。
|
|
|
|
解释器有权改写它，比如把第一个代码围栏之后的内容整段丢掉——模型常在代码块后面编造执行
|
|
结果，留着的话下一轮它会把那段幻想当成真发生过的事。库照它回填，不自己截断。
|
|
"""
|
|
step = _step(0, raw_output="```python\nprint(1)\n```")
|
|
|
|
assert history_messages([step], TEMPLATE)[0].content[0].text == "```python\nprint(1)\n```"
|
|
|
|
|
|
def test_no_steps_means_no_history_messages() -> None:
|
|
assert history_messages([], TEMPLATE) == ()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 观察模板
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_template_without_the_placeholder_is_refused() -> None:
|
|
"""照它渲染的话每一步的观察会整个消失，而模型收到的是一段看起来完全正常的固定文本。
|
|
|
|
模型会以为每次执行都返回了同样的东西，然后开始瞎猜，而轨迹里那一列明明存着真的观察。
|
|
"""
|
|
with pytest.raises(AssemblyError, match="占位符"):
|
|
check_observation_template("Output:\n```\n\n```")
|
|
|
|
|
|
def test_a_template_with_an_unescaped_brace_is_refused() -> None:
|
|
"""模板里除占位符之外的花括号要写成双份，没转义的在这里就被抓出来。"""
|
|
with pytest.raises(AssemblyError, match="渲染不了"):
|
|
check_observation_template('{"result": {observation}, "extra": {oops}}')
|
|
|
|
|
|
def test_a_template_with_escaped_braces_is_accepted() -> None:
|
|
check_observation_template('{{"result": "{observation}"}}')
|
|
|
|
|
|
def test_the_template_is_checked_before_any_message_is_built() -> None:
|
|
"""坏模板在装配入口就报错，不是走到第三步才发现。"""
|
|
with pytest.raises(AssemblyError):
|
|
assemble(
|
|
context=_context(), injections={}, steps=[_step(0)], observation_template="没有占位符"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 规模度量
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_prompt_chars_counts_every_block_of_every_message() -> None:
|
|
messages = (
|
|
_msg(Role.USER, "12345"),
|
|
Message(role=Role.ASSISTANT, content=(TextBlock(text="ab"), TextBlock(text="cde"))),
|
|
)
|
|
|
|
assert prompt_chars(messages) == 10
|
|
|
|
|
|
def test_an_empty_prompt_measures_zero() -> None:
|
|
assert prompt_chars(()) == 0
|
|
|
|
|
|
def test_an_unknown_block_type_fails_instead_of_counting_zero() -> None:
|
|
"""将来加图片块时如果漏了对应的度量，当成 0 的后果是一个真的超限的提示词悄悄通过。
|
|
|
|
那时报错来自模型网关，看不出是本库的规模判定漏了一类块。
|
|
"""
|
|
|
|
class _Mystery:
|
|
pass
|
|
|
|
messages = (Message(role=Role.USER, content=(_Mystery(),)),) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(AssemblyError, match="规模度量"):
|
|
prompt_chars(messages)
|