架构定位

基础认知

建立模型输入、上下文和工程基础，让后续系统拥有清楚、可验证的起点。

模型接收什么信息，开发者如何建立可靠输入与工程基础？

受这些层约束

概念校准

上下文

它是什么: 一次模型调用可见的信息集合，包括指令、对话、工具 Schema 与结果、检索证据和当前状态。
它不是什么: 不只是聊天记录，也不等于模型能够永久记住的全部信息。

模型运行原理与多模态输入

理解 Token、上下文、Embedding 和视觉输入，并构建统一输入适配层。

本章只做一件事让三个场景的文本、图片和结构化文件进入同一输入合同。

开始前，Forge 已经具备已完成 Module 03 的“Python、Git、HTTP 与工程基础”，其通过验收的 solution 是本章起点。

完成后，Forge 将能够让三个场景的文本、图片和结构化文件进入同一输入合同。

卡住时的最小恢复点先只解析文本和文件元数据，对不支持的媒体给出明确拒绝。

下载本章实验仓库Python 3.12 · pytest · Pydantic · SQLite · deterministic Mock

阶段能力链

01用可替换编码器估算消息 Token，并为系统指令和输出保留预算02在不可丢弃约束和可选材料之间分配上下文预算，超限时明确拒绝03计算余弦相似度并拒绝维度不一致或零向量，避免把 Embedding 当答案04规范化 UTF-8 文本、换行和大小限制，同时保留来源元数据05依据真实文件签名和像素上限验证 PNG/JPEG，而不是只相信扩展名06把 JSON、CSV 和 Markdown 解析成带来源定位的块，并对格式错误给出明确失败07根据模型能力和媒体类型路由输入，并对不支持组合返回可解释拒绝

每一步只增加一种可验证能力；后一步建立在前一步已经通过的代码和测试上。

概念校准\n\n在开始操作代码前，先把本章涉及的概念放回正确的工程边界。\n\n
\n\n### 上下文 context\n\n- 它是什么: 一次模型调用实际可见的信息集合，包括指令、对话片段、工具 Schema 与结果、检索证据、当前状态和输出约束。\n- 它不是什么: 它不只是聊天记录，也不是数据库中的全部信息；只有经过选择并放入本次调用的信息才属于本次上下文。\n- 与相邻概念的关系: RAG 选择外部证据，Memory 保存跨步骤状态，Tool Schema 描述可用动作；三者都可供给上下文，但职责不同。\n- 在 HeatStack Forge 中的位置: Forge 的 context builder 按任务、权限、预算和来源优先级组装输入，并记录每段信息的来源与版本。\n- 典型误用与修正: 误用是把更多 Token 当成更完整的理解。修正方法是建立选择、截断、引用和敏感信息过滤规则，并用固定夹具验证。\n\n

STAGE 01

用可替换编码器估算消息 Token，并为系统指令和输出保留预算

先看一个具体问题

为什么字符数不能替代 Token 数？

在构建多模态输入管道时，我们需要估算消息列表将消耗多少 Token，以便为系统指令和输出预留预算。

如果直接使用字符长度来估算，就会把“hello model”这样的短语算成 11 个 Token，而实际分词器可能只产生 2 个 Token。

这种混淆会导致预算严重失真：短词多字符的消息被高估，而长词少字符的消息被低估。

本阶段要求你实现 estimate_tokens 函数，它必须调用传入的编码器来获得每个消息的 Token 序列，而不是依赖字符数。

先做判断

假设你有一个消息列表 ["hello model", "short"]，使用 whitespace_encoder 编码，每条消息固定开销为 2。如果 estimate_tokens 错误地使用字符长度而不是编码器，结果会是多少？

7（正确使用编码器）
20（错误使用字符长度）
5（只计算单词数）
无法确定

判断依据: 正确答案是 20。字符长度计算为 len("hello model") + len("short") + 2*2 = 11 + 5 + 4 = 20。而正确使用编码器时，whitespace_encoder 将每个消息按空格分割，"hello model" 得到 2 个 Token，"short" 得到 1 个 Token，加上每条消息开销 2，总数为 2+2+1+2=7。

它是什么

Token 估算是什么

Token 估算是一个函数，它接收消息列表、编码器和每条消息的固定开销，返回总 Token 数。

编码器负责将文本转换为 Token ID 列表，估算函数只关心列表长度，不关心具体 ID 值。

每条消息的固定开销模拟了消息包装格式（如角色标记、分隔符）带来的额外 Token 消耗。

它不是什么

Token 估算不是什么

Token 估算不是字符计数：字符数忽略了分词规则，例如一个单词可能对应多个 Token，多个单词也可能合并为一个 Token。

它也不是精确的模型调用成本计算：真实模型还有注意力缓存、批处理等开销，这里只做线性估算。

它不负责选择编码器：编码器由调用方传入，估算函数必须保持可替换性。

它与相邻概念的关系

与其他概念的关系

编码器是 Token 估算的依赖：没有编码器，估算函数无法知道文本如何被切分。

每条消息开销是独立的加法项：它不参与编码，但影响总预算。

负开销校验是边界条件：如果开销为负，总 Token 数可能被低估，因此必须拒绝。

本阶段的边界决定: 当 per_message_overhead 为负数时，估算结果可能小于实际编码 Token 数，因此必须抛出 ValueError 并包含 “non-negative” 字样。

实验 ID: model-multimodal-input-01-token-accounting
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/model-multimodal-input/01-token-accounting/
验收: 待验收

教师示范

目标: 演示 estimate_tokens 如何调用 whitespace_encoder 对消息列表进行编码，并加上每条消息的固定开销，得到总 Token 数。

假设我们有一个消息列表 ["hello model", "short"]，编码器是 whitespace_encoder，每条消息开销为 2。

whitespace_encoder 将文本按空格分割，返回一个从 0 开始的整数列表，长度等于单词数。

对于 "hello model"，它返回 [0, 1]，长度为 2；对于 "short"，返回 [0]，长度为 1。

estimate_tokens 对每条消息调用编码器，取返回列表的长度，再加上每条消息开销 2，最后求和。

因此总 Token 数为 (2 + 2) + (1 + 2) = 7。

实施步骤

定义 whitespace_encoder：return list(range(len(text.split())))。
在 estimate_tokens 中，对每个 part 调用 encode(part) 得到 Token ID 列表。
使用 len(encode(part)) 获取该消息的 Token 数。
加上 per_message_overhead，然后对所有消息求和。
返回总和。

真实 solution 代码: forge/tokens.py

from collections.abc import Callable

def estimate_tokens(parts: list[str], encode: Callable[[str], list[int]], per_message_overhead: int = 3) -> int:
    if per_message_overhead < 0: raise ValueError("overhead must be non-negative")
    return sum(len(encode(part)) + per_message_overhead for part in parts)


def whitespace_encoder(text: str) -> list[int]:
    return list(range(len(text.split())))

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

测试断言 estimate_tokens(["hello model","short"], whitespace_encoder, 2) == 7，验证了编码器被正确调用。

如果函数使用字符长度，结果会是 20，测试失败。

通过这个例子，我们理解了 Token 估算必须委托给编码器，而不是自行猜测分词规则。

带提示练习

补全 estimate_tokens 中缺失的编码调用，使测试 test_estimate_uses_encoder_not_character_count 通过。

打开 forge/tokens.py 文件。
找到 estimate_tokens 函数，当前实现为 return sum(len(part) + per_message_overhead for part in parts)。
将 len(part) 替换为 len(encode(part))。
运行测试 python -m pytest -q tests/test_stage.py::test_estimate_uses_encoder_not_character_count。
确认测试通过。

逐级提示

注意 encode 是传入的函数参数，你需要调用它。
encode(part) 返回一个列表，你需要取其长度。
不要忘记保留 per_message_overhead 的加法。

完成标准

测试 test_estimate_uses_encoder_not_character_count 通过。
代码中不再出现 len(part)，而是 len(encode(part))。

故障注入与排查

你接手了一个有缺陷的 forge/tokens.py 实现。运行测试 test_estimate_uses_encoder_not_character_count 时，断言失败，实际返回 20 而不是期望的 7。你需要定位并修复这个缺陷。

本阶段已经建立了 Token 估算必须委托给编码器的心理模型。通过检查 estimate_tokens 的实现，你可以发现它错误地使用了 len(part) 而不是 len(encode(part))，这正是字符数与 Token 数混淆的典型表现。修复后，测试将通过。

已验证的故障合同

目标测试: test_estimate_uses_encoder_not_character_count
故障文件: forge/tokens.py

诊断顺序

运行测试 python -m pytest -q ../tests/test_stage.py::test_estimate_uses_encoder_not_character_count，观察失败输出。
注意到断言 assert estimate_tokens(["hello model","short"],whitespace_encoder,2) == 7 失败，实际值为 20。
检查 estimate_tokens 函数体，发现 return sum(len(part) + per_message_overhead for part in parts)。
识别出 len(part) 计算的是字符数，而不是编码后的 Token 数。
将 len(part) 改为 len(encode(part))，重新运行测试。

修复原则: 将 estimate_tokens 中的 len(part) 替换为 len(encode(part))，确保调用传入的编码器来获取 Token 序列长度。修复后的函数为：return sum(len(encode(part)) + per_message_overhead for part in parts)。

python -m pytest -q ../tests/test_stage.py::test_estimate_uses_encoder_not_character_count

迁移挑战

在办公研究场景中，你需要估算一份决策简报的 Token 数。简报由三个消息组成：["市场分析显示增长趋势", "竞争对手推出新产品", "建议增加研发投入"]。使用 whitespace_encoder 和每条消息开销 3 进行估算。

改变的条件: 消息内容从英文短词变为中文短语，每条消息开销从 2 变为 3。

编写一个 Python 脚本，导入 estimate_tokens 和 whitespace_encoder，计算上述消息列表的总 Token 数，并验证当 per_message_overhead 为 -1 时抛出 ValueError。

完成标准

脚本输出总 Token 数为 12（每个中文短语按空格分割为 1 个 Token，三条消息共 3 个 Token，加上每条消息开销 3，总计 3+3*3=12）。
脚本验证负开销被拒绝：调用 estimate_tokens(["x"], whitespace_encoder, -1) 时捕获到 ValueError，且错误信息包含 “non-negative”。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么 estimate_tokens 必须使用编码器而不是字符长度？请结合 whitespace_encoder 的行为解释，并说明如果使用字符长度会导致什么后果。

一个有力回答应包含

编码器模拟了真实模型的分词过程，不同编码器会产生不同的 Token 序列，字符长度无法反映这种差异。
whitespace_encoder 按空格分割，因此 "hello model" 产生 2 个 Token，而字符长度是 11，两者差异巨大。
使用字符长度会导致预算估算错误，可能高估或低估实际 Token 消耗，影响系统指令和输出预留的准确性。

掌握证据

测试 test_estimate_uses_encoder_not_character_count 通过，证明 estimate_tokens 正确调用了编码器。
测试 test_negative_message_overhead_is_rejected 通过，证明负开销被拒绝并抛出包含 “non-negative” 的 ValueError。
在 forge/tokens.py 中实现了 estimate_tokens 和 whitespace_encoder，且代码通过全部测试。

STAGE 02

在不可丢弃约束和可选材料之间分配上下文预算，超限时明确拒绝

先看一个具体问题

有限窗口下，可选材料挤占必需指令

你正在为多模态输入模块实现上下文选择函数 select_context。当前代码把整个窗口当作可自由分配的容量，按优先级从高到低填充所有条目，没有区分必需项和可选项，也没有为模型输出预留空间。

在测试 test_output_reserve_and_required_context_are_protected 中，传入 system（60 tokens，必需）和 notes（30 tokens，可选，优先级更高），窗口 100，输出预留 20。正确结果应只返回 ["system"]，因为扣除输出预留后可用预算为 80，必需项 system 占 60，剩余 20 不足以容纳 notes。

但当前实现会先放入优先级更高的 notes，再放入 system，返回 ['notes', 'system']，导致必需的系统指令被可选笔记挤占，且没有保护输出空间。

另一个测试 test_impossible_required_context_is_rejected 要求当必需项本身超过预算时抛出 ValueError，而不是静默返回部分结果。

先做判断

在 select_context 中，如果窗口为 100、输出预留为 20，且有一个必需项占 90 tokens，函数应该怎样处理？

返回该必需项，忽略输出预留
抛出异常，因为必需项超过可用预算
返回空列表，因为无法满足
先放入其他可选项，再尝试放入必需项

判断依据: 正确做法是抛出异常。必需项是系统指令或策略等不可丢弃的内容，如果它们本身超过扣除输出预留后的预算，任何部分结果都会破坏系统约束，因此必须立即拒绝请求。

它是什么

上下文预算分配模型

上下文预算分配是一个两阶段过程：首先从总窗口中扣除输出预留，得到可用于输入内容的预算；然后先无条件纳入所有必需项，再按优先级从高到低填充可选项，直到剩余容量不足。

必需项是系统提示、策略文档等不可丢弃的约束，它们必须全部进入上下文，否则模型行为可能违反规则。可选项是检索到的参考材料、历史对话等，它们可以按优先级取舍。

它不是什么

不是简单的优先级排序

上下文预算分配不是把所有条目按优先级排序后依次放入，因为这样做会让高优先级的可选项挤占低优先级的必需项。

它也不是把整个窗口都用于输入内容，必须为模型输出预留空间，否则模型可能没有足够 token 来生成回答。

它与相邻概念的关系

与 token 估算和上下文概念的关系

本阶段依赖 Stage 01 的 token 估算函数 estimate_tokens 来获得每个条目的 token 数，但 select_context 本身只接收已经计算好的 token 数。

上下文预算分配是上下文概念的具体实现：它决定了哪些信息进入本次模型调用，体现了“上下文是经过选择的信息集合”这一边界。

本阶段的边界决定: 当必需项的总 token 数超过扣除输出预留后的预算时，必须抛出异常拒绝请求，而不是返回部分结果或静默丢弃必需项。

实验 ID: model-multimodal-input-02-context-budget
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/model-multimodal-input/02-context-budget/
验收: 待验收

教师示范

目标: 演示 select_context 如何先预留输出空间，再纳入所有必需项，最后按优先级填充可选项。

假设我们有一个上下文窗口 100 tokens，输出预留 20 tokens。有三个条目：system（60 tokens，必需，优先级 1）、notes（30 tokens，可选，优先级 10）、history（10 tokens，可选，优先级 5）。

正确的 select_context 实现会先计算可用预算 budget = window - output_reserve = 80。然后收集所有必需项，这里只有 system，其 token 数 60 小于 80，所以可以纳入。

接着处理可选项，按优先级降序排列：notes（优先级 10）和 history（优先级 5）。当前已用 60，剩余 20，notes 需要 30，放不下；history 需要 10，可以放入。最终返回 ["system", "history"]。

实施步骤

计算可用预算：budget = window - output_reserve。
筛选出所有 required=True 的条目，检查它们的总 token 数是否超过预算，若超过则抛出 ValueError。
将必需项全部加入选择列表，并累计已用 token 数。
对可选项按优先级降序（优先级相同按 id 升序）排序，依次尝试放入，只要不超过剩余预算。
返回所有选中条目的 id 列表。

真实 solution 代码: forge/context_budget.py

from dataclasses import dataclass

@dataclass(frozen=True)
class ContextItem:
    id: str
    tokens: int
    required: bool
    priority: int


def select_context(items: list[ContextItem], window: int, output_reserve: int) -> list[str]:
    budget=window-output_reserve
    required=[item for item in items if item.required]
    if sum(item.tokens for item in required)>budget: raise ValueError("required context exceeds budget")
    chosen=list(required); used=sum(item.tokens for item in chosen)
    for item in sorted((x for x in items if not x.required),key=lambda x:(-x.priority,x.id)):
        if used+item.tokens<=budget: chosen.append(item); used+=item.tokens
    return [item.id for item in chosen]

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

输出预留从窗口大小中减去，确保模型有足够的 token 生成回答，避免输入内容占满整个窗口。

必需项先于可选项处理，保证系统指令等关键内容不会被可选材料挤占。

可选项按优先级排序，在剩余预算内尽可能多地纳入高优先级内容，实现上下文的有效利用。

带提示练习

修改 select_context 中的预算计算，使其扣除 output_reserve，从而保护必需项。

打开 forge/context_budget.py，找到 select_context 函数。
将 budget = window 改为 budget = window - output_reserve。
运行测试 python -m pytest -q tests/test_stage.py::test_output_reserve_and_required_context_are_protected，观察是否通过。
如果通过，再运行 python -m pytest -q tests/test_stage.py::test_impossible_required_context_is_rejected，确认必需项超限时抛出异常。

逐级提示

提示 1：输出预留是模型生成回答所需的空间，必须从总窗口中扣除。
提示 2：检查 budget 变量的赋值语句，它应该使用 window - output_reserve。
提示 3：修改后，可用预算变小，可选项可能无法放入，但必需项仍然会被保护。

完成标准

test_output_reserve_and_required_context_are_protected 测试通过。
test_impossible_required_context_is_rejected 测试通过。

故障注入与排查

在故障夹具 fixtures/failure.json 中，select_context 的实现将整个窗口视为可选的优先级容量，没有区分必需项和输出预留。运行测试 test_output_reserve_and_required_context_are_protected 时，函数返回 ['notes', 'system']，而期望是 ['system']。这是因为 notes 优先级更高，先被放入，导致必需的系统指令被挤占，且没有为输出预留空间。可执行故障夹具修改 forge/context_budget.py，并由 test_output_reserve_and_required_context_are_protected 稳定复现。

本阶段学习的上下文预算分配模型明确要求先扣除输出预留，再无条件纳入所有必需项，最后才按优先级填充可选项。故障代码缺少这些步骤，因此学习者可以利用刚建立的模型来诊断并修复。

已验证的故障合同

目标测试: test_output_reserve_and_required_context_are_protected
故障文件: forge/context_budget.py

诊断顺序

运行故障测试命令 python -m pytest -q ../tests/test_stage.py::test_output_reserve_and_required_context_are_protected，观察断言失败：返回 ['notes', 'system'] 而不是 ['system']。
检查 select_context 源码，发现 budget = window，没有扣除 output_reserve，且没有区分必需项和可选项，只是按优先级排序后依次放入。
根据上下文预算模型，应该先计算 budget = window - output_reserve，然后收集所有必需项并检查总 token 是否超限，最后再处理可选项。

修复原则: 修改 select_context 函数：将 budget = window 改为 budget = window - output_reserve；添加必需项收集和超限检查；先加入所有必需项，再按优先级填充可选项。修复后的代码应通过两个测试。

python -m pytest -q ../tests/test_stage.py::test_output_reserve_and_required_context_are_protected

迁移挑战

音乐创作场景中，你需要为一份发行计划分配上下文预算。系统提示是必需项，宣传素材是可选项。窗口大小为 200 tokens，输出预留 50 tokens。系统提示占 120 tokens，宣传素材 A 占 60 tokens（优先级 8），宣传素材 B 占 40 tokens（优先级 5）。

改变的条件: 将软件交付场景中的条目替换为音乐创作场景的条目，并调整窗口大小、输出预留和条目 token 数。

使用 ContextItem 和 select_context，为上述音乐创作场景编写代码，选择应纳入上下文的条目。确保系统提示被保护，输出空间不被挤占，并尽可能多地纳入高优先级宣传素材。

完成标准

调用 select_context 后，返回的 id 列表包含系统提示的 id，且不包含任何导致预算超限的可选素材。
返回的 id 列表长度符合预期：系统提示 + 可放入的宣传素材，且总 token 数不超过 window - output_reserve。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么必需项超限时要抛出异常而不是返回部分结果？请结合上下文预算的边界和系统约束解释。

一个有力回答应包含

必需项是系统指令或策略等不可丢弃的内容，如果超限，任何部分结果都会导致模型缺少关键约束，可能产生错误行为。
抛出异常可以立即通知调用方请求不可行，避免静默失败或产生不安全的输出。
返回部分结果会掩盖问题，使系统在后续步骤中可能违反规则，因此必须显式拒绝。

掌握证据

test_output_reserve_and_required_context_are_protected 测试通过，证明输出预留和必需项得到保护。
test_impossible_required_context_is_rejected 测试通过，证明必需项超限时正确抛出异常。
在 forge/context_budget.py 中实现了 select_context，包含预算扣除、必需项优先和可选项按优先级填充的逻辑。

STAGE 03

计算余弦相似度并拒绝维度不一致或零向量，避免把 Embedding 当答案

先看一个具体问题

为什么点积不能直接作为向量相似度的度量？

在上一阶段，你已经实现了 forge/tokens.py 和 forge/context_budget.py，现在需要新增 forge/embeddings.py 来计算向量相似度。

如果直接返回两个向量的点积，相似度会受向量长度影响：例如 [1,0] 与 [10,0] 方向相同，但点积为 10，而不是期望的 1。

测试 test_cosine_similarity_is_scale_invariant 要求 cosine_similarity([1,0],[10,0]) 返回 1.0，而 test_dimension_mismatch_is_rejected 要求维度不一致时抛出 ValueError。

你需要实现余弦相似度：先检查维度一致且向量非零，再计算点积并除以两个范数的乘积，从而消除长度影响。

先做判断

在实现 cosine_similarity 之前，请预测：如果函数只返回两个向量的点积，test_cosine_similarity_is_scale_invariant 会得到什么结果？

返回 1.0，测试通过
返回 10，测试失败
抛出 ValueError
返回 0

判断依据: 点积 [1,0]·[10,0] = 1*10 + 0*0 = 10，而测试期望 1.0，因此测试会失败。这说明点积受向量长度影响，不能直接作为相似度。

它是什么

余弦相似度是什么

余弦相似度通过归一化消除向量长度影响，只衡量方向一致性：它等于两个向量夹角的余弦值，范围在 -1 到 1 之间。

计算步骤：先检查两个向量维度相同且非零，然后计算点积，再除以两个向量范数的乘积。

对于方向相同的向量，无论长度如何，余弦相似度都是 1；对于正交向量，相似度为 0；对于方向相反的向量，相似度为 -1。

它不是什么

余弦相似度不是什么

余弦相似度不是点积：点积会受向量长度影响，而余弦相似度通过除以范数消除了这种影响。

余弦相似度也不是距离度量：它只关注方向，不关注向量之间的绝对距离。

余弦相似度不能处理零向量或维度不一致的向量，这些情况没有定义，必须拒绝。

它与相邻概念的关系

与其他概念的关系

余弦相似度与点积的关系：余弦相似度 = 点积 / (范数1 * 范数2)，因此点积是余弦相似度的分子部分。

余弦相似度与范数的关系：范数用于归一化，确保相似度不受向量缩放影响。

余弦相似度与 Embedding 的关系：Embedding 是向量表示，余弦相似度用于比较两个 Embedding 的方向相似性，但 Embedding 本身不是答案。

本阶段的边界决定: 当向量维度不一致或任一向量为零向量时，余弦相似度没有定义，必须抛出 ValueError；否则计算点积并除以范数乘积。

实验 ID: model-multimodal-input-03-embedding-similarity
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/model-multimodal-input/03-embedding-similarity/
验收: 待验收

教师示范

目标: 演示 cosine_similarity 如何计算两个向量的点积并除以范数，得到尺度不变的相似度。

假设我们要比较两个向量 [1,0] 和 [10,0]，它们方向相同但长度不同。

如果直接计算点积，得到 10，这不能反映方向一致性。

正确做法是计算余弦相似度：点积为 10，范数分别为 1 和 10，因此相似度为 10 / (1*10) = 1.0。

实施步骤

检查维度：两个向量长度都是 2，一致。
检查零向量：两个向量都不是零向量。
计算点积：110 + 00 = 10。
计算范数：left_norm = sqrt(1^2+0^2)=1，right_norm = sqrt(10^2+0^2)=10。
返回点积除以范数乘积：10 / (1*10) = 1.0。

真实 solution 代码: forge/embeddings.py

from math import sqrt

def cosine_similarity(left: list[float], right: list[float]) -> float:
    if len(left)!=len(right) or not left: raise ValueError("embedding dimensions must match")
    left_norm=sqrt(sum(x*x for x in left)); right_norm=sqrt(sum(x*x for x in right))
    if left_norm==0 or right_norm==0: raise ValueError("zero vector has no direction")
    return sum(a*b for a,b in zip(left,right))/(left_norm*right_norm)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

通过除以范数，相似度不再受向量长度影响，只反映方向一致性。

这个例子展示了为什么必须进行归一化：否则点积会误导相似度判断。

带提示练习

补全 cosine_similarity 中缺失的范数归一化步骤，使测试 test_cosine_similarity_is_scale_invariant 通过。

在 forge/embeddings.py 中创建 cosine_similarity 函数，参数为 left: list[float] 和 right: list[float]。
添加维度检查：如果 len(left) != len(right) 或任一向量为空，抛出 ValueError。
计算两个向量的范数：left_norm = sqrt(sum(x*x for x in left))，right_norm 同理。
检查零向量：如果任一范数为 0，抛出 ValueError。
返回点积除以范数乘积：sum(a*b for a,b in zip(left,right)) / (left_norm * right_norm)。

逐级提示

使用 math.sqrt 计算平方根。
点积可以用 sum(a*b for a,b in zip(left,right)) 计算。
范数计算：sqrt(sum(x*x for x in vector))。
确保先检查维度再计算范数，避免索引错误。

完成标准

test_cosine_similarity_is_scale_invariant 通过。
test_dimension_mismatch_is_rejected 通过。

故障注入与排查

当前 forge/embeddings.py 中的 cosine_similarity 实现只返回点积，没有进行范数归一化。运行测试 test_cosine_similarity_is_scale_invariant 时，实际得到 10 而不是期望的 1.0。你需要诊断并修复这个缺陷。

本阶段已经学习了余弦相似度的计算原理，知道必须除以范数来消除长度影响。通过检查测试失败输出，可以定位到函数缺少归一化步骤，并应用所学知识进行修复。

已验证的故障合同

目标测试: test_cosine_similarity_is_scale_invariant
故障文件: forge/embeddings.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_cosine_similarity_is_scale_invariant，观察失败输出。
看到断言错误：assert 10 == 1.0 ± 1.0e-06，说明函数返回了 10。
检查 forge/embeddings.py 中的 cosine_similarity 实现，发现只返回了点积 sum(a*b for a,b in zip(left,right))。
确认缺少范数归一化步骤，导致相似度受向量长度影响。

修复原则: 修改 cosine_similarity 函数：先检查维度一致且向量非空，然后计算范数并检查零向量，最后返回点积除以范数乘积。具体代码为：if len(left)!=len(right) or not left: raise ValueError("embedding dimensions must match")；left_norm=sqrt(sum(x*x for x in left)); right_norm=sqrt(sum(x*x for x in right))；if left_norm==0 or right_norm==0: raise ValueError("zero vector has no direction")；return sum(a*b for a,b in zip(left,right))/(left_norm*right_norm)。

python -m pytest -q ../tests/test_stage.py::test_cosine_similarity_is_scale_invariant

迁移挑战

在办公研究场景中，你需要比较两份研究摘要的 Embedding。假设有两份摘要的向量表示分别为 [2, 4, 6] 和 [1, 2, 3]，它们方向相同但长度不同。使用 cosine_similarity 计算它们的相似度，并验证当向量维度不一致时抛出异常。

改变的条件: 将测试数据从 [1,0] 和 [10,0] 改为 [2,4,6] 和 [1,2,3]，并增加一个维度不一致的测试用例。

编写一个临时测试脚本或使用 Python 交互环境，调用 cosine_similarity([2,4,6], [1,2,3]) 并打印结果；然后尝试调用 cosine_similarity([1,2], [1]) 并捕获异常。

完成标准

cosine_similarity([2,4,6], [1,2,3]) 返回 1.0（因为方向相同）。
cosine_similarity([1,2], [1]) 抛出 ValueError，且错误信息包含 “dimensions”。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么余弦相似度对向量缩放不敏感？请结合公式解释，并说明这种性质在比较 Embedding 时有什么好处。

一个有力回答应包含

余弦相似度公式为点积除以两个范数的乘积，即 cos(θ) = (A·B) / (||A|| ||B||)。
当向量 A 缩放 k 倍时，点积变为 k(A·B)，范数变为 k||A||，因此分子分母同时乘以 k，结果不变。
这种性质使得相似度只反映方向，不受 Embedding 长度影响，适合比较语义方向。

掌握证据

test_cosine_similarity_is_scale_invariant 通过：验证了余弦相似度对向量缩放不敏感。
test_dimension_mismatch_is_rejected 通过：验证了维度不一致时抛出 ValueError。
在故障实验室中成功诊断并修复了缺少归一化的问题，理解了余弦相似度的计算边界。

STAGE 04

规范化 UTF-8 文本、换行和大小限制，同时保留来源元数据

先看一个具体问题

为什么必须拒绝无效 UTF-8 而不是静默丢弃？

在办公研究场景中，你从多个来源收集文档：网页、表格、PDF 导出文件。这些文件可能包含损坏的字节序列，例如一个孤立的 0xFF 字节。如果解析器静默丢弃这些字节，文档内容就会在不知不觉中改变，导致后续引用或分析基于不完整的证据。

当前 forge/text_input.py 文件尚未创建，测试 test_invalid_utf8_is_rejected_not_silently_dropped 和 test_line_endings_are_normalized 会因导入错误而失败。你需要实现一个 parse_text 函数，它必须严格验证 UTF-8 编码，并在遇到无效字节时抛出 ValueError，而不是忽略错误继续处理。

同时，不同操作系统使用不同的换行符：Windows 使用 CRLF（\r\n），旧 Mac 使用 CR（\r），Unix 使用 LF（\n）。为了后续处理的一致性，parse_text 必须将所有换行符规范化为 LF。

先做判断

如果 parse_text 使用 errors='ignore' 解码一个包含无效字节的文档，会发生什么？

无效字节被丢弃，文档被接受，但内容可能不完整
抛出 UnicodeDecodeError，文档被拒绝
无效字节被替换为 U+FFFD，文档被接受
程序崩溃，无法继续执行

判断依据: 选择第一个选项：errors='ignore' 会静默丢弃无效字节，导致文档被接受但内容不完整。这正是测试 test_invalid_utf8_is_rejected_not_silently_dropped 要防止的行为。

它是什么

文本输入规范化是什么

文本输入规范化是将外部字节流转换为内部统一表示的过程，包括严格解码 UTF-8、统一换行符、检查大小限制，并保留来源元数据。

严格解码意味着任何无效字节都会导致显式错误，确保输入内容与原始字节完全一致，不丢失任何信息。

它不是什么

文本输入规范化不是什么

它不是一个宽松的清理过程，不能静默丢弃或替换无效字节，因为这会改变证据的完整性。

它也不是简单的字符串替换，而是需要明确区分编码错误和换行符差异，并对编码错误采取拒绝策略。

它与相邻概念的关系

与其他组件的关系

parse_text 返回的 TextDocument 对象包含 source 和 text 字段，后续的 token 计数、上下文预算和嵌入生成都依赖这个规范化后的文本。

如果文本在入口处被静默修改，所有下游处理都会基于错误的数据，因此严格验证是保证整个管道可靠性的基础。

本阶段的边界决定: 当遇到无效 UTF-8 字节时，必须抛出 ValueError 并拒绝输入，而不是尝试修复或忽略，因为任何字节丢失都可能改变文档含义。

实验 ID: model-multimodal-input-04-text-intake
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/model-multimodal-input/04-text-intake/
验收: 待验收

教师示范

目标: 演示 parse_text 如何严格解码 UTF-8 并规范化换行符，同时保留来源信息。

假设你收到一个来自 Windows 系统的文本文件，内容为 a\r\nb\rc，其中包含 CRLF 和单独的 CR 换行符。

调用 parse_text("a.txt", b"a\r\nb\rc") 应该返回一个 TextDocument，其 source 为 "a.txt"，text 为 "a\nb\nc"。

如果传入无效字节 b"\xff"，函数应该抛出 ValueError，并且错误消息中包含 "UTF-8"。

实施步骤

创建 forge/text_input.py 文件。
定义 TextDocument 数据类，包含 source 和 text 字段。
实现 parse_text 函数：检查大小限制，使用严格模式解码 UTF-8，捕获 UnicodeDecodeError 并转换为 ValueError，最后规范化换行符。
运行测试验证行为。

真实 solution 代码: forge/text_input.py

from dataclasses import dataclass

@dataclass(frozen=True)
class TextDocument:
    source: str
    text: str


def parse_text(source: str, payload: bytes, max_bytes: int=1_000_000) -> TextDocument:
    if len(payload)>max_bytes: raise ValueError("text input exceeds size limit")
    try: text=payload.decode("utf-8")
    except UnicodeDecodeError as error: raise ValueError("text input must be UTF-8") from error
    return TextDocument(source,text.replace("\r\n","\n").replace("\r","\n"))

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

严格解码使用 payload.decode("utf-8")，默认 errors='strict'，遇到无效字节会抛出 UnicodeDecodeError。

捕获该异常并重新抛出 ValueError，使调用者能够明确知道输入无效，而不是得到不完整的数据。

换行符规范化通过两次 replace 完成：先将 \r\n 替换为 \n，再将剩余的 \r 替换为 \n，确保所有换行符统一。

带提示练习

修改 parse_text 中的解码调用，将 errors='ignore' 改为严格模式，使无效字节抛出 ValueError。

打开 forge/text_input.py 文件。
找到 payload.decode("utf-8", errors="ignore") 这一行。
将其改为 payload.decode("utf-8")，即使用严格模式。
添加 try-except 块捕获 UnicodeDecodeError 并抛出 ValueError。
运行测试确认 test_invalid_utf8_is_rejected_not_silently_dropped 通过。

逐级提示

严格模式是 Python 的默认行为，不需要显式指定 errors 参数。
捕获 UnicodeDecodeError 后，使用 raise ValueError(...) from error 保留原始异常链。
错误消息中应包含 “UTF-8” 以满足测试的 match 参数。

完成标准

test_invalid_utf8_is_rejected_not_silently_dropped 测试通过。
test_line_endings_are_normalized 测试通过。

故障注入与排查

在故障实验室中，你面对一个已经实现但存在缺陷的 forge/text_input.py 文件。该文件使用 errors='ignore' 解码 UTF-8，导致无效字节被静默丢弃。运行测试 test_invalid_utf8_is_rejected_not_silently_dropped 会失败，因为函数没有抛出 ValueError。你需要诊断并修复这个缺陷。

本阶段的核心就是严格 UTF-8 解码。你已经通过工作示例和引导练习掌握了如何正确实现 parse_text。现在你需要识别出 errors='ignore' 是问题根源，并将其替换为严格模式，同时添加异常转换。

已验证的故障合同

目标测试: test_invalid_utf8_is_rejected_not_silently_dropped
故障文件: forge/text_input.py

诊断顺序

运行测试 python -m pytest -q ../tests/test_stage.py::test_invalid_utf8_is_rejected_not_silently_dropped，观察失败信息：Failed: DID NOT RAISE ValueError。
检查 forge/text_input.py 中的解码调用，发现使用了 errors='ignore'。
理解 errors='ignore' 会静默丢弃无效字节，导致文档被接受但内容不完整。
将解码改为严格模式，并添加 try-except 块将 UnicodeDecodeError 转换为 ValueError。

修复原则: 将 payload.decode("utf-8", errors="ignore") 替换为 payload.decode("utf-8")，并添加异常处理：try: text = payload.decode("utf-8") except UnicodeDecodeError as error: raise ValueError("text input must be UTF-8") from error。

python -m pytest -q ../tests/test_stage.py::test_invalid_utf8_is_rejected_not_silently_dropped

迁移挑战

在办公研究场景中，你收到一份从 Windows 系统导出的文档，其中包含 CRLF 换行符。你需要使用 parse_text 解析该文档，并验证无效 UTF-8 字节会被拒绝。

改变的条件: 将输入文档的换行符从 LF 改为 CRLF，并加入一个无效字节序列以测试拒绝行为。

编写一个测试或脚本，调用 parse_text 解析一个包含 CRLF 换行符的字节串，例如 b"line1\r\nline2\r\n"，并断言返回的 text 字段中换行符已规范化为 \n。同时，调用 parse_text 解析一个包含无效字节的字节串，例如 b"\xff"，并断言抛出 ValueError。

完成标准

解析 CRLF 文档后，text 字段中的换行符全部为 \n，且内容顺序不变。
解析无效 UTF-8 字节时抛出 ValueError，且错误消息包含 “UTF-8”。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么在文本输入规范化中使用 errors='ignore' 是危险的？请结合证据完整性解释。

一个有力回答应包含

errors='ignore' 会静默丢弃无效字节，导致文档内容不完整，破坏证据完整性。
后续处理（如 token 计数、嵌入生成）会基于错误的数据，可能产生误导性结果。
严格模式通过抛出异常强制调用者处理无效输入，确保数据质量。

掌握证据

test_invalid_utf8_is_rejected_not_silently_dropped 测试通过，证明无效 UTF-8 被拒绝。
test_line_endings_are_normalized 测试通过，证明换行符被统一为 LF。
forge/text_input.py 中的 parse_text 函数实现了严格解码、大小限制和换行符规范化。

STAGE 05

依据真实文件签名和像素上限验证 PNG/JPEG，而不是只相信扩展名

先看一个具体问题

为什么不能只相信文件扩展名？

你正在为多模态输入模块添加图片支持。测试 test_extension_cannot_override_file_signature 要求：即使文件名是 fake.png，只要内容不是真正的 PNG 签名，validate_image 就必须抛出包含 signature 的 ValueError。

当前 forge/image_input.py 尚未创建，运行 python -m pytest -q tests/test_stage.py 会因 ModuleNotFoundError: No module named 'forge.image_input' 而收集失败。

你需要实现 validate_image，它接收文件名、原始字节、宽、高和可选的最大像素数，返回 ImageAsset。关键决策是：媒体类型由文件内容签名决定，而不是由扩展名决定。

先做判断

如果 validate_image 只根据文件名后缀判断媒体类型，当传入 fake.png 和普通文本字节时会发生什么？

函数会返回 ImageAsset('image/png', ...)，测试失败。
函数会抛出 ValueError，测试通过。
函数会返回 ImageAsset('image/jpeg', ...)。
函数会崩溃，因为无法解析文本。

判断依据: 正确答案是第一个选项。只根据扩展名判断会信任用户可控的元数据，导致伪造的 .png 文件被当作图片接受，测试 test_extension_cannot_override_file_signature 会失败。

它是什么

文件签名验证是什么

文件签名是文件开头的固定字节序列，例如 PNG 的 b"\x89PNG\r\n\x1a\n" 和 JPEG 的 b"\xff\xd8\xff"。

validate_image 通过 payload.startswith(PNG) 或 payload.startswith(JPEG) 检查这些魔数，从而可靠地识别格式。

它不是什么

文件签名验证不是什么

它不是基于文件名后缀的猜测。扩展名是用户可控的元数据，不能作为格式判断的依据。

它也不是完整的图片解码或内容验证；它只检查文件头部的魔数，不保证图片数据完整或可渲染。

它与相邻概念的关系

签名、像素上限与 ImageAsset 的关系

签名检查确定媒体类型，像素上限检查防止资源滥用，两者共同保证 ImageAsset 的合法性。

ImageAsset 是不可变数据类，包含 media_type、width 和 height，只有通过验证才能创建。

本阶段的边界决定: 当文件签名不匹配任何已知格式时，必须抛出 ValueError；当宽、高非正或乘积超过 max_pixels 时，也必须抛出 ValueError。

实验 ID: model-multimodal-input-05-image-intake
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/model-multimodal-input/05-image-intake/
验收: 待验收

教师示范

目标: 演示 validate_image 如何检查 PNG 魔数并验证像素尺寸，返回 ImageAsset。

假设你收到一个文件名为 photo.png 的图片，内容以 PNG 魔数开头。

调用 validate_image("photo.png", PNG + b"data", 100, 50) 时，函数首先检查 payload.startswith(PNG) 为真，因此媒体类型设为 "image/png"。

然后检查宽 100、高 50 均为正数，且乘积 5000 小于默认的 max_pixels=20_000_000，所以返回 ImageAsset("image/png", 100, 50)。

实施步骤

定义常量 PNG 和 JPEG。
在 validate_image 中，使用 payload.startswith(PNG) 判断 PNG，否则使用 payload.startswith(JPEG) 判断 JPEG。
如果两者都不匹配，将 media 设为空字符串，并抛出 ValueError("unsupported image signature")。
检查 width < 1、height < 1 或 width * height > max_pixels，若任一成立则抛出 ValueError("invalid image dimensions")。
返回 ImageAsset(media, width, height)。

真实 solution 代码: forge/image_input.py

from dataclasses import dataclass

PNG = b"\x89PNG\r\n\x1a\n"
JPEG = b"\xff\xd8\xff"

@dataclass(frozen=True)
class ImageAsset:
    media_type: str
    width: int
    height: int


def validate_image(name: str, payload: bytes, width: int, height: int, max_pixels: int=20_000_000) -> ImageAsset:
    media="image/png" if payload.startswith(PNG) else "image/jpeg" if payload.startswith(JPEG) else ""
    if not media: raise ValueError("unsupported image signature")
    if width<1 or height<1 or width*height>max_pixels: raise ValueError("invalid image dimensions")
    return ImageAsset(media,width,height)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

魔数检查是格式识别的可靠方法，因为文件内容由数据本身决定，不受文件名影响。

像素上限防止超大图片耗尽内存或计算资源，这是安全边界的一部分。

带提示练习

补全 validate_image 中缺失的签名检查，使伪造的 .png 文件抛出 ValueError。

创建 forge/image_input.py 文件。
定义 PNG 和 JPEG 常量。
实现 ImageAsset 数据类。
在 validate_image 中，先用 payload.startswith(PNG) 判断 PNG，再用 payload.startswith(JPEG) 判断 JPEG。
如果两者都不匹配，抛出 ValueError("unsupported image signature")。
添加像素尺寸检查：宽、高必须为正，且乘积不超过 max_pixels。
运行测试验证。

逐级提示

提示 1：使用 payload.startswith(PNG) 检查 PNG 签名。
提示 2：如果签名不匹配，设置 media = "" 并抛出异常。
提示 3：像素检查条件为 width < 1 or height < 1 or width * height > max_pixels。

完成标准

test_extension_cannot_override_file_signature 通过。
test_valid_png_and_pixel_limit 通过。

故障注入与排查

当前 forge/image_input.py 存在缺陷：validate_image 仅根据文件名后缀判断媒体类型，完全忽略文件内容。运行 python -m pytest -q ../tests/test_stage.py::test_extension_cannot_override_file_signature 会失败，因为 fake.png 中的普通文本被当作 PNG 接受。

本阶段已经学习了文件签名验证和像素上限检查。通过检查 payload.startswith(PNG) 和 payload.startswith(JPEG)，可以修复这个缺陷。测试 test_extension_cannot_override_file_signature 明确要求伪造的 .png 文件必须抛出 ValueError。

已验证的故障合同

目标测试: test_extension_cannot_override_file_signature
故障文件: forge/image_input.py

诊断顺序

运行 python -m pytest -q ../tests/test_stage.py::test_extension_cannot_override_file_signature，观察测试失败，错误为 Failed: DID NOT RAISE ValueError。
检查 forge/image_input.py 中的 validate_image，发现媒体类型由 name.lower().endswith(".png") 决定，而不是检查 payload。
确认 fake.png 的内容不是 PNG 签名，但函数仍返回 ImageAsset("image/png", ...)。
修复方法：使用 payload.startswith(PNG) 和 payload.startswith(JPEG) 判断签名，不匹配时抛出 ValueError。

修复原则: 修改 validate_image，将媒体类型判断改为基于文件签名：media = "image/png" if payload.startswith(PNG) else "image/jpeg" if payload.startswith(JPEG) else ""，然后检查 if not media: raise ValueError("unsupported image signature")。同时添加像素尺寸检查：if width < 1 or height < 1 or width * height > max_pixels: raise ValueError("invalid image dimensions")。

python -m pytest -q ../tests/test_stage.py::test_extension_cannot_override_file_signature

迁移挑战

在音乐创作场景中，你需要验证一张宣传图片。该图片文件名为 promo.png，内容以 PNG 魔数开头，宽 800、高 600。

改变的条件: 将场景从软件交付改为音乐创作，使用 PNG 常量和 validate_image 验证宣传图片，并测试像素上限。

编写一个测试或脚本，使用 PNG 常量和 validate_image 验证 promo.png 的媒体类型为 image/png，并测试当宽高乘积超过 max_pixels 时抛出 ValueError。

完成标准

验证 validate_image("promo.png", PNG + b"data", 800, 600).media_type == "image/png"。
验证 validate_image("promo.png", PNG, 10000, 10000) 抛出包含 dimensions 的 ValueError。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么魔数检查比扩展名更可靠？像素上限的作用是什么？

一个有力回答应包含

魔数检查基于文件内容，不受用户可控的文件名影响，因此能可靠识别格式。
像素上限防止超大图片导致内存耗尽或计算资源滥用，是安全边界。

掌握证据

test_extension_cannot_override_file_signature 通过，证明伪造的 .png 文件被拒绝。
test_valid_png_and_pixel_limit 通过，证明有效的 PNG 被接受且像素上限生效。
forge/image_input.py 实现了基于文件签名的验证和像素尺寸检查。

STAGE 06

把 JSON、CSV 和 Markdown 解析成带来源定位的块，并对格式错误给出明确失败

先看一个具体问题

为什么 CSV 解析必须保留行号定位信息？

在办公研究场景中，你需要把一份 CSV 表格交给模型处理，但模型无法直接读取原始文件，必须先把 CSV 转换成带来源定位的文本块。

如果解析时只提取文本内容而丢弃行号，后续引用或调试时就无法追溯到原始表格的具体行，证据链会断裂。

当前代码中 forge/documents.py 尚未创建，测试 test_csv_rows_keep_source_locators 会因导入失败而报错，这是本阶段要解决的起点。

先做判断

在实现 parse_document 时，对于 text/csv 类型的输入，以下哪种做法最符合测试要求？

把整个 CSV 文本作为一个块返回，locator 设为 “document”
用 csv.DictReader 逐行解析，为每一行生成 row:i 定位符
把 CSV 当作 JSON 解析，失败时抛出异常
只返回第一行数据，忽略其余行

判断依据: 测试期望返回两个块，locator 分别为 row:1 和 row:2，因此必须逐行解析并生成行号定位符。

它是什么

来源定位是证据链的一部分

来源定位（locator）记录了每个文本块在原始文档中的精确位置，例如 CSV 的行号或 JSON 的根路径 $。

它允许后续处理阶段（如检索或生成）引用具体来源，从而支持可验证的引用和调试。

不同文档格式需要不同的定位策略：CSV 按行定位，JSON 按路径定位，Markdown 通常按整个文档定位。

它不是什么

来源定位不是可有可无的元数据

来源定位不是简单的文本标签，它必须能够唯一标识块在原始文档中的位置。

它不是对整个文档的统一标识，例如所有 CSV 行不能都使用 "document" 作为 locator。

它也不是内容的一部分，而是独立于文本内容的元数据，用于追溯和引用。

它与相邻概念的关系

解析器、块与定位符的关系

parse_document 根据媒体类型分派到不同的解析逻辑，每种解析逻辑负责生成相应的块和定位符。

每个 Chunk 对象包含 source（来源文件名）、locator（定位符）和 text（文本内容），三者共同构成可追溯的单元。

测试通过检查 locator 列表来验证解析是否正确，因此定位符的生成规则必须与测试期望一致。

本阶段的边界决定: 当媒体类型为 text/csv 时，必须使用 csv.DictReader 逐行解析，并为每一行生成 row:i 定位符；当媒体类型未知时，必须抛出 ValueError 异常。

实验 ID: model-multimodal-input-06-document-parsing
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/model-multimodal-input/06-document-parsing/
验收: 待验收

教师示范

目标: 演示 parse_document 如何根据媒体类型分派到 JSON、CSV 或 Markdown 解析器，并为 CSV 行生成 locator。

假设你收到一个 CSV 文件 scores.csv，内容为 name,score\nAda,9\nLin,8\n，媒体类型为 text/csv。

你调用 parse_document("scores.csv", "text/csv", csv_text)，期望得到两个块，分别对应 Ada 和 Lin 的行。

在实现中，你首先检查媒体类型，如果是 text/csv，则使用 csv.DictReader 读取每一行，并为每一行生成一个 Chunk，其 locator 为 row:1、row:2 等。

实施步骤

创建 forge/documents.py 文件，定义 Chunk 数据类和 parse_document 函数。
在 parse_document 中，首先处理 application/json：使用 json.loads 解析并返回一个 locator 为 $ 的块。
然后处理 text/csv：使用 csv.DictReader 读取所有行，如果行列表为空则抛出 ValueError，否则为每一行生成一个 Chunk，locator 为 row:{i}（i 从 1 开始）。
处理 text/markdown：返回一个 locator 为 document 的块，文本为去除首尾空白后的内容。
对于其他媒体类型，抛出 ValueError("unsupported document type")。

真实 solution 代码: forge/documents.py

import csv, io, json
from dataclasses import dataclass

@dataclass(frozen=True)
class Chunk:
    source: str
    locator: str
    text: str


def parse_document(source: str, media_type: str, text: str) -> list[Chunk]:
    if media_type=="application/json":
        data=json.loads(text); return [Chunk(source,"$",json.dumps(data,ensure_ascii=False,sort_keys=True))]
    if media_type=="text/csv":
        rows=list(csv.DictReader(io.StringIO(text)))
        if not rows: raise ValueError("CSV requires a header and row")
        return [Chunk(source,f"row:{i}",json.dumps(row,ensure_ascii=False,sort_keys=True)) for i,row in enumerate(rows,1)]
    if media_type=="text/markdown":
        return [Chunk(source,"document",text.strip())]
    raise ValueError("unsupported document type")

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

CSV 解析使用 csv.DictReader 将每一行转换为字典，然后通过 json.dumps 序列化为字符串，保证文本内容可读且稳定。

locator 使用 row:{i} 格式，其中 i 从 1 开始，与测试期望的 row:1、row:2 一致。

未知媒体类型抛出异常，确保调用者必须显式处理不支持的格式，避免静默错误。

带提示练习

补全 parse_document 中缺失的 CSV 分支，使用 DictReader 并生成 row:i 定位符。

打开 forge/documents.py，找到 parse_document 函数。
在函数中添加对 text/csv 的处理分支。
使用 csv.DictReader 读取输入文本，并将结果转换为列表。
如果列表为空，抛出 ValueError 异常。
遍历列表，为每一行生成一个 Chunk，locator 为 row:{i}，其中 i 从 1 开始。
运行测试 python -m pytest -q tests/test_stage.py 验证。

逐级提示

记得导入 csv 和 io 模块。
csv.DictReader 需要一个文件对象，可以使用 io.StringIO(text) 创建。
locator 的格式是 row: 后跟行号，行号从 1 开始。
如果 CSV 没有数据行，应该抛出异常。

完成标准

test_csv_rows_keep_source_locators 测试通过。
test_unknown_document_type_is_rejected 测试通过。

故障注入与排查

当前 forge/documents.py 的实现将所有非 JSON 文档都当作 Markdown 处理，导致 CSV 被解析为一个整体块，locator 为 document。运行测试 test_csv_rows_keep_source_locators 时，断言失败，因为期望的 locator 列表是 ['row:1', 'row:2']，而实际得到的是 ['document']。

本阶段已经学习了如何根据媒体类型分派解析逻辑，并且知道 CSV 需要逐行解析并生成行号定位符。通过检查 parse_document 中的分支，可以发现缺少对 text/csv 的处理，因此可以修复该缺陷。

已验证的故障合同

目标测试: test_csv_rows_keep_source_locators
故障文件: forge/documents.py

诊断顺序

运行测试 python -m pytest -q ../tests/test_stage.py::test_csv_rows_keep_source_locators，观察失败输出。
查看 forge/documents.py 中的 parse_document 函数，发现只有 application/json 分支，其他类型都返回一个 locator 为 document 的块。
对比测试期望，测试要求 CSV 返回两个块，locator 分别为 row:1 和 row:2。
因此，需要添加 text/csv 分支，使用 csv.DictReader 逐行解析并生成相应 locator。

修复原则: 在 parse_document 中添加 text/csv 分支：使用 csv.DictReader 读取所有行，如果行列表为空则抛出 ValueError，否则为每一行生成一个 Chunk，locator 为 row:{i}（i 从 1 开始）。同时，对于未知媒体类型，抛出 ValueError("unsupported document type")。

python -m pytest -q ../tests/test_stage.py::test_csv_rows_keep_source_locators

迁移挑战

在办公研究场景中，你收到一份新的 CSV 表格 employees.csv，内容为 name,department\nAlice,Engineering\nBob,Marketing\n。你需要使用 parse_document 解析它，并验证未知类型被拒绝。

改变的条件: 将 CSV 数据内容从学生成绩改为员工部门信息，但保持两行数据和相同的列结构。

编写一个 Python 脚本或交互式命令，调用 parse_document 解析上述 CSV 内容，并打印每个块的 locator 和 text。同时，尝试解析一个未知类型（如 application/pdf），确认抛出 ValueError 异常。

完成标准

解析 employees.csv 后，返回两个块，locator 分别为 row:1 和 row:2。
解析未知类型 application/pdf 时，抛出 ValueError 异常，异常消息包含 unsupported。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么 CSV 需要逐行分块而不是整体作为一个块？请从来源定位和后续处理的角度解释。

一个有力回答应包含

逐行分块允许每个数据行独立定位和引用，便于追溯证据。
整体作为一个块会丢失行级信息，无法满足需要精确引用的场景。
逐行分块也便于后续按行进行检索或分析。

掌握证据

test_csv_rows_keep_source_locators 测试通过，证明 CSV 解析正确生成行号定位符。
test_unknown_document_type_is_rejected 测试通过，证明未知类型会抛出异常。
能够解释 CSV 逐行分块的原因，并理解来源定位在证据链中的作用。

STAGE 07

根据模型能力和媒体类型路由输入，并对不支持组合返回可解释拒绝

先看一个具体问题

为什么纯文本模型必须拒绝图片输入？

在上一阶段，你已经实现了文档解析器，能够从图片中提取文本信息。现在需要为系统增加一个统一的输入入口 route_input，它根据媒体类型和模型能力决定是否接受输入。

当前测试 test_text_only_model_rejects_image_with_reason 要求：当模型只支持文本模态时，传入 image/png 必须返回 unsupported 状态，并且 reason 中包含 image。

但一个常见的错误是只根据媒体类型路由，认为解析器能处理图片就代表模型也能接受图片。这会导致纯文本模型收到图片输入，产生静默错误。

你需要实现 forge/intake.py，定义 IntakeDecision 数据类和 route_input 函数，同时检查解析器能力和模型模态，对不支持的组合返回可解释的拒绝。

先做判断

在实现 route_input 时，如果只检查媒体类型而忽略 model_modalities，会发生什么？

图片会被路由到 image_adapter，即使模型只支持文本，导致测试失败。
图片会被正确拒绝，因为解析器不支持图片。
文本输入也会被拒绝，因为模型模态检查会误判。
系统会崩溃，因为无法确定适配器。

判断依据: 正确答案是第一个选项。只检查媒体类型会错误地将图片路由到 image_adapter，而测试期望纯文本模型拒绝图片。必须同时检查模型能力。

它是什么

输入路由是双重检查

输入路由是一个决策函数，它接收媒体类型和模型支持的模态集合，返回一个 IntakeDecision 对象，包含状态、适配器和拒绝原因。

路由必须同时验证两件事：解析器是否能处理该媒体类型，以及模型是否具备该模态的输入能力。只有两者都满足时才返回 ready。

它不是什么

不是简单的媒体类型映射

输入路由不是仅仅根据媒体类型字符串映射到适配器名称。如果忽略模型能力，就会把不支持的输入交给模型，造成静默错误。

它也不是解析器能力的唯一检查。即使解析器能提取图片中的文本，纯文本模型也无法直接处理图片的原始像素数据。

它与相邻概念的关系

解析器、模型与适配器的关系

解析器负责将原始输入转换为模型可理解的格式，例如从图片中提取文本。模型能力定义了模型可以接收的模态类型，例如 text 或 image。

适配器是连接输入和模型的具体实现，例如 text_adapter 处理文本输入。路由函数根据媒体类型确定模态，再检查该模态是否在模型能力集合中，最后选择对应的适配器。

本阶段的边界决定: 当媒体类型无法识别或模型不支持该模态时，必须返回 unsupported 状态并附带明确的 reason，而不是返回 ready 或抛出异常。

实验 ID: model-multimodal-input-07-unified-intake-boundary
输入: 软件交付：为前端仓库增加功能、修改代码并补测试。
证据目录: evidence/model-multimodal-input/07-unified-intake-boundary/
验收: 待验收

教师示范

目标: 演示 route_input 如何根据媒体类型确定模态，并检查模型能力，返回 IntakeDecision。

假设我们有一个纯文本模型，其能力集合为 {"text"}。现在传入媒体类型 "text/plain"。

首先，route_input 根据媒体类型确定模态为 "text"。然后检查 "text" 是否在模型能力集合中，发现存在，因此返回 IntakeDecision("ready", "text_adapter", None)。

这个例子展示了正常路由的流程：媒体类型识别、模态检查、适配器选择。

实施步骤

定义 IntakeDecision 数据类，包含 status、adapter、reason 字段。
实现 route_input 函数，根据媒体类型确定模态：image/* 对应 image，text/*、application/json、text/csv 对应 text，其他为 unknown。
如果模态为 unknown，返回 unsupported 状态，reason 为 "no parser for media type"。
如果模态不在 model_modalities 中，返回 unsupported 状态，reason 为 f"model lacks {modality} input"。
否则返回 ready 状态，适配器为 f"{modality}_adapter"。

真实 solution 代码: forge/intake.py

from dataclasses import dataclass

@dataclass(frozen=True)
class IntakeDecision:
    status: str
    adapter: str | None
    reason: str | None


def route_input(media_type: str, model_modalities: set[str]) -> IntakeDecision:
    modality="image" if media_type.startswith("image/") else "text" if media_type.startswith("text/") or media_type in {"application/json","text/csv"} else "unknown"
    if modality=="unknown": return IntakeDecision("unsupported",None,"no parser for media type")
    if modality not in model_modalities: return IntakeDecision("unsupported",None,f"model lacks {modality} input")
    return IntakeDecision("ready",f"{modality}_adapter",None)

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

为什么这样实现

在这个例子中，媒体类型 text/plain 被识别为文本模态，模型能力集合包含 text，因此路由成功。

注意 route_input 返回的是不可变的数据类实例，这有助于避免意外修改决策结果。

带提示练习

修改 route_input 中的条件，增加对 model_modalities 的检查，使纯文本模型拒绝图片。

打开 forge/intake.py，找到 route_input 函数。
在确定模态后，添加检查：如果模态不在 model_modalities 中，返回 unsupported 状态和相应原因。
运行测试 test_text_only_model_rejects_image_with_reason，确认图片被拒绝。

逐级提示

检查模态是否在 model_modalities 集合中，使用 if modality not in model_modalities:。
拒绝时 reason 应包含 image 字样，例如 f"model lacks {modality} input"。
确保返回的 IntakeDecision 状态为 "unsupported"，适配器为 None。

完成标准

test_text_only_model_rejects_image_with_reason 测试通过。
test_supported_text_routes_to_text_adapter 测试仍然通过。

故障注入与排查

在故障夹具 fixtures/failure.json 中，forge/intake.py 存在缺陷：它只根据媒体类型路由，没有检查 model_modalities。运行测试 test_text_only_model_rejects_image_with_reason 时，传入 image/png 和模型能力 {"text"}，但函数返回了 ready 状态和 image_adapter，导致断言失败。

本阶段的核心是双重检查：解析器能力和模型模态。故障代码缺少模型模态检查，这正是我们要修复的机制。通过诊断这个故障，你能理解为什么只检查解析器是不够的，并学会如何添加正确的检查。

已验证的故障合同

目标测试: test_text_only_model_rejects_image_with_reason
故障文件: forge/intake.py

诊断顺序

运行故障命令 python -m pytest -q ../tests/test_stage.py::test_text_only_model_rejects_image_with_reason，观察测试失败。
查看 forge/intake.py 中的 route_input 函数，发现它只根据媒体类型确定模态，然后直接返回 ready 状态，没有检查 model_modalities。
对比测试期望：当模型能力为 {"text"} 时，图片输入应返回 unsupported 状态。
定位缺陷：缺少对 model_modalities 的检查，导致图片被错误路由到 image_adapter。

修复原则: 在 route_input 函数中，确定模态后添加检查：如果模态不在 model_modalities 中，返回 IntakeDecision("unsupported", None, f"model lacks {modality} input")。这样纯文本模型就会拒绝图片输入。

python -m pytest -q ../tests/test_stage.py::test_text_only_model_rejects_image_with_reason

迁移挑战

在软件交付场景中，你需要使用 route_input 路由一个 JSON 文档到文本适配器，并验证图片被纯文本模型拒绝。

改变的条件: 将输入媒体类型从 text/plain 改为 application/json，同时保持模型能力为 {"text"}。

在 Python 交互环境或临时脚本中，调用 route_input("application/json", {"text"})，确认返回 ready 状态和 text_adapter；再调用 route_input("image/png", {"text"})，确认返回 unsupported 状态且 reason 包含 image。无需编写新的测试文件。

完成标准

在交互式验证中，route_input("application/json", {"text"}) 返回 ready 状态和 text_adapter。
route_input("image/png", {"text"}) 返回 unsupported 状态，且 reason 包含 image。

运行真实测试

python -m pytest -q tests/test_stage.py

运行真实测试

..                                                                       [100%]
2 passed in <duration>s

设计解释题: 为什么需要同时检查解析器能力和模型模态？如果只检查其中一个，会导致什么问题？

一个有力回答应包含

解析器能力决定了系统能否将输入转换为模型可理解的格式，而模型模态决定了模型能否直接处理该格式。
如果只检查解析器能力，纯文本模型可能会收到图片输入，导致模型无法正确处理，产生静默错误。
如果只检查模型模态，可能会拒绝解析器能够处理但模型不直接支持的输入，例如从图片中提取的文本，但这种情况需要额外的转换步骤。
双重检查确保了输入在进入模型之前既被正确解析，又被模型接受，避免了不必要的错误。

掌握证据

test_text_only_model_rejects_image_with_reason 通过，证明纯文本模型拒绝图片输入并返回原因。
test_supported_text_routes_to_text_adapter 通过，证明文本输入正确路由到文本适配器。
在故障实验室中，成功诊断并修复了缺少模型模态检查的缺陷，使测试通过。

完成本章

让三个场景的文本、图片和结构化文件进入同一输入合同。

FORGE / LOCAL CHECK

本地实验自检

未开始
2阅读中
3实验已下载
4测试结果已读取
5本地自检通过

verification.json 只在当前浏览器中解析，不会上传。这里验证的是实验合同，不是服务器认证或第三方背书。

选择 verification.json

主要层

依赖能力

为后续提供

受这些层约束

概念校准

上下文

模型运行原理与多模态输入

先看一个具体问题

为什么字符数不能替代 Token 数？

先做判断

它是什么

Token 估算是什么

它不是什么

Token 估算不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

有限窗口下，可选材料挤占必需指令

先做判断

它是什么

上下文预算分配模型

它不是什么

不是简单的优先级排序

它与相邻概念的关系

与 token 估算和上下文概念的关系

先看一个具体问题

为什么点积不能直接作为向量相似度的度量？

先做判断

它是什么

余弦相似度是什么

它不是什么

余弦相似度不是什么

它与相邻概念的关系

与其他概念的关系

先看一个具体问题

为什么必须拒绝无效 UTF-8 而不是静默丢弃？

先做判断

它是什么

文本输入规范化是什么

它不是什么

文本输入规范化不是什么

它与相邻概念的关系

与其他组件的关系

先看一个具体问题

为什么不能只相信文件扩展名？

先做判断

它是什么

文件签名验证是什么

它不是什么

文件签名验证不是什么

它与相邻概念的关系

签名、像素上限与 ImageAsset 的关系

先看一个具体问题

为什么 CSV 解析必须保留行号定位信息？

先做判断

它是什么

来源定位是证据链的一部分

它不是什么

来源定位不是可有可无的元数据

它与相邻概念的关系

解析器、块与定位符的关系

先看一个具体问题

为什么纯文本模型必须拒绝图片输入？

先做判断

它是什么

输入路由是双重检查

它不是什么

不是简单的媒体类型映射

它与相邻概念的关系

解析器、模型与适配器的关系