"""上下文装配的行为。 最要紧的一条是**空注入等同**：注入为空时，装配结果必须与「根本没有这个槽位」逐字节相同。 某个下游第一阶段不注入任何东西，库要是悄悄多写一个空标题或一个换行，那一阶段的基线就和后续 阶段不可比了——而这件事不会报错，只表现成「这批分数有点不一样」。 """ 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)