iomgaa
2367d3afbc
0010 取代 0006 决策三里 tool_section_template 那一行。三条理由：两个真实消费者都是项目侧 自己把工具清单拼进上下文的（一个塞在 run 级模板里伪装成示例演示的一次执行输出，另一个的 render_tool_docs 全仓零生产调用方）；留着它的话迁移的人会去填，填完提示词里有两份工具清单， 而那个下游的验收标准是轨迹逐字段可比、变了还不报错；参考框架 pi 的内核同样不渲染，它的应用层 虽有 Available tools 段，但每行来自与 description 分开的 promptSnippet 字段——说明就算要渲染， 那段文字也不该是从校验用的 schema 生成的。四者同源不受影响，schema_for_model() 还在。 段序照唯一跑通了的下游：run 级片段 → 注入槽 → 目标级片段 → 逐步的模型输出/观察交替， 按变化频率从低到高排（供应商按前缀缓存计费）。库不合成任何 system 消息——有一个下游全程 没有 system 消息，加一条它的提示词会凭空多出一段。 注入槽每条一条 USER 消息、正文原样、不加任何分隔符（分隔符是渲染格式，归项目侧），通道按 名字排序（映射的迭代顺序取决于调用方怎么构造，用集合建出来的每进程都不同）。空注入等同有 测试守着。观察模板必须含占位符，缺了就报错——不校验的话每一步的观察会整个消失，而模型收到 的是一段看起来完全正常的固定文本。规模度量逐块问，认不得的块类型直接失败而不是当成 0。 零业务假设扫描第三次抓到我自己（模块 docstring 里写了「实验因子」），已改成中性说法。 migrations/govdoc-saas.md 登记了「模型原生工具调用没有路」这个缺口。
261 lines
9.2 KiB
Python
261 lines
9.2 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_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"),),
|
|
}
|
|
|
|
assert injected_entry_ids(entries) == ("a", "z")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 历史轮次
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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)
|